optimalizace BT runu - offline loader skrz queue

v2realbot/ENTRY_ClassicSL_v01.py

@@ -6,15 +6,15 @@ from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Orde
 from v2realbot.indicators.indicators import ema, natr, roc
 from v2realbot.indicators.oscillators import rsi
 from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus, TradeStoplossType
-from v2realbot.utils.utils import ltp, isrising, isfalling,trunc,AttributeDict, zoneNY, price2dec, print, safe_get, round2five, is_open_rush, is_close_rush, is_still, is_window_open, eval_cond_dict, Average, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff
+from v2realbot.utils.utils import ltp, isrising, isfalling,trunc,AttributeDict, zoneNY, price2dec, print, safe_get, round2five, is_open_rush, is_close_rush, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff
 from v2realbot.utils.directive_utils import get_conditions_from_configuration
 from v2realbot.common.model import SLHistory
 from datetime import datetime, timedelta
 from v2realbot.config import KW
 from uuid import uuid4
-import random
+#import random
 import json
-from numpy import inf
+import numpy as np
 #from icecream import install, ic
 #from rich import print
 from threading import Event
@@ -57,7 +57,6 @@ Hlavní loop:
     - if not exit - eval optimalizations
 
 """
-
 def next(data, state: StrategyState):
     print(10*"*","NEXT START",10*"*")
     # important vars state.avgp, state.positions, state.vars, data
@@ -808,10 +807,14 @@ def next(data, state: StrategyState):
                         #ten bude prumerem hodnot lookback_offset a to tak ze polovina offsetu z kazde strany
                         array_od = slope_lookback + int(lookback_offset/2)
                         array_do = slope_lookback - int(lookback_offset/2)
-                        lookbackprice_array = state.bars.vwap[-array_od:-array_do]
 
-                            #dame na porovnani jen prumer
-                        lookbackprice = round(sum(lookbackprice_array)/lookback_offset,3)
+                        #lookbackprice_array = state.bars.vwap[-array_od:-array_do]
+                        #lookbackprice = round(sum(lookbackprice_array)/lookback_offset,3)
+
+                        #jako optimalizace pouzijeme NUMPY
+                        lookbackprice = np.mean(state.bars.vwap[-array_od:-array_do])
+                        # Round the lookback price to 3 decimal places
+                        lookbackprice = round(lookbackprice, 3)
                             #lookbackprice = round((min(lookbackprice_array)+max(lookbackprice_array))/2,3)
                         # else:
                         #     #puvodni lookback a od te doby dozadu offset
@@ -836,10 +839,13 @@ def next(data, state: StrategyState):
                         cnt = len(state.bars.close)
                         if cnt>5:
                             sliced_to = int(cnt/5)
-                            lookbackprice= Average(state.bars.vwap[:sliced_to])
+                            
+                            lookbackprice = np.mean(state.bars.vwap[:sliced_to])
+                            #lookbackprice= Average(state.bars.vwap[:sliced_to])
                             lookbacktime = state.bars.time[int(sliced_to/2)]
                         else:
-                            lookbackprice = Average(state.bars.vwap)
+                            lookbackprice = np.mean(state.bars.vwap)
+                            #lookbackprice = Average(state.bars.vwap)
                             lookbacktime = state.bars.time[0]
                         
                         state.ilog(lvl=1,e=f"IND {name} slope - not enough data bereme left bod open", slope_lookback=slope_lookback, lookbackprice=lookbackprice)
@@ -882,7 +888,7 @@ def next(data, state: StrategyState):
         if len(state.vars.last_50_deltas) >=50:
             state.vars.last_50_deltas.pop(0)
         state.vars.last_50_deltas.append(last_update_delta)
-        avg_delta = Average(state.vars.last_50_deltas)
+        avg_delta = np.mean(state.vars.last_50_deltas)
 
         state.ilog(lvl=1,e=f"-----{data['index']}-{conf_bar}--delta:{last_update_delta}---AVGdelta:{avg_delta}", data=data)
 
@@ -1740,6 +1746,11 @@ def next(data, state: StrategyState):
             state.ilog(lvl=1,e=f"SIGNAL {signalname} - WINDOW CLOSED", msg=f"{window_open=} {window_close=} ")
             return False           
 
+        min_bar_index = safe_get(options, "min_bar_index",safe_get(state.vars, "min_bar_index",0))
+        if int(data["index"]) < int(min_bar_index):
+            state.ilog(lvl=1,e=f"MIN BAR INDEX {min_bar_index} waiting - TOO SOON", currindex=data["index"])
+            return False
+
         next_signal_offset = safe_get(options, "next_signal_offset_from_last_exit",safe_get(state.vars, "next_signal_offset_from_last_exit",0))
 
         if state.vars.last_exit_index is not None:

v2realbot/ENTRY_Vykladaci_RSI_MYSELL_type.py

@@ -14,6 +14,7 @@ from msgpack import packb, unpackb
 import asyncio
 import os
 from traceback import format_exc
+#from codetiming import Timer
 
 print(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 """"
