classed indicators draft

v2realbot/strategy/base.py

@@ -793,6 +793,7 @@ class StrategyState:
         self.mode = None
         self.wait_for_fill = None
         self.today_market_close = None
+        self.classed_indicators = {}
     
     def release(self):
         #release large variables

v2realbot/strategyblocks/indicators/custom/classed.py

@@ -0,0 +1,60 @@
+from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
+from v2realbot.strategy.base import StrategyState
+from v2realbot.indicators.indicators import ema, natr, roc
+from v2realbot.strategyblocks.indicators.helpers import get_source_series, evaluate_directive_conditions
+from rich import print as printanyway
+from traceback import format_exc
+import numpy as np
+from collections import defaultdict
+import importlib
+#EXAMPLE of directives:
+# [stratvars.indicators.cusum]
+#     type = "custom"
+#     subtype = "classed"
+#     on_confirmed_only = true
+#     cp.source = "close" #source posilany explicitne do next, note that next has access to state
+# [stratvars.indicators.cusum.cp.init] #params is send to init
+#     threshold = 1000
+
+#ROZSIRENI DO BUDOUCNA:
+# [stratvars.indicators.cusum.cp.next] #params explciitely sent to next, note that next has access to state
+#    source = "close" #indicator
+
+#OBECNA trida pro statefull indicators - realized by class with the same name,  deriving from parent IndicatorBase class
+#todo v initu inicializovat state.classed_indicators a ve stopu uklidit - resetovat
+def classed(state, params, name):
+    funcName = "classed"
+    if params is None:
+        return -2, "params required"
+    
+    init_params = safe_get(params, "init", None) #napr sekce obcahuje threshold = 1222, ktere jdou kwargs do initu fce
+    #next_params = safe_get(params, "next", None)
+
+    source = safe_get(params, "source", None) #source, ktery jde do initu
+    source = get_source_series(state, source)
+    #lookback = int(value_or_indicator(state, lookback))
+
+    #class_next_params = safe_get(params, "class_next_params", None)
+    
+    try:
+        if name not in state.classed_indicators:
+            classname = name
+            class_module = importlib.import_module("v2realbot.strategyblocks.indicators.custom.classes."+classname)
+            indicatorClass = getattr(class_module, classname)
+            instance = indicatorClass(state=state, **init_params)
+            print("instance vytvorena", instance)
+            state.classed_indicators[name] = instance
+            state.ilog(lvl=1,e=f"IND CLASS {name} INITIALIZED", **params)
+
+        if source is not None:
+            val = state.classed_indicators[name].next(source[-1])
+        else:
+            val = state.classed_indicators[name].next()
+
+        state.ilog(lvl=1,e=f"IND CLASS {name} NEXT {val}", **params)
+        return 0, val
+
+    except Exception as e:
+        printanyway(str(e)+format_exc())
+        return -2, str(e)+format_exc()        
+

v2realbot/strategyblocks/indicators/custom/classes/CUSUM.py

@@ -0,0 +1,30 @@
+from v2realbot.strategyblocks.indicators.custom.classes.indicatorbase import IndicatorBase
+
+class CUSUM(IndicatorBase):
+    def __init__(self, state, threshold):
+        super().__init__(state)
+        self.threshold = threshold
+        self.cumulative_sum = 0
+        self.previous_price = None
+
+    def next(self, new_price):
+        if self.previous_price is None:
+            # First data point, no previous price to compare with
+            self.previous_price = new_price
+            return 0
+
+        # Calculate price change
+        price_change = new_price - self.previous_price
+        self.previous_price = new_price
+
+        # Update cumulative sum
+        self.cumulative_sum += price_change
+
+        if self.cumulative_sum > self.threshold:
+            self.cumulative_sum = 0  # Reset
+            return 1  # Buy signal
+        elif self.cumulative_sum < -self.threshold:
+            self.cumulative_sum = 0  # Reset
+            return -1  # Sell signal
+
+        return 0  # No signal

v2realbot/strategyblocks/indicators/custom/classes/__init__.py

@@ -0,0 +1,10 @@
+# import os
+# # import importlib
+# # from v2realbot.strategyblocks.indicators.custom.opengap import opengap
+
+# for filename in os.listdir("v2realbot/strategyblocks/indicators/custom"):
+#     if filename.endswith(".py") and filename != "__init__.py":
+#        # __import__(filename[:-3])
+#         __import__(f"v2realbot.strategyblocks.indicators.custom.{filename[:-3]}")
+#         #importlib.import_module()
+

v2realbot/strategyblocks/indicators/custom/classes/indicatorbase.py

@@ -0,0 +1,10 @@
+class IndicatorBase:
+    def __init__(self, state):
+        self.state = state
+        self.data = None
+        self.prev_data = None
+
+    def next(self, data):
+        #something
+        self.prev_data = data
+        return None  # No signal