from typing import List
from hummingbot.data_feed.candles_feed.data_types import CandlesConfig
from hummingbot.strategy_v2.controllers.directional_trading_controller_base import (
DirectionalTradingControllerBase,
DirectionalTradingControllerConfigBase,
)
class BollingerV1ControllerConfig(DirectionalTradingControllerConfigBase):
controller_name: str = "bollinger_v1"
interval: str = "3m"
bb_length: int = 100
bb_std: float = 2.0
bb_long_threshold: float = 0.0
bb_short_threshold: float = 1.0
class BollingerV1Controller(DirectionalTradingControllerBase):
def __init__(self, config: BollingerV1ControllerConfig, *args, **kwargs):
self.config = config
self.max_records = config.bb_length
super().__init__(config, *args, **kwargs)
def get_candles_config(self) -> List[CandlesConfig]:
return [CandlesConfig(
connector=self.config.candles_connector,
trading_pair=self.config.candles_trading_pair,
interval=self.config.interval,
max_records=self.max_records,
)]
async def update_processed_data(self):
df = self.market_data_provider.get_candles_df(
connector_name=self.config.candles_connector,
trading_pair=self.config.candles_trading_pair,
interval=self.config.interval,
max_records=self.max_records,
)
df.ta.bbands(length=self.config.bb_length, std=self.config.bb_std, append=True)
bbp = df[f"BBP_{self.config.bb_length}_{self.config.bb_std}"]
df["signal"] = 0
df.loc[bbp < self.config.bb_long_threshold, "signal"] = 1
df.loc[bbp > self.config.bb_short_threshold, "signal"] = -1
self.processed_data["signal"] = df["signal"].iloc[-1]
self.processed_data["features"] = df
