{ "cells": [ { "cell_type": "markdown", "id": "e4c71b14", "metadata": {}, "source": [ "# Generate ticks from candles\n", "\n", "Test candles data\n", "\n", "```\n", "381 1752267300000 1.16916 1.16921 1.16902 1.16902 106.0\n", "382 1752267360000 1.16900 1.16905 1.16850 1.16850 80.0\n", "383 1752267420000 1.16851 1.16883 1.16833 1.16882 34.0\n", "384 1752267480000 1.16881 1.16897 1.16879 1.16883 33.0\n", "385 1752267540000 1.16882 1.16884 1.16845 1.16845 32.0\n", "```" ] }, { "cell_type": "code", "execution_count": null, "id": "ee266eab", "metadata": {}, "outputs": [], "source": [ "from datetime import datetime\n", "from typing import Optional\n", "\n", "import numpy as np\n", "from bokeh.io import output_notebook\n", "from bokeh.plotting import figure, show\n", "\n", "from studio_shared.tick_generator import generate_ticks_from_candle\n", "from tl_bt_adapter.types import TLCandleModel\n", "from tl_shared.utils import (\n", " bt_to_tl_resolution,\n", " resolution_in_seconds,\n", ")\n", "from tradelocker_wss.types import TLWSSLevel1Model" ] }, { "cell_type": "code", "execution_count": null, "id": "0833a216", "metadata": {}, "outputs": [], "source": [ "# first version used\n", "def original_interpolate_candle_to_ticks(\n", " candle: TLCandleModel,\n", " timeframe: int,\n", " compression: int,\n", " tick_interval: float,\n", " seed: Optional[int],\n", ") -> list[TLWSSLevel1Model]:\n", " if seed is not None:\n", " np.random.seed(seed)\n", "\n", " duration = resolution_in_seconds(bt_to_tl_resolution(timeframe, compression))\n", " number_of_ticks = int(duration / tick_interval)\n", " if number_of_ticks < 4:\n", " raise ValueError(\"At least 4 ticks are required to interpolate a candle.\")\n", "\n", " steps = np.random.normal(0, 1, size=number_of_ticks - 2)\n", " walk: list[float] = np.cumsum(steps)\n", " walk = walk / (walk[-1] if walk[-1] != 0 else 1e-9)\n", " interpolated = candle.o + walk * (candle.c - candle.o)\n", " min_price, max_price = np.min(interpolated), np.max(interpolated)\n", " scale = (candle.h - candle.l) / (max_price - min_price) if max_price != min_price else 1\n", " interpolated_prices = (interpolated - min_price) * scale + candle.l\n", " interpolated_prices[np.argmin(interpolated_prices)] = candle.l\n", " interpolated_prices[np.argmax(interpolated_prices)] = candle.h\n", " prices = np.empty(number_of_ticks)\n", " prices[0] = candle.o\n", " prices[-1] = candle.c\n", " prices[1:-1] = interpolated_prices\n", "\n", " base_ts = candle.t / 1000 # ms to seconds\n", " timestamps = (base_ts + np.arange(number_of_ticks) * tick_interval) * 1000\n", " # move first tick off the boundary so a tick does not start exactly at\n", " # candle start datetime\n", " # timestamps[0] += BOUNDARY_OFFSET_MS\n", "\n", " ticks: list[TLWSSLevel1Model] = []\n", " for ts, price in zip(timestamps, prices):\n", " # prepare data for TLWSSLevel1Model\n", " tick = {\n", " \"t\": int(ts),\n", " \"ap\": price,\n", " \"bp\": price,\n", " \"as\": 0,\n", " \"bs\": 0,\n", " \"tiId\": 278,\n", " \"rId\": 0,\n", " }\n", " ticks.append(TLWSSLevel1Model.from_raw(tick))\n", "\n", " return ticks\n", "\n", "\n", "# helper method for plotting\n", "def plot_results(values, timestamps):\n", " output_notebook()\n", "\n", " y_timedates = []\n", " for timestamp in timestamps:\n", " y_timedates.append(datetime.fromtimestamp(timestamp / 1000))\n", " p = figure(\n", " title=\"plot results\",\n", " x_axis_type=\"datetime\",\n", " x_axis_label=\"datetime\",\n", " y_axis_label=\"value\",\n", " height=300,\n", " )\n", " p.line(y_timedates, values, line_width=2)\n", "\n", " show(p)" ] }, { "cell_type": "code", "execution_count": null, "id": "0b54e1f1", "metadata": {}, "outputs": [], "source": [ "candle_test_1 = TLCandleModel(t=1752267360000, o=1.16900, h=1.16905, l=1.16850, c=1.16850, v=80)\n", "candle_test_2 = TLCandleModel(t=1752267360000, o=1.16851, h=1.16883, l=1.16833, c=1.16882, v=34)" ] }, { "cell_type": "code", "execution_count": null, "id": "f0b6667e", "metadata": {}, "outputs": [], "source": [ "timedates, values = generate_ticks_from_candle(\n", " duration_in_seconds=1,\n", " open_=candle_test_1.o,\n", " high_=candle_test_1.h,\n", " low_=candle_test_1.l,\n", " close_=candle_test_1.c,\n", " number_of_ticks=240,\n", " seed=42,\n", ")\n", "print(candle_test_1, values[0], values[-1], timedates[-1])\n", "assert candle_test_1.o in values\n", "assert candle_test_1.h in values\n", "assert candle_test_1.l in values\n", "assert candle_test_1.c in values\n", "\n", "plot_results(values, timedates)" ] }, { "cell_type": "code", "execution_count": null, "id": "3a90bcd5", "metadata": {}, "outputs": [], "source": [ "timedates, values = generate_ticks_from_candle(\n", " duration_in_seconds=1,\n", " open_=candle_test_2.o,\n", " high_=candle_test_2.h,\n", " low_=candle_test_2.l,\n", " close_=candle_test_2.c,\n", " number_of_ticks=240,\n", " seed=42,\n", ")\n", "print(candle_test_2)\n", "plot_results(values, timedates)\n", "assert len(values) == 240, len(values)\n", "assert candle_test_2.o in values\n", "assert candle_test_2.h in values\n", "assert candle_test_2.l in values\n", "assert candle_test_2.c in values" ] }, { "cell_type": "code", "execution_count": null, "id": "4facf9e6", "metadata": {}, "outputs": [], "source": [ "ticks = original_interpolate_candle_to_ticks(\n", " candle=candle_test_1, timeframe=4, compression=1, tick_interval=0.25, seed=42\n", ")\n", "timedates = []\n", "values = []\n", "for tick in ticks:\n", " timedates.append(tick.t)\n", " values.append(tick.bp)\n", "print(candle_test_1.o, candle_test_1)\n", "assert candle_test_1.o in values\n", "assert candle_test_1.l in values\n", "assert candle_test_1.h in values\n", "assert candle_test_1.c in values\n", "plot_results(values, timedates)" ] }, { "cell_type": "code", "execution_count": null, "id": "157803f1", "metadata": {}, "outputs": [], "source": [ "def generate_ticks_from_candle_local(\n", " number_of_ticks: int,\n", " duration_in_seconds: int,\n", " boundary_offset_in_ms: int,\n", " open_: float,\n", " high_: float,\n", " low_: float,\n", " close_: float,\n", " volume_: float,\n", " seed: Optional[int] = None,\n", " jitter_scale_percent: float = 0.02,\n", ") -> tuple[list[int], list[float], list[float]]:\n", " if seed is not None:\n", " np.random.seed(seed)\n", "\n", " # timestamps in ms\n", " timestamps_in_ms = np.linspace(\n", " boundary_offset_in_ms,\n", " duration_in_seconds * 1000,\n", " number_of_ticks,\n", " endpoint=False,\n", " dtype=int,\n", " )\n", "\n", " # volume\n", " per_tick_vol = volume_ / number_of_ticks\n", " volumes = np.full(number_of_ticks, per_tick_vol, dtype=float)\n", "\n", " # trivial case\n", " if number_of_ticks < 4:\n", " base = np.linspace(open_, close_, number_of_ticks)\n", " return list(timestamps_in_ms), list(base), list(volumes)\n", "\n", " # compute distances\n", " d_oh, d_ol = abs(open_ - high_), abs(open_ - low_)\n", " d_ch, d_cl = abs(close_ - high_), abs(close_ - low_)\n", "\n", " # label each extreme by which end it's closer to\n", " side_h = \"open\" if d_oh < d_ch else \"close\" if d_ch < d_oh else \"either\"\n", " side_l = \"open\" if d_ol < d_cl else \"close\" if d_cl < d_ol else \"either\"\n", "\n", " # assign first_extreme & second_extreme\n", " if side_h == \"open\" and side_l == \"close\":\n", " first_extreme, second_extreme = high_, low_\n", " elif side_l == \"open\" and side_h == \"close\":\n", " first_extreme, second_extreme = low_, high_\n", " elif side_h == \"open\" and side_l == \"open\":\n", " # both closer to open → first = the nearer to open\n", " if d_oh < d_ol:\n", " first_extreme, second_extreme = high_, low_\n", " else:\n", " first_extreme, second_extreme = low_, high_\n", " elif side_h == \"close\" and side_l == \"close\":\n", " # both closer to close → second = the nearer to close\n", " if d_ch < d_cl:\n", " first_extreme, second_extreme = low_, high_\n", " else:\n", " first_extreme, second_extreme = high_, low_\n", " else:\n", " # fallback to “closest to open” then “closest to close”\n", " first_extreme = high_ if d_oh < d_ol else low_\n", " cand_second = high_ if d_ch < d_cl else low_\n", " second_extreme = (\n", " cand_second\n", " if cand_second != first_extreme\n", " else (low_ if first_extreme == high_ else high_)\n", " )\n", "\n", " # pick two interior indices for the extremes\n", " idx1 = np.random.randint(1, number_of_ticks - 2)\n", " idx2 = np.random.randint(idx1 + 1, number_of_ticks - 1)\n", "\n", " # define anchors: (position, value)\n", " anchors = [\n", " (0, open_),\n", " (idx1, first_extreme),\n", " (idx2, second_extreme),\n", " (number_of_ticks - 1, close_),\n", " ]\n", "\n", " # jitter scale\n", " jitter_scale = (high_ - low_) * jitter_scale_percent\n", "\n", " ticks = np.zeros(number_of_ticks, dtype=float)\n", "\n", " # build each segment with Brownian‐bridge jitter\n", " for (start_i, start_v), (end_i, end_v) in zip(anchors[:-1], anchors[1:]):\n", " length = end_i - start_i + 1\n", " lin = np.linspace(start_v, end_v, length)\n", "\n", " if length > 2 and jitter_scale > 0:\n", " # generate standard Brownian increments\n", " M = length - 1\n", " eps = np.random.randn(M) * jitter_scale\n", " W = np.concatenate([[0.0], np.cumsum(eps)])\n", " # make it a bridge (0 at both ends)\n", " B = W - np.linspace(0, 1, length) * W[-1]\n", " else:\n", " B = np.zeros(length)\n", "\n", " ticks[start_i : end_i + 1] = lin + B\n", "\n", " # ensure no tick crosses the candle's high or low extremes\n", " ticks = np.clip(ticks, low_, high_)\n", " return list(timestamps_in_ms), list(ticks), list(volumes)" ] }, { "cell_type": "code", "execution_count": null, "id": "357a2884", "metadata": {}, "outputs": [], "source": [ "candle = candle_test_2\n", "t, values, volumes = generate_ticks_from_candle_local(\n", " duration_in_seconds=1,\n", " boundary_offset_in_ms=1,\n", " open_=candle.o,\n", " high_=candle.h,\n", " low_=candle.l,\n", " close_=candle.c,\n", " volume_=candle.v,\n", " number_of_ticks=240,\n", " seed=42,\n", ")\n", "print(candle)\n", "plot_results(values, t)\n", "plot_results(volumes, t)\n", "print(values)\n", "assert candle.o == values[0]\n", "assert candle.l in values[1:-1]\n", "assert candle.h in values[1:-1]\n", "assert candle.c == values[-1]" ] }, { "cell_type": "code", "execution_count": null, "id": "ec50958d", "metadata": {}, "outputs": [], "source": [ "steps = np.random.normal(0, 1, size=240)\n", "plot_results(steps, range(len(steps)))" ] }, { "cell_type": "code", "execution_count": null, "id": "ced94168", "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "id": "abe15200", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.8" } }, "nbformat": 4, "nbformat_minor": 5 }