@@ -90,6 +91,7 @@ minsize=100
 exthours=false
 """
 
+#@Timer(name="nextfce-timers")
 def next(data, state: StrategyState):
     print(10*"*","NEXT START",10*"*")
     #ic(state.avgp, state.positions)

v2realbot/controller/services.py

@@ -25,6 +25,7 @@ from traceback import format_exc
 from datetime import timedelta, time
 from threading import Lock
 from v2realbot.common.db import pool
+#from pyinstrument import Profiler
 #adding lock to ensure thread safety of TinyDB (in future will be migrated to proper db)
 lock = Lock()
 
@@ -284,7 +285,11 @@ def capsule(target: object, db: object, inter_batch_params: dict = None):
     #TODO zde odchytit pripadnou exceptionu a zapsat do history
     #cil aby padnuti jedne nezpusobilo pad enginu
     try:
+
+        # profiler = Profiler()
+        # profiler.start()
         target.start()
+
         print("Strategy instance stopped. Update runners")
         reason = "SHUTDOWN OK"
     except Exception as e:
@@ -294,6 +299,12 @@ def capsule(target: object, db: object, inter_batch_params: dict = None):
         print(reason)
         send_to_telegram(reason)
     finally:
+        # profiler.stop()
+        # now = datetime.now()
+        # results_file = "profiler"+now.strftime("%Y-%m-%d_%H-%M-%S")+".html"
+        # with open(results_file, "w", encoding="utf-8") as f_html:
+        #     f_html.write(profiler.output_html())
+
         # remove runners after thread is stopped and save results to stratin history
         for i in db.runners:
             if i.run_instance == target:
@@ -407,7 +418,7 @@ def run_stratin(id: UUID, runReq: RunRequest, synchronous: bool = False, inter_b
         if runReq.bt_to is None:
             runReq.bt_to = datetime.now().astimezone(zoneNY)
     
-    print("hodnota ID pred",id)
+    #print("hodnota ID pred",id)
     #volani funkce instantiate_strategy
     for i in db.stratins:
         if str(i.id) == str(id):
@@ -493,9 +504,9 @@ def run_stratin(id: UUID, runReq: RunRequest, synchronous: bool = False, inter_b
                         run_instance = instance,
                         run_stratvars_toml=i.stratvars_conf)
                 db.runners.append(runner)
-                print(db.runners)
-                print(i)
-                print(enumerate())
+                #print(db.runners)
+                #print(i)
+                #print(enumerate())
 
                 #pokud spoustime v batch módu, tak čekáme na výsledek a pak pouštíme další run
                 if synchronous:
@@ -584,6 +595,8 @@ def populate_metrics_output_directory(strat: StrategyInstance, inter_batch_param
         max_profit_time = None
         long_cnt = 0
         short_cnt = 0
+
+        if "prescribedTrades" in strat.state.vars:
             for trade in strat.state.vars.prescribedTrades:
                 if trade.profit_sum > max_profit:
                     max_profit = trade.profit_sum

v2realbot/loader/aggregator.py

@@ -107,9 +107,9 @@ class TradeAggregator:
         except KeyError:
             ltp.price[symbol]=data['p']
 
-
-        if float(data['p']) > float(ltp.price[symbol]) + (float(data['p'])/100*pct_off) or float(data['p']) < float(ltp.price[symbol])-(float(data['p'])/100*pct_off):
-            print("ZLO", data,ltp.price[symbol])
+        #DOCASNE VYPNUTO - VYMYSLET JINAK
+        #if float(data['p']) > float(ltp.price[symbol]) + (float(data['p'])/100*pct_off) or float(data['p']) < float(ltp.price[symbol])-(float(data['p'])/100*pct_off):
+            #print("ZLO", data,ltp.price[symbol])
             #nechavame zlo zatim projit
             ##return []
             # with open("cache/wrongtrades.txt", 'a') as fp:

v2realbot/loader/trade_offline_streamer.py

@@ -17,6 +17,8 @@ from msgpack import packb
 from pandas import to_datetime
 import pickle
 import os
+from rich import print
+import queue
 
 """
     Trade offline data streamer, based on Alpaca historical data.
