(feat) final refactor of strategies configs

quants_lab/strategy/directional_strategy_base.py

@@ -8,8 +8,7 @@ from quants_lab.labeling.triple_barrier_method import triple_barrier_method
 
 
 class DirectionalStrategyBase:
-
-    def __init__(self, config: BaseModel):
+    def __init__(self, config):
         self.config = config
 
     def get_data(self, start: Optional[str] = None, end: Optional[str] = None):

quants_lab/strategy/experiments/bollinger.py

@@ -15,32 +15,22 @@ class BollingerConf(BaseModel):
 
 
 class Bollinger(DirectionalStrategyBase):
-    def __init__(self, config: BollingerConf):
-        super().__init__(config)
-        self.exchange = config.exchange
-        self.trading_pair = config.trading_pair
-        self.interval = config.interval
-        self.bb_length = config.bb_length
-        self.bb_std = config.bb_std
-        self.bb_long_threshold = config.bb_long_threshold
-        self.bb_short_threshold = config.bb_short_threshold
-
     def get_raw_data(self):
         df = self.get_candles(
-            exchange=self.exchange,
-            trading_pair=self.trading_pair,
-            interval=self.interval,
+            exchange=self.config.exchange,
+            trading_pair=self.config.trading_pair,
+            interval=self.config.interval,
         )
         return df
 
     def preprocessing(self, df):
-        df.ta.bbands(length=self.bb_length, std=self.bb_std, append=True)
+        df.ta.bbands(length=self.config.bb_length, std=self.config.bb_std, append=True)
         return df
 
     def predict(self, df):
         df["side"] = 0
-        long_condition = df[f"BBP_{self.bb_length}_{self.bb_std}"] < self.bb_long_threshold
-        short_condition = df[f"BBP_{self.bb_length}_{self.bb_std}"] > self.bb_short_threshold
+        long_condition = df[f"BBP_{self.config.bb_length}_{self.config.bb_std}"] < self.config.bb_long_threshold
+        short_condition = df[f"BBP_{self.config.bb_length}_{self.config.bb_std}"] > self.config.bb_short_threshold
         df.loc[long_condition, "side"] = 1
         df.loc[short_condition, "side"] = -1
         return df

quants_lab/strategy/experiments/macd_bb.py

@@ -1,52 +1,43 @@
 import pandas_ta as ta
+from pydantic import BaseModel, Field
+
 from quants_lab.strategy.directional_strategy_base import DirectionalStrategyBase
 
-from quants_lab.utils import data_management
+
+class MACDBBConfig(BaseModel):
+    exchange: str = Field(default="binance_perpetual")
+    trading_pair: str = Field(default="ETH-USDT")
+    interval: str = Field(default="1h")
+    bb_length: int = Field(default=24, ge=2, le=1000)
+    bb_std: float = Field(default=2.0, ge=0.5, le=4.0)
+    bb_long_threshold: float = Field(default=0.0, ge=-3.0, le=0.5)
+    bb_short_threshold: float = Field(default=1.0, ge=0.5, le=3.0)
+    fast_macd: int = Field(default=21, ge=2, le=100)
+    slow_macd: int = Field(default=42, ge=fast_macd, le=1000)
+    signal_macd: int = Field(default=9, ge=2, le=100)
 
 
-class MACDBB(DirectionalStrategyBase):
-    def __init__(self,
-                 exchange="binance_perpetual",
-                 trading_pair="ETH-USDT",
-                 interval="1h",
-                 bb_length=24,
-                 bb_std=2.0,
-                 bb_long_threshold=0.05,
-                 bb_short_threshold=0.95,
-                 fast_macd=21,
-                 slow_macd=42,
-                 signal_macd=9):
-        self.exchange = exchange
-        self.trading_pair = trading_pair
-        self.interval = interval
-        self.bb_length = bb_length
-        self.bb_std = bb_std
-        self.bb_long_threshold = bb_long_threshold
-        self.bb_short_threshold = bb_short_threshold
-        self.fast_macd = fast_macd
-        self.slow_macd = slow_macd
-        self.signal_macd = signal_macd
-
+class MacdBollinger(DirectionalStrategyBase):
     def get_raw_data(self):
