diff --git a/quants_lab/controllers/trend_follower_v1.py b/quants_lab/controllers/trend_follower_v1.py new file mode 100644 index 0000000..ce32758 --- /dev/null +++ b/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 TrendFollowerConfig(DirectionalTradingControllerConfigBase): + strategy_name: str = "trend_follower" + 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=30, le=200) + bb_std: float = Field(default=2.0, ge=1.0, le=3.0) + bb_threshold: float = Field(default=0.2, ge=0.7, le=0.3) + + +class TrendFollower(DirectionalTradingControllerBase): + + def __init__(self, config: TrendFollowerConfig): + 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}"]