@@ -55,6 +57,7 @@ class Trade_Offline_Streamer(Thread):
    # Override the run() function of Thread class
    #odebrano async
     def main(self):
+        trade_queue = queue.Queue()
         print(10*"*","Trade OFFLINE streamer STARTED", current_thread().name,10*"*")
         
         if not self.streams:
@@ -91,7 +94,10 @@ class Trade_Offline_Streamer(Thread):
         ##PREPSAT jednoduse tak, aby podporovalo jen jeden symbol
         #agregator2list bude mit vstup list
 
+        #datetime.fromtimestamp(data['updated']).astimezone(zoneNY))
         #REFACTOR STARTS HERE
+        #print(f"{self.time_from=} {self.time_to=}")
+              
         calendar_request = GetCalendarRequest(start=self.time_from,end=self.time_to)
         cal_dates = self.clientTrading.get_calendar(calendar_request)
         #ic(cal_dates)
@@ -101,15 +107,20 @@ class Trade_Offline_Streamer(Thread):
         #minimalni jednotka pro CACHE je 1 den - a to jen marketopen to marketclose (extended hours not supported yet)
         for day in cal_dates:
             print("Processing DAY", day.date)
-            print(day.date)
+            #print(day.date)
             print(day.open)
             print(day.close)
             #make it offset aware
-            day.open = day.open.replace(tzinfo=zoneNY)
+            day.open = zoneNY.localize(day.open)
+            #day.open.replace(tzinfo=zoneNY)
             #add 20 minutes of premarket
             #day.open = day.open - timedelta(minutes=20)
-            day.close = day.close.replace(tzinfo=zoneNY)
-                                     
+            day.close = zoneNY.localize(day.close)
+            #day.close = day.close.replace(tzinfo=zoneNY)
+            #print(day.open)
+            #print(day.close)  
+            #print("dayopentimestamp", day.open.timestamp())      
+            #print("dayclosetimestamp", day.close.timestamp())     
             ##pokud datum do je mensi day.open, tak tento den neresime
             if self.time_to < day.open:
                 print("time_to je pred zacatkem marketu. Vynechavame tento den.")
@@ -184,7 +195,14 @@ class Trade_Offline_Streamer(Thread):
 
                     #poustime i 20 minut premarketu pro presnejsi populaci slopu v prvnich minutech
                     # - timedelta(minutes=20)
-                    if self.time_from < to_datetime(t['t']) < self.time_to:
+                    #homogenizace timestampu s online streamem
+                    #tmp = to_datetime(t['t'], utc=True).timestamp()
+                    
+
+
+                    datum = to_datetime(t['t'], utc=True)
+
+                    if self.time_from < datum < self.time_to:
                         #poustime dal, jinak ne
                         if wait_for_q:
                             #cekame na Q nebo na O (nekterym dnum chybelo Q)
@@ -194,7 +212,10 @@ class Trade_Offline_Streamer(Thread):
                                 wait_for_q = False
                         
                         #homogenizace timestampu s online streamem
-                        t['t'] = Timestamp.from_unix(to_datetime(t['t']).timestamp())
+                        t['t'] = Timestamp.from_unix(datum.timestamp())
+                        #print(f"{t['t']}")
+                        #t['t'] = Timestamp.from_unix(to_datetime(t['t']).timestamp())
+                        #print(to_datetime(t['t']).timestamp())
                         
                         #print("PROGRESS ",cnt,"/",celkem)
                         #print(t)
@@ -204,23 +225,38 @@ class Trade_Offline_Streamer(Thread):
                             #print("zaznam",t)
                             #print("Ingest", s, "zaznam", t)
                             #await s.ingest_trade(packb(t))
