vectorized aggregator, minor changes

* RunManagerRecord class has a new attribute market. Market enum is imported.

* row_to_runmanager function considers market column

* add_run_manager_record and update_run_manager_record functions are changed. fetch_all_markets_in_run_manager is new.

* new Market enumeration class is defined

* market_value used for job scheduling. start and stop functions have modifications of market parameter input

* new is_market_day function + modifications of get_todays_market_times function

* market attribute set default to US

* row_to_runmanager function has no string formatter for market attribute

* add_run_manager_record function adn update_run_manager_record function update the DB column market based on record.market data

* start_runman_record and stop_runman_record have got no market parameter

* get_todays_market_times function is changed

* default value for market atribute is Market.US

* update_run_manager_record function has no if condition for market key

* market_value deleted, used enumaration value Market.US instead of string US

* get_todays_market_times has a new if condition for Market.CRYPTO

* update includes market column in the run_manager table

* market attribute in Run Manager record has value given by enumeration as Market.US

* documentation of changes made in the branch

* remove README_feature_market.md

* back to original state

* Delete README_feature_market.md

* _start_runman_record has an additional else condition

* is_market_day renamed to is_US_market_day

* transferables column added into runner_header table

research/loading_trades_aggregation.ipynb

@@ -0,0 +1,316 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Loading trades and vectorized aggregation"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import pandas as pd\n",
+    "import numpy as np\n",
+    "from numba import jit\n",
+    "from alpaca.data.historical import StockHistoricalDataClient\n",
+    "from v2realbot.config import ACCOUNT1_PAPER_API_KEY, ACCOUNT1_PAPER_SECRET_KEY, DATA_DIR\n",
+    "from alpaca.data.requests import StockTradesRequest\n",
+    "from v2realbot.enums.enums import BarType\n",
+    "import time\n",
+    "\n",
+    "from datetime import datetime\n",
+    "from v2realbot.utils.utils import parse_alpaca_timestamp, ltp, zoneNY, send_to_telegram, fetch_calendar_data\n",
+    "import pyarrow\n",
+    "from v2realbot.loader.aggregator_vectorized import fetch_daily_stock_trades, fetch_trades_parallel, generate_time_bars_nb, aggregate_trades\n",
+    "import vectorbtpro as vbt\n",
+    "\n",
+    "vbt.settings.set_theme(\"dark\")\n",
+    "vbt.settings['plotting']['layout']['width'] = 1280\n",
+    "vbt.settings.plotting.auto_rangebreaks = True\n",
+    "# Set the option to display with pagination\n",
+    "pd.set_option('display.notebook_repr_html', True)\n",
+    "pd.set_option('display.max_rows', 10)  # Number of rows per page"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "symbol = \"SPY\"\n",
+    "#datetime in zoneNY \n",
+    "day_start = datetime(2024, 5, 15, 9, 30, 0)\n",
+    "day_stop = datetime(2024, 5, 16, 16, 00, 0)\n",
+    "day_start = zoneNY.localize(day_start)\n",
+    "day_stop = zoneNY.localize(day_stop)\n",
+    "#neslo by zrychlit, kdyz se zobrazuje pomalu Searching cache - nejaky bottle neck?\n",
+    "df = fetch_trades_parallel(symbol, day_start, day_stop, minsize=50) #exclude_conditions=['C','O','4','B','7','V','P','W','U','Z','F'])\n",
+    "ohlcv_df = aggregate_trades(symbol=symbol, trades_df=df, resolution=1, type=BarType.TIME)\n",
+    "#df.info()\n",
+    "ohlcv_df\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "df"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "basic_data = vbt.Data.from_data(vbt.symbol_dict({symbol: ohlcv_df}), tz_convert=zoneNY)\n",
+    "vbt.settings['plotting']['auto_rangebreaks'] = True\n",
+    "basic_data.ohlcv.plot()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import pickle\n",
+    "from v2realbot.config import ACCOUNT1_PAPER_API_KEY, ACCOUNT1_PAPER_SECRET_KEY, DATA_DIR\n",
+    "import gzip\n",
+    "\n",
+    "file_path = f\"{DATA_DIR}/tradecache/BAC-1709044200-1709067600.cache.gz\"\n",
+    "\n",
+    "with gzip.open(file_path, 'rb') as fp:\n",
+    "    tradesResponse = pickle.load(fp)\n",
+    "\n",
+    "tradesResponse"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def convert_dict_to_multiindex_df(tradesResponse):\n",
+    "    # Create a DataFrame for each key and add the key as part of the MultiIndex\n",
+    "    dfs = []\n",
+    "    for key, values in tradesResponse.items():\n",
+    "        df = pd.DataFrame(values)\n",
+    "        # Rename columns\n",
+    "        # Select and order columns explicitly\n",
+    "        #print(df)\n",
+    "        df = df[['t', 'x', 'p', 's', 'i', 'c','z']]\n",
+    "        df.rename(columns={'t': 'timestamp', 'c': 'conditions', 'p': 'price', 's': 'size', 'x': 'exchange', 'z':'tape', 'i':'id'}, inplace=True)\n",
+    "        df['symbol'] = key  # Add ticker as a column\n",
+    "        df['timestamp'] = pd.to_datetime(df['timestamp'])  # Convert 't' from string to datetime before setting it as an index\n",
+    "        df.set_index(['symbol', 'timestamp'], inplace=True)  # Set the multi-level index using both 'ticker' and 't'\n",
+    "        df = df.tz_convert(zoneNY, level='timestamp')\n",
+    "        dfs.append(df)\n",
+    "\n",
+    "    # Concatenate all DataFrames into a single DataFrame with MultiIndex\n",
+    "    final_df = pd.concat(dfs)\n",
+    "\n",
+    "    return final_df\n",
+    "\n",
+    "# Convert and print the DataFrame\n",
+    "df = convert_dict_to_multiindex_df(tradesResponse)\n",
+    "df\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "df.info()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "ohlcv_df.info()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "ohlcv_df.info()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "ohlcv_df = aggregate_trades(symbol=symbol, trades_df=df, resolution=1000, type=\"dollar\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "ohlcv_df.index.strftime('%Y-%m-%d %H').unique()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#ohlcv_df.groupby(ohlcv_df.index.date).size()\n",
+    "ohlcv_df.head(100)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "df"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#access just BCA\n",
+    "df_filtered = df.loc[\"BAC\"]\n",
+    "\n",
+    "df_filtered.info()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "df_filtered= df_filtered.reset_index()\n",
+    "ticks = df_filtered[['timestamp', 'price', 'size']].to_numpy()\n",
+    "ticks\n",
+    "timestamps = ticks[:, 0]\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "df_filtered= df_filtered.reset_index()\n",
+    "ticks = df_filtered[['timestamp', 'price', 'size']].to_numpy()\n",
+    "\n",
+    "#timestamp to integer\n",
+    "# Extract the timestamps column (assuming it's the first column)\n",
+    "timestamps = ticks[:, 0]\n",
+    "\n",
+    "# Convert the timestamps to Unix timestamps in seconds with microsecond precision\n",
+    "unix_timestamps_s = np.array([ts.timestamp() for ts in timestamps], dtype='float64')\n",
+    "\n",
+    "# Replace the original timestamps in the NumPy array with the converted Unix timestamps\n",
+    "ticks[:, 0] = unix_timestamps_s\n",
+    "\n",
+    "#ticks[:, 0] = pd.to_datetime(ticks[:, 0]).astype('int64') // 1_000_000_000  # Convert to Unix timestamp\n",
+    "ticks\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "ticks = ticks.astype(np.float64)\n",
+    "ticks"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "\n",
+    "resolution = 1  # Example resolution of 60 seconds\n",
+    "ohlcv_bars = generate_time_bars_nb(ticks, resolution)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "ohlcv_bars"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Convert the resulting array back to a DataFrame\n",
+    "columns = ['time', 'open', 'high', 'low', 'close', 'volume', 'trades']\n",
+    "ohlcv_df = pd.DataFrame(ohlcv_bars, columns=columns)\n",
+    "ohlcv_df['time'] = pd.to_datetime(ohlcv_df['time'], unit='s')\n",
+    "ohlcv_df.set_index('time', inplace=True)\n",
+    "ohlcv_df.index = ohlcv_df.index.tz_localize('UTC').tz_convert(zoneNY)\n",
+    "#ohlcv_df = ohlcv_df.loc[\"2024-03-1 15:50:00\":\"2024-03-28 13:40:00\"]\n",
+    "#ohlcv_df.index.strftime('%Y-%m-%d %H').unique()\n",
+    "\n",
+    "ohlcv_df"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "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.10.11"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

