Feat/simplify strategies

quants_lab/controllers/bollinger.py → quants_lab/controllers/bollinger_v1.py

@@ -1,3 +1,4 @@
+import time
 from typing import Optional
 
 import pandas as pd
@@ -9,18 +10,17 @@
 from pydantic import Field
 
 
-class BollingerConf(DirectionalTradingControllerConfigBase):
- strategy_name = "bollinger"
- bb_length: int = Field(default=100, 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)
- std_span: Optional[int] = Field(default=100, ge=100, le=400)
+class BollingerV1Conf(DirectionalTradingControllerConfigBase):
+ strategy_name = "bollinger_v1"
+ bb_length: int = Field(default=100, ge=100, le=400)
+ bb_std: float = Field(default=2.0, ge=2.0, le=3.0)
+ bb_long_threshold: float = Field(default=0.0, ge=-1.0, le=0.2)
+ bb_short_threshold: float = Field(default=1.0, ge=0.8, le=2.0)
 
 
-class Bollinger(DirectionalTradingControllerBase):
+class BollingerV1(DirectionalTradingControllerBase):
 
- def __init__(self, config: BollingerConf):
+ def __init__(self, config: BollingerV1Conf):
  super().__init__(config)
  self.config = config
 
@@ -54,8 +54,7 @@ def get_processed_data(self) -> pd.DataFrame:
  df["signal"] = 0
  df.loc[long_condition, "signal"] = 1
  df.loc[short_condition, "signal"] = -1
-
- # Optional: Generate spread multiplier
- if self.config.std_span:
- df["target"] = df["close"].rolling(self.config.std_span).std() / df["close"]
  return df
+
+ def extra_columns_to_show(self):
+ return [f"BBP_{self.config.bb_length}_{self.config.bb_std}"]

quants_lab/controllers/macd_bb_v1.py

@@ -0,0 +1,69 @@
+import time
+from typing import Optional
+
+import pandas as pd
+from pydantic import Field
+
+from hummingbot.smart_components.executors.position_executor.position_executor import PositionExecutor
+from hummingbot.smart_components.strategy_frameworks.data_types import OrderLevel
+from hummingbot.smart_components.strategy_frameworks.directional_trading.directional_trading_controller_base import (
+ DirectionalTradingControllerBase,
+ DirectionalTradingControllerConfigBase,
+)
+
+
+class MACDBBV1Config(DirectionalTradingControllerConfigBase):
+ strategy_name: str = "macd_bb_v1"
+ bb_length: int = Field(default=24, ge=100, le=200)
+ bb_std: float = Field(default=2.0, ge=2.0, le=3.0)
+ bb_long_threshold: float = Field(default=0.0, ge=-1.0, le=0.2)
+ bb_short_threshold: float = Field(default=1.0, ge=0.8, le=2.0)
+ macd_fast: int = Field(default=21, ge=12, le=60)
+ macd_slow: int = Field(default=42, ge=26, le=200)
+ macd_signal: int = Field(default=9, ge=8, le=20)
+
+
+class MACDBBV1(DirectionalTradingControllerBase):
+
+ def __init__(self, config: MACDBBV1Config):
+ super().__init__(config)
+ self.config = config
+
+ def early_stop_condition(self, executor: PositionExecutor, order_level: OrderLevel) -> bool:
+ """
+ If an executor has an active position, should we close it based on a condition.
+ """
+ return False
+
+ def cooldown_condition(self, executor: PositionExecutor, order_level: OrderLevel) -> bool:
+ """
+ After finishing an order, the executor will be in cooldown for a certain amount of time.
+ This prevents the executor from creating a new order immediately after finishing one and execute a lot
+ of orders in a short period of time from the same side.
+ """
+ if executor.close_timestamp and executor.close_timestamp + order_level.cooldown_time > time.time():
+ return True
+ return False
+
+ def get_processed_data(self) -> pd.DataFrame:
+ df = self.candles[0].candles_df
+
+ # Add indicators
+ df.ta.bbands(length=self.config.bb_length, std=self.config.bb_std, append=True)
+ df.ta.macd(fast=self.config.macd_fast, slow=self.config.macd_slow, signal=self.config.macd_signal, append=True)
+ bbp = df[f"BBP_{self.config.bb_length}_{self.config.bb_std}"]
+ macdh = df[f"MACDh_{self.config.macd_fast}_{self.config.macd_slow}_{self.config.macd_signal}"]
+ macd = df[f"MACD_{self.config.macd_fast}_{self.config.macd_slow}_{self.config.macd_signal}"]
+
+ # Generate signal
+ 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["signal"] = 0
+ df.loc[long_condition, "signal"] = 1
+ df.loc[short_condition, "signal"] = -1
+ return df
+
+ def extra_columns_to_show(self):
+ return [f"BBP_{self.config.bb_length}_{self.config.bb_std}",
+ f"MACDh_{self.config.macd_fast}_{self.config.macd_slow}_{self.config.macd_signal}",
+ f"MACD_{self.config.macd_fast}_{self.config.macd_slow}_{self.config.macd_signal}"]