-                            asyncio.run(s.ingest_trade(packb(t)))
+                            trade_queue.put((s,t))
+
+                            ##asyncio.run(s.ingest_trade(packb(t)))
                         cnt += 1
                     #protoze jsou serazene, tak prvni ktery je vetsi muze prerusit
-                    elif to_datetime(t['t']) > self.time_to:
-                        print("prerusujeme")
+                    elif datum > self.time_to:
+                        #print(f"{datum=}")
+                        #print(to_datetime(t['t']))
+                        #print(f"{self.time_to=}")
+                        #print("prerusujeme")
                         break
         #vsem streamum posleme last TODO: (tuto celou cast prepsat a zjednodusit)
         #po loadovani vsech dnu
         print("naloadovane vse posilame last")
         for s in self.to_run[symbpole[0]]:
             #zde bylo await
-            asyncio.run(s.ingest_trade(packb("last")))
+            trade_queue.put((s,"last"))
+            ##asyncio.run(s.ingest_trade(packb("last")))
             print("poslano last")
 
-        #loop = asyncio.get_running_loop()
-        print("stoping loop")
-        #loop.stop()
+        async def process_trade_queue(trade_queue):
+            while not trade_queue.empty():
+                #print("send trade")
+                s, trade = trade_queue.get()
+                await s.ingest_trade(packb(trade))
+
+        #spusteni asyncio run - tentokrat jednou, ktera spusti proces jez to z queue odesle
+        #nevyhoda reseni - kdyz je to pres vice dnu, tak se naloaduji vsechny dny do queue
+        #ale to mi u Classic zatim nevadi - poustim per days
+        #uvidim jak to ovlivn rychlost
+        asyncio.run(process_trade_queue(trade_queue))
+        print("skoncilo zpracovani ASYNCIO RUN TRADE QUEUE - zpracovany vsechny trady v agreagtorech")
         print(10*"*","Trade OFFLINE streamer STOPPED", current_thread().name,10*"*")
 
 

v2realbot/main.py

@@ -228,7 +228,7 @@ def _get_stratin(stratin_id) -> StrategyInstance:
 
 @app.put("/stratins/{stratin_id}/run", dependencies=[Depends(api_key_auth)], status_code=status.HTTP_200_OK)
 def _run_stratin(stratin_id: UUID, runReq: RunRequest):
-    print(runReq)
+    #print(runReq)
     if runReq.test_batch_id is not None:
         res, id = cs.run_batch_stratin(id=stratin_id, runReq=runReq)
     else:

v2realbot/strategy/base.py

@@ -24,7 +24,10 @@ from alpaca.trading.enums import TradeEvent, OrderStatus
 from threading import Event, current_thread
 import json
 from uuid import UUID
+from rich import print as printnow
+#from pyinstrument import Profiler
 
+#profiler = Profiler()
 # obecna Parent strategie podporující queues
 class Strategy:
     def __init__(self, name: str, symbol: str, next: callable, init: callable, account: Account, mode: str = Mode.PAPER, stratvars: AttributeDict = None, open_rush: int = 30, close_rush: int = 30, pe: Event = None, se: Event = None, runner_id: UUID = None, ilog_save: bool = False) -> None:
@@ -324,7 +327,10 @@ class Strategy:
             pass
             #self.state.ilog(e="Rush hour - skipping")
         else:
+            # Profile the function
+            #profiler.start()
             self.next(item, self.state)
+            #profiler.stop()
             self.after_iteration(item)
             
     ##run strategy live
@@ -353,6 +359,7 @@ class Strategy:
             try:
                 #block 5s, after that check signals
                 item = self.q1.get(timeout=HEARTBEAT_TIMEOUT)
+                #printnow(current_thread().name, "Items waiting in queue:", self.q1.qsize())
             except queue.Empty:
                 #check signals
                 if self.se.is_set():
@@ -386,6 +393,11 @@ class Strategy:
         tlog(f"FINISHED")
         print(40*"*",self.mode, "STRATEGY ", self.name,"STOPPING",40*"*")
 
+        # now = datetime.now()
+        # results_file = "profiler"+now.strftime("%Y-%m-%d_%H-%M-%S")+".html"
+        # with open(results_file, "w", encoding="utf-8") as f_html:
+        #     f_html.write(profiler.output_html())
+
         self.stop()
 
         if self.mode == Mode.BT:

v2realbot/utils/utils.py