research/test1sbars.ipynb

@@ -0,0 +1,421 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from v2realbot.tools.loadbatch import load_batch\n",
+    "from v2realbot.utils.utils import zoneNY\n",
+    "import pandas as pd\n",
+    "import numpy as np\n",
+    "import vectorbtpro as vbt\n",
+    "from itables import init_notebook_mode, show\n",
+    "\n",
+    "init_notebook_mode(all_interactive=True)\n",
+    "\n",
+    "vbt.settings.set_theme(\"dark\")\n",
+    "vbt.settings['plotting']['layout']['width'] = 1280\n",
+    "vbt.settings.plotting.auto_rangebreaks = True\n",
+    "# Set the option to display with pagination\n",
+    "pd.set_option('display.notebook_repr_html', True)\n",
+    "pd.set_option('display.max_rows', 10)  # Number of rows per page\n",
+    "\n",
+    "res, df = load_batch(batch_id=\"0fb5043a\", #46 days 1.3 - 6.5.\n",
+    "                     space_resolution_evenly=False,\n",
+    "                     indicators_columns=[\"Rsi14\"],\n",
+    "                     main_session_only=True,\n",
+    "                     verbose = False)\n",
+    "if res < 0:\n",
+    "    print(\"Error\" + str(res) + str(df))\n",
+    "df = df[\"bars\"]\n",
+    "\n",
+    "df"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# filter dates"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#filter na dny\n",
+    "# dates_of_interest = pd.to_datetime(['2024-04-22', '2024-04-23']).tz_localize('US/Eastern')\n",
+    "# filtered_df = df.loc[df.index.normalize().isin(dates_of_interest)]\n",
+    "\n",
+    "# df = filtered_df\n",
+    "# df.info()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import plotly.io as pio\n",
+    "pio.renderers.default = 'notebook'\n",
+    "\n",
+    "#naloadujeme do vbt symbol as column\n",
+    "basic_data = vbt.Data.from_data({\"BAC\": df}, tz_convert=zoneNY)\n",
+    "start_date = pd.Timestamp('2024-03-12 09:30', tz=zoneNY)\n",
+    "end_date = pd.Timestamp('2024-03-13 16:00', tz=zoneNY)\n",
+    "\n",
+    "#basic_data = basic_data.transform(lambda df: df[df.index.date == start_date.date()])\n",
+    "#basic_data = basic_data.transform(lambda df: df[(df.index >= start_date) & (df.index <= end_date)])\n",
+    "#basic_data.data[\"BAC\"].info()\n",
+    "\n",
+    "# fig = basic_data.plot(plot_volume=False)\n",
+    "# pivot_info = basic_data.run(\"pivotinfo\", up_th=0.003, down_th=0.002)\n",
+    "# #pivot_info.plot()\n",
+    "# pivot_info.plot(fig=fig, conf_value_trace_kwargs=dict(visible=True))\n",
+    "# fig.show()\n",
+    "\n",
+    "\n",
+    "# rsi14 = basic_data.data[\"BAC\"][\"Rsi14\"].rename(\"Rsi14\")\n",
+    "\n",
+    "# rsi14.vbt.plot().show()\n",
+    "#basic_data.xloc[\"09:30\":\"10:00\"].data[\"BAC\"].vbt.ohlcv.plot().show()\n",
+    "\n",
+    "vbt.settings.plotting.auto_rangebreaks = True\n",
+    "#basic_data.data[\"BAC\"].vbt.ohlcv.plot()\n",
+    "\n",
+    "#basic_data.data[\"BAC\"]"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "m1_data = basic_data[['Open', 'High', 'Low', 'Close', 'Volume']]\n",
+    "\n",
+    "m1_data.data[\"BAC\"]\n",
+    "#m5_data = m1_data.resample(\"5T\")\n",
+    "\n",
+    "#m5_data.data[\"BAC\"].head(10)\n",
+    "\n",
+    "# m15_data = m1_data.resample(\"15T\")\n",
+    "\n",
+    "# m15 = m15_data.data[\"BAC\"]\n",
+    "\n",
+    "# m15.vbt.ohlcv.plot()\n",
+    "\n",
+    "# m1_data.wrapper.index\n",
+    "\n",
+    "# m1_resampler = m1_data.wrapper.get_resampler(\"1T\")\n",
+    "# m1_resampler.index_difference(reverse=True)\n",
+    "\n",
+    "\n",
+    "# m5_resampler.prettify()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# defining ENTRY WINDOW and forced EXIT window"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#m1_data.data[\"BAC\"].info()\n",
+    "import datetime\n",
+    "# Define the market open and close times\n",
+    "market_open = datetime.time(9, 30)\n",
+    "market_close = datetime.time(16, 0)\n",
+    "entry_window_opens = 1\n",
+    "entry_window_closes = 350\n",
+    "\n",
+    "forced_exit_start = 380\n",
+    "forced_exit_end = 390\n",
+    "\n",
+    "forced_exit = m1_data.symbol_wrapper.fill(False)\n",
+    "entry_window_open=  m1_data.symbol_wrapper.fill(False)\n",
+    "\n",
+    "# Calculate the time difference in minutes from market open for each timestamp\n",
+    "elapsed_min_from_open = (forced_exit.index.hour - market_open.hour) * 60 + (forced_exit.index.minute - market_open.minute)\n",
+    "\n",
+    "entry_window_open[(elapsed_min_from_open >= entry_window_opens) & (elapsed_min_from_open < entry_window_closes)] = True\n",
+    "forced_exit[(elapsed_min_from_open >= forced_exit_start) & (elapsed_min_from_open < forced_exit_end)] = True\n",
+    "\n",
+    "#entry_window_open.info()\n",
+    "# forced_exit.tail(100)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "close = m1_data.close\n",
+    "\n",
+    "rsi = vbt.RSI.run(close, window=14)\n",
+    "\n",
+    "long_entries = (rsi.rsi.vbt.crossed_below(20) & entry_window_open)\n",
+    "long_exits = (rsi.rsi.vbt.crossed_above(70) | forced_exit)\n",
+    "#long_entries.info()\n",
+    "#number of trues and falses in long_entries\n",
+    "long_entries.value_counts()\n",
+    "#long_exits.value_counts()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def plot_rsi(rsi, close, entries, exits):\n",
+    "    fig = vbt.make_subplots(rows=1, cols=1, shared_xaxes=True, specs=[[{\"secondary_y\": True}]], vertical_spacing=0.02, subplot_titles=(\"RSI\", \"Price\" ))\n",
+    "    close.vbt.plot(fig=fig, add_trace_kwargs=dict(secondary_y=True))\n",
+    "    rsi.plot(fig=fig, add_trace_kwargs=dict(secondary_y=False))\n",
+    "    entries.vbt.signals.plot_as_entries(rsi.rsi, fig=fig, add_trace_kwargs=dict(secondary_y=False))  \n",
+    "    exits.vbt.signals.plot_as_exits(rsi.rsi, fig=fig, add_trace_kwargs=dict(secondary_y=False))  \n",
+    "    return fig\n",
+    "\n",
+    "plot_rsi(rsi, close, long_entries, long_exits)\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "vbt.phelp(vbt.Portfolio.from_signals)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "sl_stop = np.arange(0.03/100, 0.2/100, 0.02/100).tolist()\n",
+    "# Using the round function\n",
+    "sl_stop = [round(val, 4) for val in sl_stop]\n",
+    "print(sl_stop)\n",
+    "sl_stop = vbt.Param(sl_stop) #np.nan mean s no stoploss\n",
+    "\n",
+    "pf = vbt.Portfolio.from_signals(close=close, entries=long_entries, sl_stop=sl_stop, tp_stop = sl_stop, exits=long_exits,fees=0.0167/100, freq=\"1s\") #sl_stop=sl_stop, tp_stop = sl_stop, \n",
+    "\n",
+    "#pf.stats()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "pf.plot()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "pf[(0.0015,0.0013)].plot()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "pf[0.03].plot_trade_signals()\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# pristup k pf jako multi index"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#pf[0.03].plot()\n",
+    "#pf.order_records\n",
+    "pf[(0.03)].stats()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#zgrupovane statistiky\n",
+    "stats_df = pf.stats([\n",
+    "    'total_return',\n",
+    "    'total_trades',\n",
+    "    'win_rate',\n",
+    "    'expectancy'\n",
+    "], agg_func=None)\n",
+    "stats_df\n",
+    "\n",
+    "\n",
+    "stats_df.nlargest(50, 'Total Return [%]')\n",
+    "#stats_df.info()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "pf[(0.0011,0.0013)].plot()\n",
+    "\n",
+    "#pf[(0.0011,0.0013000000000000002)].plot()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pandas.tseries.offsets import DateOffset\n",
+    "\n",
+    "temp_data = basic_data['2024-4-22']\n",
+    "temp_data\n",
+    "res1m = temp_data[[\"Open\", \"High\", \"Low\", \"Close\", \"Volume\"]]\n",
+    "\n",
+    "# Define a custom date offset that starts at 9:30 AM and spans 4 hours\n",
+    "custom_offset = DateOffset(hours=4, minutes=30)\n",
+    "\n",
+    "# res1m = res1m.get().resample(\"4H\").agg({  \n",
+    "#     \"Open\": \"first\",\n",
+    "#     \"High\": \"max\",\n",
+    "#     \"Low\": \"min\",\n",
+    "#     \"Close\": \"last\",\n",
+    "#     \"Volume\": \"sum\"\n",
+    "# })\n",
+    "\n",
+    "res4h = res1m.resample(\"1h\", resample_kwargs=dict(origin=\"start\"))\n",
+    "\n",
+    "res4h.data\n",
+    "\n",
+    "res15m = res1m.resample(\"15T\", resample_kwargs=dict(origin=\"start\"))\n",
+    "\n",
+    "res15m.data[\"BAC\"]"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "@vbt.njit\n",
+    "def long_entry_place_func_nb(c, low, close, time_in_ns, rsi14, window_open, window_close):\n",
+    "    market_open_minutes = 570  # 9 hours * 60 minutes + 30 minutes\n",
+    "\n",
+    "    for out_i in range(len(c.out)):\n",
+    "        i = c.from_i + out_i\n",
+    "\n",
+    "        current_minutes = vbt.dt_nb.hour_nb(time_in_ns[i]) * 60 + vbt.dt_nb.minute_nb(time_in_ns[i])\n",
+    "        #print(\"current_minutes\", current_minutes)\n",
+    "        # Calculate elapsed minutes since market open at 9:30 AM\n",
+    "        elapsed_from_open = current_minutes - market_open_minutes\n",
+    "        elapsed_from_open = elapsed_from_open if elapsed_from_open >= 0 else 0\n",
+    "        #print( \"elapsed_from_open\", elapsed_from_open)\n",
+    "\n",
+    "        #elapsed_from_open = elapsed_minutes_from_open_nb(time_in_ns) \n",
+    "        in_window = elapsed_from_open > window_open and elapsed_from_open < window_close\n",
+    "        #print(\"in_window\", in_window)\n",
+    "        # if in_window:\n",
+    "        #     print(\"in window\")\n",
+    "\n",
+    "        if in_window and rsi14[i] > 60: # and low[i, c.col] <= hit_price: # and hour == 9:  # (4)!\n",
+    "            return out_i\n",
+    "    return -1\n",
+    "\n",
+    "@vbt.njit\n",
+    "def long_exit_place_func_nb(c, high, close, time_index, tp, sl):  # (5)!\n",
+    "    entry_i = c.from_i - c.wait\n",
+    "    entry_price = close[entry_i, c.col]\n",
+    "    hit_price = entry_price * (1 + tp)\n",
+    "    stop_price = entry_price * (1 - sl)\n",
+    "    for out_i in range(len(c.out)):\n",
+    "        i = c.from_i + out_i\n",
+    "        last_bar_of_day = vbt.dt_nb.day_changed_nb(time_index[i], time_index[i + 1])\n",
+    "\n",
+    "        #print(next_day)\n",
+    "        if last_bar_of_day: #pokud je dalsi next day, tak zavirame posledni\n",
+    "            print(\"ted\",out_i)\n",
+    "            return out_i\n",
+    "        if close[i, c.col] >= hit_price or close[i, c.col] <= stop_price :\n",
+    "            return out_i\n",
+    "    return -1\n",
+    "\n",
+    "\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "df = pd.DataFrame(np.random.random(size=(5, 10)), columns=list('abcdefghij'))\n",
+    "\n",
+    "df"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "df.sum()"
+   ]
+  }
+ ],
+ "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.10.11"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