quants_lab/controllers/trend_follower_v1.py

@@ -0,0 +1,66 @@
+import time
+from typing import Optional
+
+import pandas as pd
+from pydantic import Field
+
+from hummingbot.smart_components.executors.position_executor.position_executor import PositionExecutor
+from hummingbot.smart_components.strategy_frameworks.data_types import OrderLevel
+from hummingbot.smart_components.strategy_frameworks.directional_trading.directional_trading_controller_base import (
+ DirectionalTradingControllerBase,
+ DirectionalTradingControllerConfigBase,
+)
+
+
+class TrendFollowerV1Config(DirectionalTradingControllerConfigBase):
+ strategy_name: str = "trend_follower_v1"
+ sma_fast: int = Field(default=20, ge=10, le=150)
+ sma_slow: int = Field(default=100, ge=50, le=400)
+ bb_length: int = Field(default=100, ge=100, le=200)
+ bb_std: float = Field(default=2.0, ge=2.0, le=3.0)
+ bb_threshold: float = Field(default=0.2, ge=0.1, le=0.5)
+
+
+class TrendFollowerV1(DirectionalTradingControllerBase):
+
+ def __init__(self, config: TrendFollowerV1Config):
+ super().__init__(config)
+ self.config = config
+
+ def early_stop_condition(self, executor: PositionExecutor, order_level: OrderLevel) -> bool:
+ # If an executor has an active position, should we close it based on a condition. This feature is not available
+ # for the backtesting yet
+ return False
+
+ def cooldown_condition(self, executor: PositionExecutor, order_level: OrderLevel) -> bool:
+ # After finishing an order, the executor will be in cooldown for a certain amount of time.
+ # This prevents the executor from creating a new order immediately after finishing one and execute a lot
+ # of orders in a short period of time from the same side.
+ if executor.close_timestamp and executor.close_timestamp + order_level.cooldown_time > time.time():
+ return True
+ return False
+
+ def get_processed_data(self) -> pd.DataFrame:
+ df = self.candles[0].candles_df
+ df.ta.sma(length=self.config.sma_fast, append=True)
+ df.ta.sma(length=self.config.sma_slow, append=True)
+ df.ta.bbands(length=self.config.bb_length, std=2.0, append=True)
+
+
+ # Generate long and short conditions
+ bbp = df[f"BBP_{self.config.bb_length}_2.0"]
+ inside_bounds_condition = (bbp < 0.5 + self.config.bb_threshold) & (bbp > 0.5 - self.config.bb_threshold)
+
+ long_cond = (df[f'SMA_{self.config.sma_fast}'] > df[f'SMA_{self.config.sma_slow}'])
+ short_cond = (df[f'SMA_{self.config.sma_fast}'] < df[f'SMA_{self.config.sma_slow}'])
+
+ # Choose side
+ df['signal'] = 0
+ df.loc[long_cond & inside_bounds_condition, 'signal'] = 1
+ df.loc[short_cond & inside_bounds_condition, 'signal'] = -1
+ return df
+
+ def extra_columns_to_show(self):
+ return [f"BBP_{self.config.bb_length}_{self.config.bb_std}",
+ f"SMA_{self.config.sma_fast}",
+ f"SMA_{self.config.sma_slow}"]

utils/file_templates.py

@@ -66,7 +66,7 @@ def get_optuna_suggest_str(field_name: str, properties: Dict):
  if field_name == "candles_config":
  return f"""{field_name}=[
  CandlesConfig(connector=exchange, trading_pair=trading_pair, 
- interval="3m", max_records=1000000) # Max number of candles for the real-time bot,
+ interval="1h", max_records=1000000)
  ]"""
  if field_name == "strategy_name":
  return f"{field_name}='{properties.get('default', '_')}'"
@@ -111,8 +111,8 @@ def strategy_optimization_template(strategy_info: dict):
 def objective(trial):
  try:
  # General configuration for the backtesting
- exchange = trial.suggest_categorical('exchange', ['binance_perpetual', ])
- trading_pair = trial.suggest_categorical('trading_pair', ['BTC-USDT', ])
+ exchange = "binance_perpetual"
+ trading_pair = "BTC-USDT"
  start = "2023-01-01"
  end = "2023-08-01"
  initial_portfolio_usd = 1000.0