@@ -3,7 +3,7 @@ import math
 from queue import Queue
 from datetime import datetime, timezone, time, timedelta, date
 import pytz
-from dateutil import tz
+#from dateutil import tz
 from rich import print as richprint
 import decimal
 from v2realbot.enums.enums import RecordType, Mode, StartBarAlign
@@ -291,7 +291,8 @@ class Store:
 
 qu = Queue()
 
-zoneNY = tz.gettz('America/New_York')
+#zoneNY = tz.gettz('America/New_York')
+zoneNY = pytz.timezone('US/Eastern')
 
 def print(*args, **kwargs):
     if QUIET_MODE:
@@ -300,13 +301,45 @@ def print(*args, **kwargs):
         ####ic(*args, **kwargs)
         richprint(*args, **kwargs)
 
+#optimized by BARD
 def price2dec(price: float, decimals: int = 2) -> float:
+  """Rounds a price to a specified number of decimal places, but only if the
+  price has more than that number of decimals.
+
+  Args:
+    price: The price to round.
+    decimals: The number of decimals to round to.
+
+  Returns:
+    A rounded price, or the original price if the price has less than or equal
+    to the specified number of decimals.
+  """
+
+  if price.is_integer():
+    return price
+
+  # Calculate the number of decimal places in the price.
+  num_decimals = int(math.floor(math.log10(abs(price - math.floor(price)))))
+
+  # If the price has more than the specified number of decimals, round it.
+  if num_decimals > decimals:
+    return round(price, decimals)
+  else:
+    return price
+
+def price2dec_old(price: float, decimals: int = 2) -> float:
     """
     pousti maximalne 2 decimals
     pokud je trojmistne a vic pak zakrouhli na 2, jinak necha
     """
     return round(price,decimals) if count_decimals(price) > decimals else price
 
+def count_decimals(number: float) -> int:
+    """
+    Count the number of decimals in a given float: 1.4335 -> 4 or 3 -> 0
+    """
+    return abs(decimal.Decimal(str(number)).as_tuple().exponent)
+
 def round2five(price: float):
     """
     zatim jen na 3 mista -pripadne predelat na dynamicky
@@ -315,12 +348,6 @@ def round2five(price: float):
     """ 
     return (round(price*100*2)/2)/100
 
-def count_decimals(number: float) -> int:
-    """
-    Count the number of decimals in a given float: 1.4335 -> 4 or 3 -> 0
-    """
-    return abs(decimal.Decimal(str(number)).as_tuple().exponent)
-
 def p(var, n = None):
     if n: print(n, f'{var = }')
     else: print(f'{var = }')
@@ -337,8 +364,35 @@ def is_open_rush(dt: datetime, mins: int = 30):
     rushtime = (datetime.combine(date.today(), business_hours["from"]) + timedelta(minutes=mins)).time()
     return business_hours["from"] <= dt.time() < rushtime
 
-#TODO market time pro dany den si dotahnout z Alpaca
+#optimalized by BARD
 def is_window_open(dt: datetime, start: int = 0, end: int = 390):
+    """"
+    Returns true if time (start in minutes and end in minutes) is in working window
+    """
+    # Check if start and end are within bounds early to avoid unnecessary computations.
+    if start < 0 or start > 389 or end < 0 or end > 389:
+        return False
+
+    # Convert the datetime object to the New York time zone.
+    dt = dt.astimezone(zoneNY)
+
+    # Get the business hours start and end times.
+    business_hours_start = time(hour=9, minute=30)
+    business_hours_end = time(hour=16, minute=0)
+
+    # Check if the datetime is within business hours.
+    if not business_hours_start <= dt.time() <= business_hours_end:
+        return False
+
+    # Calculate the start and end times of the working window.
+    working_window_start = (datetime.combine(date.today(), business_hours_start) + timedelta(minutes=start)).time()
+    working_window_end = (datetime.combine(date.today(), business_hours_start) + timedelta(minutes=end)).time()
+
+    # Check if the datetime is within the working window.
+    return working_window_start <= dt.time() <= working_window_end
+#puvodni neoptimalizovana verze
+#TODO market time pro dany den si dotahnout z Alpaca
+def is_window_open_old(dt: datetime, start: int = 0, end: int = 390):
     """"
     Returns true if time (start in minutes and end in minutes) is in working window
     """