restart.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
 # file: restart.sh
-
+ 
 # Usage: ./restart.sh [test|prod|all]
 
 # Define server addresses

v2realbot/common/model.py

@@ -5,7 +5,7 @@ from rich import print
 from typing import Any, Optional, List, Union
 from datetime import datetime, date
 from pydantic import BaseModel, Field
-from v2realbot.enums.enums import Mode, Account, SchedulerStatus, Moddus
+from v2realbot.enums.enums import Mode, Account, SchedulerStatus, Moddus, Market
 from alpaca.data.enums import Exchange
 
 
@@ -159,6 +159,7 @@ class RunManagerRecord(BaseModel):
     mode: Mode
     note: Optional[str] = None
     ilog_save: bool = False
+    market: Optional[Market] = Market.US
     bt_from: Optional[datetime] = None
     bt_to: Optional[datetime] = None
     #weekdays filter

v2realbot/common/transform.py

@@ -5,9 +5,7 @@ import v2realbot.controller.services as cs
 
 #prevede dict radku zpatky na objekt vcetme retypizace
 def row_to_runmanager(row: dict) -> RunManagerRecord:
-
     is_running = cs.is_runner_running(row['runner_id']) if row['runner_id'] else False
-
     res = RunManagerRecord(
         moddus=row['moddus'],
         id=row['id'],
@@ -17,6 +15,7 @@ def row_to_runmanager(row: dict) -> RunManagerRecord:
         account=row['account'],
         note=row['note'],
         ilog_save=bool(row['ilog_save']),
+        market=row['market'] if row['market'] is not None else None,
         bt_from=datetime.fromisoformat(row['bt_from']) if row['bt_from'] else None,
         bt_to=datetime.fromisoformat(row['bt_to']) if row['bt_to'] else None,
         weekdays_filter=[int(x) for x in row['weekdays_filter'].split(',')] if row['weekdays_filter'] else [],

v2realbot/controller/run_manager.py

@@ -172,14 +172,14 @@ def add_run_manager_record(new_record: RunManagerRecord):
         # Construct a suitable INSERT query based on your RunManagerRecord fields
         insert_query = """
             INSERT INTO run_manager (moddus, id, strat_id, symbol,account, mode, note,ilog_save,
-                                    bt_from, bt_to, weekdays_filter, batch_id,
+                                    market, bt_from, bt_to, weekdays_filter, batch_id,
                                     start_time, stop_time, status, last_processed,
                                     history, valid_from, valid_to, testlist_id)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?) 
         """
         values = [
             new_record.moddus, str(new_record.id), str(new_record.strat_id), new_record.symbol, new_record.account, new_record.mode, new_record.note,
-            int(new_record.ilog_save),
+            int(new_record.ilog_save), new_record.market,
             new_record.bt_from.isoformat() if new_record.bt_from is not None else None,
             new_record.bt_to.isoformat() if new_record.bt_to is not None else None,
             ",".join(str(x) for x in new_record.weekdays_filter) if new_record.weekdays_filter else None,

v2realbot/enums/enums.py

@@ -1,6 +1,11 @@
 from enum import Enum
 from alpaca.trading.enums import OrderSide, OrderStatus, OrderType
 
+class BarType(str, Enum):
+    TIME = "time"
+    VOLUME = "volume"
+    DOLLAR = "dollar"
+
 class Env(str, Enum):
     PROD = "prod"
     TEST = "test"
@@ -103,4 +108,10 @@ class StartBarAlign(str, Enum):
         RANDOM = first bar starts when first trade occurs
     """ 
     ROUND = "round"
-    RANDOM = "random"
+    RANDOM = "random"
+
+class Market(str, Enum):
+    US = "US"
+    CRYPTO = "CRYPTO"
+
+     

v2realbot/loader/aggregator_vectorized.py

@@ -0,0 +1,535 @@
+import pandas as pd
+import numpy as np
+from numba import jit
+from alpaca.data.historical import StockHistoricalDataClient
+from sqlalchemy import column
+from v2realbot.config import ACCOUNT1_PAPER_API_KEY, ACCOUNT1_PAPER_SECRET_KEY, DATA_DIR
+from alpaca.data.requests import StockTradesRequest
+import time as time_module
+from v2realbot.utils.utils import parse_alpaca_timestamp, ltp, zoneNY, send_to_telegram, fetch_calendar_data
+import pyarrow
+from traceback import format_exc
+from datetime import timedelta, datetime, time
+from concurrent.futures import ThreadPoolExecutor
+import os
+import gzip
+import pickle
+import random
+from alpaca.data.models import BarSet, QuoteSet, TradeSet
+import v2realbot.utils.config_handler as cfh
+from v2realbot.enums.enums import BarType
+""""
+Module used for vectorized aggregation of trades.
+
+Includes fetch (remote/cached) methods and numba aggregator function for TIME BASED, VOLUME BASED and DOLLAR BARS
+
+"""""
+
+def aggregate_trades(symbol: str, trades_df: pd.DataFrame, resolution: int, type: BarType = BarType.TIME):
+    """"
+    Accepts dataframe with trades keyed by symbol. Preparess dataframe to 
+    numpy and call nNumba optimized aggregator for given bar type. (time/volume/dollar)
+    """""
+    trades_df = trades_df.loc[symbol]
+    trades_df= trades_df.reset_index()
+    ticks = trades_df[['timestamp', 'price', 'size']].to_numpy()
+    # Extract the timestamps column (assuming it's the first column)
+    timestamps = ticks[:, 0]
+    # Convert the timestamps to Unix timestamps in seconds with microsecond precision
+    unix_timestamps_s = np.array([ts.timestamp() for ts in timestamps], dtype='float64')
+    # Replace the original timestamps in the NumPy array with the converted Unix timestamps
+    ticks[:, 0] = unix_timestamps_s
+    ticks = ticks.astype(np.float64)
+    #based on type, specific aggregator function is called
+    match type:
+        case BarType.TIME:
+            ohlcv_bars = generate_time_bars_nb(ticks, resolution)
+        case BarType.VOLUME:
+            ohlcv_bars = generate_volume_bars_nb(ticks, resolution)
+        case BarType.DOLLAR:
+            ohlcv_bars = generate_dollar_bars_nb(ticks, resolution)
+        case _:
+            raise ValueError("Invalid bar type. Supported types are 'time', 'volume' and 'dollar'.")
+    # Convert the resulting array back to a DataFrame
+    columns = ['time', 'open', 'high', 'low', 'close', 'volume', 'trades']
+    if type == BarType.DOLLAR:
+        columns.append('amount')
+    ohlcv_df = pd.DataFrame(ohlcv_bars, columns=columns)
+    ohlcv_df['time'] = pd.to_datetime(ohlcv_df['time'], unit='s')
+    ohlcv_df.set_index('time', inplace=True)
+    ohlcv_df.index = ohlcv_df.index.tz_localize('UTC').tz_convert(zoneNY)
+    return ohlcv_df
+
+def convert_dict_to_multiindex_df(tradesResponse):
+    """"
+    Converts dictionary from cache or from remote (raw input) to multiindex dataframe.
+    """""
+    # Create a DataFrame for each key and add the key as part of the MultiIndex
+    dfs = []
+    for key, values in tradesResponse.items():
+        df = pd.DataFrame(values)
+        # Rename columns
+        # Select and order columns explicitly
+        #print(df)
+        df = df[['t', 'x', 'p', 's', 'i', 'c','z']]
+        df.rename(columns={'t': 'timestamp', 'c': 'conditions', 'p': 'price', 's': 'size', 'x': 'exchange', 'z':'tape', 'i':'id'}, inplace=True)
+        df['symbol'] = key  # Add ticker as a column
+        df['timestamp'] = pd.to_datetime(df['timestamp'])  # Convert 't' from string to datetime before setting it as an index
+        df.set_index(['symbol', 'timestamp'], inplace=True)  # Set the multi-level index using both 'ticker' and 't'
+        df = df.tz_convert(zoneNY, level='timestamp')
+        dfs.append(df)
+
+    # Concatenate all DataFrames into a single DataFrame with MultiIndex
+    final_df = pd.concat(dfs)
+
+    return final_df
+
+def dict_to_df(tradesResponse, start, end, exclude_conditions = None, minsize = None):
+    """"
+    Transforms dict to Tradeset, then df and to zone aware
+    Also filters to start and end if necessary (ex. 9:30 to 15:40 is required only)
+   
+   NOTE: prepodkladame, ze tradesResponse je dict from Raw data (cached/remote)
+   """""
+
+    df = convert_dict_to_multiindex_df(tradesResponse)
+
+    #REQUIRED FILTERING
+    #pokud je zacatek pozdeji nebo konec driv tak orizneme
+    if (start.time() > time(9, 30) or end.time() < time(16, 0)):
+        print(f"filtrujeme {start.time()} {end.time()}")
+        # Define the time range
+        # start_time = pd.Timestamp(start.time(), tz=zoneNY).time()
+        # end_time = pd.Timestamp(end.time(), tz=zoneNY).time()
+
+        # Create a mask to filter rows within the specified time range
+        mask = (df.index.get_level_values('timestamp') >= start) & \
+            (df.index.get_level_values('timestamp') <= end)
+
+        # Apply the mask to the DataFrame
+        df = df[mask]
+
+    if exclude_conditions is not None:
+        print(f"excluding conditions {exclude_conditions}")
+        # Create a mask to exclude rows with any of the specified conditions
+        mask = df['conditions'].apply(lambda x: any(cond in exclude_conditions for cond in x))
+
+        # Filter out the rows with specified conditions
+        df = df[~mask]
+
+    if minsize is not None:
+        print(f"minsize {minsize}")
+        #exclude conditions
+        df = df[df['size'] >= minsize]
+    return df
+
+#fetches daily stock tradess - currently only main session is supported
+def fetch_daily_stock_trades_old(symbol, start, end, exclude_conditions = None, minsize = None, force_remote = False, max_retries=5, backoff_factor=1):
+    """
+    Attempts to fetch stock trades with exponential backoff. Raises an exception if all retries fail.
+
+    :param symbol: The stock symbol to fetch trades for.
+    :param start: The start time for the trade data.
+    :param end: The end time for the trade data.
+    :param max_retries: Maximum number of retries.
+    :param backoff_factor: Factor to determine the next sleep time.
+    :return: TradesResponse object.
+    :raises: ConnectionError if all retries fail.
+
+    We use tradecache only for main sessison request = 9:30 to 16:00
+    """
+    use_daily_tradecache = False
+    if (start.time() >= time(9, 30) and end.time() <= time(16, 0)):
+        use_daily_tradecache = True
+        filename_start = zoneNY.localize(datetime.combine(start.date(), time(9, 30)))
+        filename_end= zoneNY.localize(datetime.combine(end.date(), time(16, 0)))
+        daily_file = "TS" + str(symbol) + '-' + str(int(filename_start.timestamp())) + '-' + str(int(filename_end.timestamp())) + '.cache.gz'
+        file_path = DATA_DIR + "/tradecache/"+daily_file
+
+    if use_daily_tradecache and not force_remote and os.path.exists(file_path):
+        print("Searching cache: " + daily_file)
+        with gzip.open (file_path, 'rb') as fp:
+            tradesResponse = pickle.load(fp)
+            print("FOUND in CACHE", daily_file)
+            #response je vzdy ulozena jako raw(dict), davame zpet do TradeSetu, ktery umi i df
+            return dict_to_df(tradesResponse, start, end, exclude_conditions, minsize)
+
+    #daily file doesnt exist
+    else:
+        print("NOT FOUND. Fetching from remote")
+        client =  StockHistoricalDataClient(ACCOUNT1_PAPER_API_KEY, ACCOUNT1_PAPER_SECRET_KEY, raw_data=False)
+        stockTradeRequest = StockTradesRequest(symbol_or_symbols=symbol, start=start, end=end)
+        last_exception = None
+
+        for attempt in range(max_retries):
+            try:
+                tradesResponse = client.get_stock_trades(stockTradeRequest)
+                is_empty = not tradesResponse[symbol]
+                print(f"Remote fetched: {is_empty=}", start, end)
+                #pokud jde o dnešní den a nebyl konec trhu tak cache neukládáme, pripadne pri iex datapointu necachujeme
+                if use_daily_tradecache and not is_empty:
+                    if (start < datetime.now().astimezone(zoneNY) < end):
+                        print("not saving trade cache, market still open today")
+                    else:
+                        with gzip.open(file_path, 'wb') as fp:
+                            pickle.dump(tradesResponse, fp)
+                            print("Saving to Trade CACHE", file_path)
+                return pd.DataFrame() if is_empty else dict_to_df(tradesResponse, start, end) 
+            except Exception as e:
+                print(f"Attempt {attempt + 1} failed: {e}")
+                last_exception = e
+                time_module.sleep(backoff_factor * (2 ** attempt))
+
+        print("All attempts to fetch data failed.")
+        raise ConnectionError(f"Failed to fetch stock trades after {max_retries} retries. Last exception: {str(last_exception)} and {format_exc()}")
+
+def fetch_daily_stock_trades(symbol, start, end, exclude_conditions=None, minsize=None, force_remote=False, max_retries=5, backoff_factor=1):
+    #doc for this function
+    """
+    Attempts to fetch stock trades either from cache or remote. When remote, it uses retry mechanism with exponential backoff.
+    Also it stores the data to cache if it is not already there.
+    by using force_remote - forcess using remote data always and thus refreshing cache for these dates
+        Attributes:
+        :param symbol: The stock symbol to fetch trades for.
+        :param start: The start time for the trade data.
+        :param end: The end time for the trade data.
+        :exclude_conditions: list of string conditions to exclude from the data
+        :minsize minimum size of trade to be included in the data
+        :force_remote will always use remote data and refresh cache
+        :param max_retries: Maximum number of retries.
+        :param backoff_factor: Factor to determine the next sleep time.
+        :return: TradesResponse object.
+        :raises: ConnectionError if all retries fail.
+    
+    We use tradecache only for main sessison requests = 9:30 to 16:00
+    Do budoucna ukládat celý den BAC-20240203.cache.gz a z toho si pak filtrovat bud main sesssionu a extended
+    Ale zatim je uloženo jen main session v BAC-timestampopenu-timestampclose.cache.gz
+    """
+    # Determine if the requested times fall within the main session
+    in_main_session = (time(9, 30) <= start.time() < time(16, 0)) and (time(9, 30) <= end.time() <= time(16, 0))
+    file_path = ''
+    
+    if in_main_session:
+        filename_start = zoneNY.localize(datetime.combine(start.date(), time(9, 30)))
+        filename_end = zoneNY.localize(datetime.combine(end.date(), time(16, 0)))
+        daily_file = f"{symbol}-{int(filename_start.timestamp())}-{int(filename_end.timestamp())}.cache.gz"
+        file_path = f"{DATA_DIR}/tradecache/{daily_file}"
+        if not force_remote and os.path.exists(file_path):
+            print("Searching cache: " + daily_file)
+            with gzip.open(file_path, 'rb') as fp:
+                tradesResponse = pickle.load(fp)
+                print("FOUND in CACHE", daily_file)
+                return dict_to_df(tradesResponse, start, end, exclude_conditions, minsize)
+
+    print("NOT FOUND. Fetching from remote")
+    client = StockHistoricalDataClient(ACCOUNT1_PAPER_API_KEY, ACCOUNT1_PAPER_SECRET_KEY, raw_data=True)
+    stockTradeRequest = StockTradesRequest(symbol_or_symbols=symbol, start=start, end=end)
+    last_exception = None
+
+    for attempt in range(max_retries):
+        try:
+            tradesResponse = client.get_stock_trades(stockTradeRequest)
+            is_empty = not tradesResponse[symbol]
+            print(f"Remote fetched: {is_empty=}", start, end)
+            if in_main_session and not is_empty:
+                current_time = datetime.now().astimezone(zoneNY)
+                if not (start < current_time < end):
+                    with gzip.open(file_path, 'wb') as fp:
+                        pickle.dump(tradesResponse, fp)
+                        print("Saving to Trade CACHE", file_path)
+
+                else:  # Don't save the cache if the market is still open
+                    print("Not saving trade cache, market still open today")
+            return pd.DataFrame() if is_empty else dict_to_df(tradesResponse, start, end) 
+        except Exception as e:
+            print(f"Attempt {attempt + 1} failed: {e}")
+            last_exception = e
+            time_module.sleep(backoff_factor * (2 ** attempt) + random.uniform(0, 1))  # Adding random jitter
+
+    print("All attempts to fetch data failed.")
+    raise ConnectionError(f"Failed to fetch stock trades after {max_retries} retries. Last exception: {str(last_exception)} and {format_exc()}")
+
+
+def fetch_trades_parallel(symbol, start_date, end_date, exclude_conditions = cfh.config_handler.get_val('AGG_EXCLUDED_TRADES'), minsize = 100, force_remote = False):
+    """
+    Fetches trades for each day between start_date and end_date during market hours (9:30-16:00) in parallel and concatenates them into a single DataFrame.
+
+    :param symbol: Stock symbol.
+    :param start_date: Start date as datetime.
+    :param end_date: End date as datetime.
+    :return: DataFrame containing all trades from start_date to end_date.
+    """
+    futures = []
+    results = []
+    
+    market_open_days = fetch_calendar_data(start_date, end_date)
+    day_count = len(market_open_days)
+    print("Contains", day_count, " market days")
+    max_workers = min(10, max(5, day_count // 2))  # Heuristic: half the days to process, but at least 1 and no more than 10
+
+
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        #for single_date in (start_date + timedelta(days=i) for i in range((end_date - start_date).days + 1)):
+        for market_day in market_open_days:
+            #start = datetime.combine(single_date, time(9, 30))  # Market opens at 9:30 AM
+            #end = datetime.combine(single_date, time(16, 0))   # Market closes at 4:00 PM
+            
+            interval_from = zoneNY.localize(market_day.open)
+            interval_to = zoneNY.localize(market_day.close)
+
+            #pripadne orizneme pokud je pozadovane pozdejsi zacatek a drivejsi konek
+            start = start_date if interval_from < start_date else interval_from
+            #start = max(start_date, interval_from)
+            end = end_date if interval_to > end_date else interval_to
+            #end = min(end_date, interval_to)
+
+            future = executor.submit(fetch_daily_stock_trades, symbol, start, end, exclude_conditions, minsize, force_remote)
+            futures.append(future)
+        
+        for future in futures:
+            try:
+                result = future.result()
+                results.append(result)
+            except Exception as e:
+                print(f"Error fetching data for a day: {e}")
+
+    return pd.concat(results, ignore_index=False)
+
+@jit(nopython=True)
+def generate_dollar_bars_nb(ticks, amount_per_bar):
+    """"
+    Generates Dollar based bars from ticks.
+
+    There is also simple prevention of aggregation from different days
+    as described here https://chatgpt.com/c/17804fc1-a7bc-495d-8686-b8392f3640a2
+    Downside: split days by UTC (which is ok for main session, but when extended hours it should be reworked by preprocessing new column identifying session)
+    
+    
+    When trade is split into multiple bars it is counted as trade in each of the bars.
+    Other option: trade count can be proportionally distributed by weight (0.2 to 1st bar, 0.8 to 2nd bar) - but this is not implemented yet
+    https://chatgpt.com/c/ff4802d9-22a2-4b72-8ab7-97a91e7a515f
+    """""
+    ohlcv_bars = []
+    remaining_amount = amount_per_bar
+
+    # Initialize bar values based on the first tick to avoid uninitialized values
+    open_price = ticks[0, 1]
+    high_price = ticks[0, 1]
+    low_price = ticks[0, 1]
+    close_price = ticks[0, 1]
+    volume = 0
+    trades_count = 0
+    current_day = np.floor(ticks[0, 0] / 86400)  # Calculate the initial day from the first tick timestamp
+    bar_time = ticks[0, 0]  # Initialize bar time with the time of the first tick
+
+    for tick in ticks:
+        tick_time = tick[0]
+        price = tick[1]
+        tick_volume = tick[2]
+        tick_amount = price * tick_volume
+        tick_day = np.floor(tick_time / 86400)  # Calculate the day of the current tick
+
+        # Check if the new tick is from a different day, then close the current bar
+        if tick_day != current_day:
+            if trades_count > 0:
+                ohlcv_bars.append([bar_time, open_price, high_price, low_price, close_price, volume, trades_count, amount_per_bar])
+            # Reset for the new day using the current tick data
+            open_price = price
+            high_price = price
+            low_price = price
+            close_price = price
+            volume = 0
+            trades_count = 0
+            remaining_amount = amount_per_bar
+            current_day = tick_day
+            bar_time = tick_time
+
+        # Start new bar if needed because of the dollar value
+        while tick_amount > 0:
+            if tick_amount < remaining_amount:
+                # Add the entire tick to the current bar
+                high_price = max(high_price, price)
+                low_price = min(low_price, price)
+                close_price = price
+                volume += tick_volume
+                remaining_amount -= tick_amount
+                trades_count += 1
+                tick_amount = 0
+            else:
+                # Calculate the amount of volume that fits within the remaining dollar amount
+                volume_to_add = remaining_amount / price
+                volume += volume_to_add  # Update the volume here before appending and resetting
+
+                # Append the partially filled bar to the list
+                ohlcv_bars.append([bar_time, open_price, high_price, low_price, close_price, volume, trades_count + 1, amount_per_bar])
+
+                # Fill the current bar and continue with a new bar
+                tick_volume -= volume_to_add
+                tick_amount -= remaining_amount
+
+                # Reset bar values for the new bar using the current tick data
+                open_price = price
+                high_price = price
+                low_price = price
+                close_price = price
+                volume = 0  # Reset volume for the new bar
+                trades_count = 0
+                remaining_amount = amount_per_bar
+                
+                # Increment bar time if splitting a trade
+                if tick_volume > 0: #pokud v tradu je jeste zbytek nastavujeme cas o nanosekundu vetsi
+                    bar_time = tick_time + 1e-6
+                else:
+                    bar_time = tick_time #jinak nastavujeme cas ticku
+                #bar_time = tick_time
+
+    # Add the last bar if it contains any trades
+    if trades_count > 0:
+        ohlcv_bars.append([bar_time, open_price, high_price, low_price, close_price, volume, trades_count, amount_per_bar])
+
+    return np.array(ohlcv_bars)
+
+
+@jit(nopython=True)
+def generate_volume_bars_nb(ticks, volume_per_bar):
+    """"
+    Generates Volume based bars from ticks.
+    
+    NOTE: UTC day split here (doesnt aggregate trades from different days)
+    but realized from UTC (ok for main session) - but needs rework for extension by preprocessing ticks_df and introduction sesssion column
+    
+    When trade is split into multiple bars it is counted as trade in each of the bars.
+    Other option: trade count can be proportionally distributed by weight (0.2 to 1st bar, 0.8 to 2nd bar) - but this is not implemented yet
+    https://chatgpt.com/c/ff4802d9-22a2-4b72-8ab7-97a91e7a515f
+    """""
+    ohlcv_bars = []
+    remaining_volume = volume_per_bar
+
+    # Initialize bar values based on the first tick to avoid uninitialized values
+    open_price = ticks[0, 1]
+    high_price = ticks[0, 1]
+    low_price = ticks[0, 1]
+    close_price = ticks[0, 1]
+    volume = 0
+    trades_count = 0
+    current_day = np.floor(ticks[0, 0] / 86400)  # Calculate the initial day from the first tick timestamp
+    bar_time = ticks[0, 0]  # Initialize bar time with the time of the first tick
+
+    for tick in ticks:
+        tick_time = tick[0]
+        price = tick[1]
+        tick_volume = tick[2]
+        tick_day = np.floor(tick_time / 86400)  # Calculate the day of the current tick
+
+        # Check if the new tick is from a different day, then close the current bar
+        if tick_day != current_day:
+            if trades_count > 0:
+                ohlcv_bars.append([bar_time, open_price, high_price, low_price, close_price, volume, trades_count])
+            # Reset for the new day using the current tick data
+            open_price = price
+            high_price = price
+            low_price = price
+            close_price = price
+            volume = 0
+            trades_count = 0
+            remaining_volume = volume_per_bar
+            current_day = tick_day
+            bar_time = tick_time  # Update bar time to the current tick time
+
+        # Start new bar if needed because of the volume
+        while tick_volume > 0:
+            if tick_volume < remaining_volume:
+                # Add the entire tick to the current bar
+                high_price = max(high_price, price)
+                low_price = min(low_price, price)
+                close_price = price
+                volume += tick_volume
+                remaining_volume -= tick_volume
+                trades_count += 1
+                tick_volume = 0
+            else:
+                # Fill the current bar and continue with a new bar
+                volume_to_add = remaining_volume
+                volume += volume_to_add
+                tick_volume -= volume_to_add
+                trades_count += 1
+                # Append the completed bar to the list
+                ohlcv_bars.append([bar_time, open_price, high_price, low_price, close_price, volume, trades_count])
+
+                # Reset bar values for the new bar using the current tick data
+                open_price = price
+                high_price = price
+                low_price = price
+                close_price = price
+                volume = 0
+                trades_count = 0
+                remaining_volume = volume_per_bar
+                                # Increment bar time if splitting a trade
+                if tick_volume > 0: #pokud v tradu je jeste zbytek nastavujeme cas o nanosekundu vetsi
+                    bar_time = tick_time + 1e-6
+                else:
+                    bar_time = tick_time #jinak nastavujeme cas ticku
+
+
+    # Add the last bar if it contains any trades
+    if trades_count > 0:
+        ohlcv_bars.append([bar_time, open_price, high_price, low_price, close_price, volume, trades_count])
+
+    return np.array(ohlcv_bars)
+
+@jit(nopython=True)
+def generate_time_bars_nb(ticks, resolution):
+    # Initialize the start and end time
+    start_time = np.floor(ticks[0, 0] / resolution) * resolution
+    end_time = np.floor(ticks[-1, 0] / resolution) * resolution
+    
+    # # Calculate number of bars
+    # num_bars = int((end_time - start_time) // resolution + 1)
+    
+    # Using a list to append data only when trades exist
+    ohlcv_bars = []
+    
+    # Variables to track the current bar
+    current_bar_index = -1
+    open_price = 0
+    high_price = -np.inf
+    low_price = np.inf
+    close_price = 0
+    volume = 0
+    trades_count = 0
+    
+    for tick in ticks:
+        tick_time = np.floor(tick[0] / resolution) * resolution
+        price = tick[1]
+        tick_volume = tick[2]
+        
+        # Check if the tick belongs to a new bar
+        if tick_time != start_time + current_bar_index * resolution:
+            if current_bar_index >= 0 and trades_count > 0:  # Save the previous bar if trades happened
+                ohlcv_bars.append([start_time + current_bar_index * resolution, open_price, high_price, low_price, close_price, volume, trades_count])
+            
+            # Reset bar values
+            current_bar_index = int((tick_time - start_time) / resolution)
+            open_price = price
+            high_price = price
+            low_price = price
+            volume = 0
+            trades_count = 0
+        
+        # Update the OHLCV values for the current bar
+        high_price = max(high_price, price)
+        low_price = min(low_price, price)
+        close_price = price
+        volume += tick_volume
+        trades_count += 1
+    
+    # Save the last processed bar
+    if trades_count > 0:
+        ohlcv_bars.append([start_time + current_bar_index * resolution, open_price, high_price, low_price, close_price, volume, trades_count])
+    
+    return np.array(ohlcv_bars)
+
+# Example usage
+if __name__ == '__main__':
+    pass
+    #example in agg_vect.ipynb

v2realbot/scheduler/ap_scheduler.py

@@ -2,7 +2,7 @@ from uuid import UUID
 from typing import Any, List, Tuple
 from uuid import UUID, uuid4
 from v2realbot.enums.enums import Moddus, SchedulerStatus, RecordType, StartBarAlign, Mode, Account, OrderSide
-from v2realbot.common.model import RunManagerRecord, StrategyInstance, RunDay, StrategyInstance, Runner, RunRequest, RunArchive, RunArchiveView, RunArchiveViewPagination, RunArchiveDetail, RunArchiveChange, Bar, TradeEvent, TestList, Intervals, ConfigItem, InstantIndicator, DataTablesRequest
+from v2realbot.common.model import RunManagerRecord, StrategyInstance, RunDay, StrategyInstance, Runner, RunRequest, RunArchive, RunArchiveView, RunArchiveViewPagination, RunArchiveDetail, RunArchiveChange, Bar, TradeEvent, TestList, Intervals, ConfigItem, InstantIndicator, DataTablesRequest, Market
 from v2realbot.utils.utils import validate_and_format_time, AttributeDict, zoneNY, zonePRG, safe_get, dict_replace_value, Store, parse_toml_string, json_serial, is_open_hours, send_to_telegram, concatenate_weekdays, transform_data
 from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus, TradeStoplossType
 from datetime import datetime
@@ -116,7 +116,8 @@ def initialize_jobs(run_manager_records: RunManagerRecord = None):
             scheduler.add_job(start_runman_record, start_trigger, id=f"scheduler_start_{record.id}", args=[record.id])
             scheduler.add_job(stop_runman_record, stop_trigger, id=f"scheduler_stop_{record.id}", args=[record.id])
                         
-    #scheduler.add_job(print_hello, 'interval', seconds=10, id=f"scheduler_testinterval")
+    #scheduler.add_job(print_hello, 'interval', seconds=10, id=
+    # f"scheduler_testinterval")
     scheduled_jobs = scheduler.get_jobs()
     print(f"APS jobs refreshed ({len(scheduled_jobs)})")
     current_jobs_dict = format_apscheduler_jobs(scheduled_jobs)
@@ -124,9 +125,9 @@ def initialize_jobs(run_manager_records: RunManagerRecord = None):
     return 0, current_jobs_dict
 
 #zastresovaci funkce resici error handling a printing
-def start_runman_record(id: UUID, market = "US", debug_date = None):
+def start_runman_record(id: UUID, debug_date = None):
     record = None
-    res, record, msg = _start_runman_record(id=id, market=market, debug_date=debug_date)
+    res, record, msg = _start_runman_record(id=id, debug_date=debug_date)
 
     if record is not None:
         market_time_now = datetime.now().astimezone(zoneNY) if debug_date is None else debug_date
@@ -165,8 +166,8 @@ def update_runman_record(record: RunManagerRecord):
         err_msg= f"STOP: Error updating {record.id} errir {set} with values {record}"
         return -2, err_msg#toto stopne zpracovani dalsich zaznamu pri chybe, zvazit continue
 
-def stop_runman_record(id: UUID, market = "US", debug_date = None):
-    res, record, msg = _stop_runman_record(id=id, market=market, debug_date=debug_date)
+def stop_runman_record(id: UUID, debug_date = None):
+    res, record, msg = _stop_runman_record(id=id, debug_date=debug_date)
     #results : 0 - ok, -1 not running/already running/not specific, -2 error
 
     #report vzdy zapiseme do history, pokud je record not None, pripadna chyba se stala po dotazeni recordu
@@ -196,7 +197,7 @@ def stop_runman_record(id: UUID, market = "US", debug_date = None):
         print(f"STOP JOB: {id} FINISHED")
 
 #start function that is called from the job
-def _start_runman_record(id: UUID, market = "US", debug_date = None):
+def _start_runman_record(id: UUID, debug_date = None):
     print(f"Start scheduled record {id}")
 
     record : RunManagerRecord = None
@@ -207,15 +208,16 @@ def _start_runman_record(id: UUID, market = "US", debug_date = None):
     
     record = result
 
-    if market is not None and market == "US":
-        res, sada = sch.get_todays_market_times(market=market, debug_date=debug_date)
+    if record.market == Market.US or record.market == Market.CRYPTO:
+        res, sada = sch.get_todays_market_times(market=record.market, debug_date=debug_date)
         if res == 0:
             market_time_now, market_open_datetime, market_close_datetime = sada
             print(f"OPEN:{market_open_datetime} CLOSE:{market_close_datetime}")
         else:
-            sada = f"Market {market} Error getting market times (CLOSED): " + str(sada)
+            sada = f"Market {record.market} Error getting market times (CLOSED): " + str(sada)
             return res, record, sada
-    
+    else:
+        print("Market type is unknown.")
     if cs.is_stratin_running(record.strat_id):
         return -1, record, f"Stratin {record.strat_id} is already running"
 
@@ -229,7 +231,7 @@ def _start_runman_record(id: UUID, market = "US", debug_date = None):
     return 0, record, record.runner_id
 
 #stop function that is called from the job
-def _stop_runman_record(id: UUID, market = "US", debug_date = None):
+def _stop_runman_record(id: UUID, debug_date = None):
     record = None
     #get all records
     print(f"Stopping record {id}")
@@ -304,5 +306,5 @@ if __name__ == "__main__":
     # print(f"CALL FINISHED, with {debug_date} RESULT: {res}, {result}")
 
 
-    res, result = stop_runman_record(id=id, market = "US", debug_date = debug_date)
+    res, result = stop_runman_record(id=id, debug_date = debug_date)
     print(f"CALL FINISHED, with {debug_date} RESULT: {res}, {result}")

v2realbot/scheduler/scheduler.py

@@ -2,10 +2,10 @@ import json
 import datetime
 import v2realbot.controller.services as cs
 import v2realbot.controller.run_manager as rm
-from v2realbot.common.model import RunnerView, RunManagerRecord, StrategyInstance, Runner, RunRequest, Trade, RunArchive, RunArchiveView, RunArchiveViewPagination, RunArchiveDetail, Bar, RunArchiveChange, TestList, ConfigItem, InstantIndicator, DataTablesRequest, AnalyzerInputs
+from v2realbot.common.model import RunnerView, RunManagerRecord, StrategyInstance, Runner, RunRequest, Trade, RunArchive, RunArchiveView, RunArchiveViewPagination, RunArchiveDetail, Bar, RunArchiveChange, TestList, ConfigItem, InstantIndicator, DataTablesRequest, AnalyzerInputs, Market
 from uuid import uuid4, UUID
-from v2realbot.utils.utils import json_serial, send_to_telegram, zoneNY, zonePRG, fetch_calendar_data
-from datetime import datetime, timedelta
+from v2realbot.utils.utils import json_serial, send_to_telegram, zoneNY, zonePRG, zoneUTC, fetch_calendar_data
+from datetime import datetime, timedelta, time
 from traceback import format_exc
 from rich import print
 import requests
@@ -18,9 +18,18 @@ from v2realbot.config import WEB_API_KEY
 #naplanovany jako samostatni job a triggerován pouze jednou v daný čas pro start a stop 
 #novy kod v aps_scheduler.py
 
-def get_todays_market_times(market = "US", debug_date = None):
+def is_US_market_day(date):
+    cal_dates = fetch_calendar_data(date, date)
+    if len(cal_dates) == 0:
+        print("Today is not a market day.")
+        return False, cal_dates
+    else:
+        print("Market is open")
+        return True, cal_dates
+
+def get_todays_market_times(market, debug_date = None):
     try:
-        if market == "US":
+        if market == Market.US:
             #zjistit vsechny podminky - mozna loopovat - podminky jsou vlevo
             if debug_date is not None:
                 nowNY = debug_date
@@ -28,17 +37,20 @@ def get_todays_market_times(market = "US", debug_date = None):
                 nowNY = datetime.now().astimezone(zoneNY)
             nowNY_date = nowNY.date()
             #is market open - nyni pouze US
-            cal_dates = fetch_calendar_data(nowNY_date, nowNY_date)
-
-            if len(cal_dates) == 0:
-                print("No Market Day today")
-                return -1, "Market Closed"
+            stat, calendar_dates = is_US_market_day(nowNY_date)
+            if stat:
             #zatim podpora pouze main session
-
             #pouze main session
-            market_open_datetime = zoneNY.localize(cal_dates[0].open)
-            market_close_datetime = zoneNY.localize(cal_dates[0].close)
-            return 0, (nowNY, market_open_datetime, market_close_datetime)
+                market_open_datetime = zoneNY.localize(calendar_dates[0].open)
+                market_close_datetime = zoneNY.localize(calendar_dates[0].close)
+                return 0, (nowNY, market_open_datetime, market_close_datetime)
+            else:
+                return -1, "Market is closed."
+        elif market == Market.CRYPTO:
+            now_market_datetime = datetime.now().astimezone(zoneUTC)
+            market_open_datetime = datetime.combine(datetime.now(), time.min)
+            matket_close_datetime = datetime.combine(datetime.now(), time.max)
+            return 0, (now_market_datetime, market_open_datetime, matket_close_datetime)
         else:
             return -1, "Market not supported"
     except Exception as e:

v2realbot/static/index.html

@@ -347,6 +347,7 @@
                                 <th>testlist_id</th>
                                 <th>Running</th>
                                 <th>RunnerId</th>
+                                <th>Market</th>
                             </tr>
                         </thead>
                         <tbody></tbody>
@@ -1149,7 +1150,7 @@
         <script src="/static/js/config.js?v=1.04"></script>
         <!-- tady zacina polska docasna lokalizace -->
         <!-- <script type="text/javascript" src="https://unpkg.com/lightweight-charts/dist/lightweight-charts.standalone.production.js"></script> -->
-        <script type="text/javascript" src="/static/js/libs/lightweightcharts/lightweight-charts.standalone.production410.js"></script>
+        <script type="text/javascript" src="/static/js/libs/lightweightcharts/lightweight-charts.standalone.production413.js"></script>
         <script src="/static/js/dynamicbuttons.js?v=1.05"></script>
     
        

v2realbot/static/js/tables/runmanager/init.js

@@ -45,7 +45,8 @@ function initialize_runmanagerRecords() {
                         {data: 'valid_to', visible: true},
                         {data: 'testlist_id', visible: true},
                         {data: 'strat_running', visible: true},
-                        {data: 'runner_id', visible: true},                  
+                        {data: 'runner_id', visible: true},
+                        {data: 'market', visible: true},                  
                     ],
             paging: true,
             processing: true,

v2realbot/static/js/utils/utils.js

@@ -371,9 +371,10 @@ function initialize_chart() {
     }
 
     chart = LightweightCharts.createChart(document.getElementById('chart'), chartOptions);
-    chart.applyOptions({ timeScale: { visible: true, timeVisible: true, secondsVisible: true }, crosshair: {
+    chart.applyOptions({ timeScale: { visible: true, timeVisible: true, secondsVisible: true, minBarSpacing: 0.003}, crosshair: {
         mode: LightweightCharts.CrosshairMode.Normal, labelVisible: true
     }})
+    console.log("chart intiialized")
 }
 
 //mozna atributy last value visible