-        df = data_management.get_dataframe(
-            exchange=self.exchange,
-            trading_pair=self.trading_pair,
-            interval=self.interval,
+        df = self.get_candles(
+            exchange=self.config.exchange,
+            trading_pair=self.config.trading_pair,
+            interval=self.config.interval,
         )
         return df
 
     def preprocessing(self, df):
-        df.ta.bbands(length=self.bb_length, std=self.bb_std, append=True)
-        df.ta.macd(fast=self.fast_macd, slow=self.slow_macd, signal=self.signal_macd, append=True)
+        df.ta.bbands(length=self.config.bb_length, std=self.config.bb_std, append=True)
+        df.ta.macd(fast=self.config.fast_macd, slow=self.config.slow_macd, signal=self.config.signal_macd, append=True)
         return df
 
     def predict(self, df):
-        bbp = df[f"BBP_{self.bb_length}_{self.bb_std}"]
-        macdh = df[f"MACDh_{self.fast_macd}_{self.slow_macd}_{self.signal_macd}"]
-        macd = df[f"MACD_{self.fast_macd}_{self.slow_macd}_{self.signal_macd}"]
+        bbp = df[f"BBP_{self.config.bb_length}_{self.config.bb_std}"]
+        macdh = df[f"MACDh_{self.config.fast_macd}_{self.config.slow_macd}_{self.config.signal_macd}"]
+        macd = df[f"MACD_{self.config.fast_macd}_{self.config.slow_macd}_{self.config.signal_macd}"]
 
-        long_condition = (bbp < self.bb_long_threshold) & (macdh > 0) & (macd < 0)
-        short_condition = (bbp > self.bb_short_threshold) & (macdh < 0) & (macd > 0)
+        long_condition = (bbp < self.config.bb_long_threshold) & (macdh > 0) & (macd < 0)
+        short_condition = (bbp > self.config.bb_short_threshold) & (macdh < 0) & (macd > 0)
 
         df["side"] = 0
         df.loc[long_condition, "side"] = 1

quants_lab/strategy/experiments/stat_arb.py

@@ -16,26 +16,16 @@ class StatArbConfig(BaseModel):
 
 
 class StatArb(DirectionalStrategyBase):
-    def __init__(self, config: StatArbConfig):
-        super().__init__(config)
-        self.exchange = config.exchange
-        self.trading_pair = config.trading_pair
-        self.target_trading_pair = config.target_trading_pair
-        self.interval = config.interval
-        self.lookback = config.lookback
-        self.z_score_long = config.z_score_long
-        self.z_score_short = config.z_score_short
-
     def get_raw_data(self):
         df = self.get_candles(
-            exchange=self.exchange,
-            trading_pair=self.trading_pair,
-            interval=self.interval,
+            exchange=self.config.exchange,
+            trading_pair=self.config.trading_pair,
+            interval=self.config.interval,
         )
         df_target = self.get_candles(
-            exchange=self.exchange,
-            trading_pair=self.target_trading_pair,
-            interval=self.interval,
+            exchange=self.config.exchange,
+            trading_pair=self.config.target_trading_pair,
+            interval=self.config.interval,
         )
         df = pd.merge(df, df_target, on="timestamp", how='inner', suffixes=('', '_target'))
         return df
@@ -44,14 +34,14 @@ class StatArb(DirectionalStrategyBase):
         df["pct_change_original"] = df["close"].pct_change()
         df["pct_change_target"] = df["close_target"].pct_change()
         df["spread"] = df["pct_change_target"] - df["pct_change_original"]
-        df["cum_spread"] = df["spread"].rolling(self.lookback).sum()
-        df["z_score"] = ta.zscore(df["cum_spread"], length=self.lookback)
+        df["cum_spread"] = df["spread"].rolling(self.config.lookback).sum()
+        df["z_score"] = ta.zscore(df["cum_spread"], length=self.config.lookback)
         return df
 
     def predict(self, df):
         df["side"] = 0
-        short_condition = df["z_score"] < - self.z_score_short
-        long_condition = df["z_score"] > self.z_score_long
+        short_condition = df["z_score"] < - self.config.z_score_short
+        long_condition = df["z_score"] > self.config.z_score_long
         df.loc[long_condition, "side"] = 1
         df.loc[short_condition, "side"] = -1
         return df