(feat) add directional controllers

bots/controllers/directional_trading/bollinger_v1.py

@@ -0,0 +1,101 @@
+from typing import List
+
+import pandas_ta as ta  # noqa: F401
+from pydantic import Field, validator
+
+from hummingbot.client.config.config_data_types import ClientFieldData
+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 = "bollinger_v1"
+    candles_config: List[CandlesConfig] = []
+    candles_connector: str = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt_on_new=True,
+            prompt=lambda mi: "Enter the connector for the candles data, leave empty to use the same exchange as the connector: ", )
+    )
+    candles_trading_pair: str = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt_on_new=True,
+            prompt=lambda mi: "Enter the trading pair for the candles data, leave empty to use the same trading pair as the connector: ", )
+    )
+    interval: str = Field(
+        default="3m",
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the candle interval (e.g., 1m, 5m, 1h, 1d): ",
+            prompt_on_new=False))
+    bb_length: int = Field(
+        default=100,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands length: ",
+            prompt_on_new=True))
+    bb_std: float = Field(
+        default=2.0,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands standard deviation: ",
+            prompt_on_new=False))
+    bb_long_threshold: float = Field(
+        default=0.0,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands long threshold: ",
+            prompt_on_new=True))
+    bb_short_threshold: float = Field(
+        default=1.0,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands short threshold: ",
+            prompt_on_new=True))
+
+    @validator("candles_connector", pre=True, always=True)
+    def set_candles_connector(cls, v, values):
+        if v is None or v == "":
+            return values.get("connector_name")
+        return v
+
+    @validator("candles_trading_pair", pre=True, always=True)
+    def set_candles_trading_pair(cls, v, values):
+        if v is None or v == "":
+            return values.get("trading_pair")
+        return v
+
+
+class BollingerV1Controller(DirectionalTradingControllerBase):
+    def __init__(self, config: BollingerV1ControllerConfig, *args, **kwargs):
+        self.config = config
+        self.max_records = self.config.bb_length
+        if len(self.config.candles_config) == 0:
+            self.config.candles_config = [CandlesConfig(
+                connector=config.candles_connector,
+                trading_pair=config.candles_trading_pair,
+                interval=config.interval,
+                max_records=self.max_records
+            )]
+        super().__init__(config, *args, **kwargs)
+
+    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)
+        # Add indicators
+        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}"]
+
+        # Generate signal
+        long_condition = bbp < self.config.bb_long_threshold
+        short_condition = bbp > self.config.bb_short_threshold
+
+        # Generate signal
+        df["signal"] = 0
+        df.loc[long_condition, "signal"] = 1
+        df.loc[short_condition, "signal"] = -1
+
+        # Update processed data
+        self.processed_data["signal"] = df["signal"].iloc[-1]
+        self.processed_data["features"] = df

bots/controllers/directional_trading/dman_v3.py

@@ -0,0 +1,222 @@
+import time
+from decimal import Decimal
+from typing import List, Optional, Tuple
+
+import pandas_ta as ta  # noqa: F401
+from pydantic import Field, validator
+
+from hummingbot.client.config.config_data_types import ClientFieldData
+from hummingbot.core.data_type.common import TradeType
+from hummingbot.data_feed.candles_feed.data_types import CandlesConfig
+from hummingbot.strategy_v2.controllers.directional_trading_controller_base import (
+    DirectionalTradingControllerBase,
+    DirectionalTradingControllerConfigBase,
+)
+from hummingbot.strategy_v2.executors.dca_executor.data_types import DCAExecutorConfig, DCAMode
+from hummingbot.strategy_v2.executors.position_executor.data_types import TrailingStop
+
+
+class DManV3ControllerConfig(DirectionalTradingControllerConfigBase):
+    controller_name: str = "dman_v3"
+    candles_config: List[CandlesConfig] = []
+    candles_connector: str = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt_on_new=True,
+            prompt=lambda mi: "Enter the connector for the candles data, leave empty to use the same exchange as the connector: ",)
+    )
+    candles_trading_pair: str = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt_on_new=True,
+            prompt=lambda mi: "Enter the trading pair for the candles data, leave empty to use the same trading pair as the connector: ",)
+    )
+    interval: str = Field(
+        default="30m",
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the candle interval (e.g., 1m, 5m, 1h, 1d): ",
+            prompt_on_new=True))
+    bb_length: int = Field(
+        default=100,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands length: ",
+            prompt_on_new=True))
+    bb_std: float = Field(
+        default=2.0,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands standard deviation: ",
+            prompt_on_new=False))
+    bb_long_threshold: float = Field(
+        default=0.0,
+        client_data=ClientFieldData(
+            is_updatable=True,
+            prompt=lambda mi: "Enter the Bollinger Bands long threshold: ",
+            prompt_on_new=True))
+    bb_short_threshold: float = Field(
+        default=1.0,
+        client_data=ClientFieldData(
+            is_updatable=True,
+            prompt=lambda mi: "Enter the Bollinger Bands short threshold: ",
+            prompt_on_new=True))
+    dca_spreads: List[Decimal] = Field(
+        default="0.001,0.018,0.15,0.25",
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the spreads for each DCA level (comma-separated) if dynamic_spread=True this value "
+                              "will multiply the Bollinger Bands width, e.g. if the Bollinger Bands width is 0.1 (10%)"
+                              "and the spread is 0.2, the distance of the order to the current price will be 0.02 (2%) ",
+            prompt_on_new=True))
+    dca_amounts_pct: List[Decimal] = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the amounts for each DCA level (as a percentage of the total balance, "
+                              "comma-separated). Don't worry about the final sum, it will be normalized. ",
+            prompt_on_new=True))
+    dynamic_order_spread: bool = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Do you want to make the spread dynamic? (Yes/No) ",
+            prompt_on_new=True))
+    dynamic_target: bool = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Do you want to make the target dynamic? (Yes/No) ",
+            prompt_on_new=True))
+
+    activation_bounds: Optional[List[Decimal]] = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the activation bounds for the orders "
+                              "(e.g., 0.01 activates the next order when the price is closer than 1%): ",
+            prompt_on_new=True))
+
+    @validator("activation_bounds", pre=True, always=True)
+    def parse_activation_bounds(cls, v):
+        if isinstance(v, str):
+            if v == "":
+                return None
+            return [Decimal(val) for val in v.split(",")]
+        if isinstance(v, list):
+            return [Decimal(val) for val in v]
+        return v
+
+    @validator('dca_spreads', pre=True, always=True)
+    def validate_spreads(cls, v):
+        if isinstance(v, str):
+            return [Decimal(val) for val in v.split(",")]
+        return v
+
+    @validator('dca_amounts_pct', pre=True, always=True)
+    def validate_amounts(cls, v, values):
+        spreads = values.get("dca_spreads")
+        if isinstance(v, str):
+            if v == "":
+                return [Decimal('1.0') / len(spreads) for _ in spreads]
+            amounts = [Decimal(val) for val in v.split(",")]
+            if len(amounts) != len(spreads):
+                raise ValueError("Amounts and spreads must have the same length")
+            return amounts
+        if v is None:
+            return [Decimal('1.0') / len(spreads) for _ in spreads]
+        return v
+
+    def get_spreads_and_amounts_in_quote(self, trade_type: TradeType, total_amount_quote: Decimal) -> Tuple[List[Decimal], List[Decimal]]:
+        amounts_pct = self.dca_amounts_pct
+        if amounts_pct is None:
+            # Equally distribute if amounts_pct is not set
+            spreads = self.dca_spreads
+            normalized_amounts_pct = [Decimal('1.0') / len(spreads) for _ in spreads]
+        else:
+            if trade_type == TradeType.BUY:
+                normalized_amounts_pct = [amt_pct / sum(amounts_pct) for amt_pct in amounts_pct]
+            else:  # TradeType.SELL
+                normalized_amounts_pct = [amt_pct / sum(amounts_pct) for amt_pct in amounts_pct]
+
+        return self.dca_spreads, [amt_pct * total_amount_quote for amt_pct in normalized_amounts_pct]
+
+    @validator("candles_connector", pre=True, always=True)
+    def set_candles_connector(cls, v, values):
+        if v is None or v == "":
+            return values.get("connector_name")
+        return v
+
+    @validator("candles_trading_pair", pre=True, always=True)
+    def set_candles_trading_pair(cls, v, values):
+        if v is None or v == "":
+            return values.get("trading_pair")
+        return v
+
+
+class DManV3Controller(DirectionalTradingControllerBase):
+    """
+    Mean reversion strategy with Grid execution making use of Bollinger Bands indicator to make spreads dynamic
+    and shift the mid-price.
+    """
+    def __init__(self, config: DManV3ControllerConfig, *args, **kwargs):
+        self.config = config
+        self.max_records = config.bb_length
+        if len(self.config.candles_config) == 0:
+            self.config.candles_config = [CandlesConfig(
+                connector=config.candles_connector,
+                trading_pair=config.candles_trading_pair,
+                interval=config.interval,
+                max_records=self.max_records
+            )]
+        super().__init__(config, *args, **kwargs)
+
+    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)
+        # Add indicators
+        df.ta.bbands(length=self.config.bb_length, std=self.config.bb_std, append=True)
+
+        # Generate signal
+        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
+
+        # Generate signal
+        df["signal"] = 0
+        df.loc[long_condition, "signal"] = 1
+        df.loc[short_condition, "signal"] = -1
+
+        # Update processed data
+        self.processed_data["signal"] = df["signal"].iloc[-1]
+        self.processed_data["features"] = df
+
+    def get_spread_multiplier(self) -> Decimal:
+        if self.config.dynamic_order_spread:
+            df = self.processed_data["features"]
+            bb_width = df[f"BBB_{self.config.bb_length}_{self.config.bb_std}"].iloc[-1]
+            return Decimal(bb_width / 200)
+        else:
+            return Decimal("1.0")
+
+    def get_executor_config(self, trade_type: TradeType, price: Decimal, amount: Decimal) -> DCAExecutorConfig:
+        spread, amounts_quote = self.config.get_spreads_and_amounts_in_quote(trade_type, amount * price)
+        spread_multiplier = self.get_spread_multiplier()
+        if trade_type == TradeType.BUY:
+            prices = [price * (1 - spread * spread_multiplier) for spread in spread]
+        else:
+            prices = [price * (1 + spread * spread_multiplier) for spread in spread]
+        if self.config.dynamic_target:
+            stop_loss = self.config.stop_loss * spread_multiplier
+            trailing_stop = TrailingStop(activation_price=self.config.trailing_stop.activation_price * spread_multiplier,
+                                         trailing_delta=self.config.trailing_stop.trailing_delta * spread_multiplier)
+        else:
+            stop_loss = self.config.stop_loss
+            trailing_stop = self.config.trailing_stop
+        return DCAExecutorConfig(
+            timestamp=time.time(),
+            connector_name=self.config.connector_name,
+            trading_pair=self.config.trading_pair,
+            side=trade_type,
+            mode=DCAMode.MAKER,
+            prices=prices,
+            amounts_quote=amounts_quote,
+            time_limit=self.config.time_limit,
+            stop_loss=stop_loss,
+            trailing_stop=trailing_stop,
+            leverage=self.config.leverage,
+            activation_bounds=self.config.activation_bounds,
+        )

bots/controllers/directional_trading/macd_bb_v1.py

@@ -0,0 +1,120 @@
+from typing import List
+
+import pandas_ta as ta  # noqa: F401
+from pydantic import Field, validator
+
+from hummingbot.client.config.config_data_types import ClientFieldData
+from hummingbot.data_feed.candles_feed.data_types import CandlesConfig
+from hummingbot.strategy_v2.controllers.directional_trading_controller_base import (
+    DirectionalTradingControllerBase,
+    DirectionalTradingControllerConfigBase,
+)
+
+
+class MACDBBV1ControllerConfig(DirectionalTradingControllerConfigBase):
+    controller_name = "macd_bb_v1"
+    candles_config: List[CandlesConfig] = []
+    candles_connector: str = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt_on_new=True,
+            prompt=lambda mi: "Enter the connector for the candles data, leave empty to use the same exchange as the connector: ", )
+    )
+    candles_trading_pair: str = Field(
+        default=None,
+        client_data=ClientFieldData(
+            prompt_on_new=True,
+            prompt=lambda mi: "Enter the trading pair for the candles data, leave empty to use the same trading pair as the connector: ", )
+    )
+    interval: str = Field(
+        default="3m",
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the candle interval (e.g., 1m, 5m, 1h, 1d): ",
+            prompt_on_new=False))
+    bb_length: int = Field(
+        default=100,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands length: ",
+            prompt_on_new=True))
+    bb_std: float = Field(
+        default=2.0,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands standard deviation: ",
+            prompt_on_new=False))
+    bb_long_threshold: float = Field(
+        default=0.0,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands long threshold: ",
+            prompt_on_new=True))
+    bb_short_threshold: float = Field(
+        default=1.0,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the Bollinger Bands short threshold: ",
+            prompt_on_new=True))
+    macd_fast: int = Field(
+        default=21,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the MACD fast period: ",
+            prompt_on_new=True))
+    macd_slow: int = Field(
+        default=42,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the MACD slow period: ",
+            prompt_on_new=True))
+    macd_signal: int = Field(
+        default=9,
+        client_data=ClientFieldData(
+            prompt=lambda mi: "Enter the MACD signal period: ",
+            prompt_on_new=True))
+
+    @validator("candles_connector", pre=True, always=True)
+    def set_candles_connector(cls, v, values):
+        if v is None or v == "":
+            return values.get("connector_name")
+        return v
+
+    @validator("candles_trading_pair", pre=True, always=True)
+    def set_candles_trading_pair(cls, v, values):
+        if v is None or v == "":
+            return values.get("trading_pair")
+        return v
+
+
+class MACDBBV1Controller(DirectionalTradingControllerBase):
+
+    def __init__(self, config: MACDBBV1ControllerConfig, *args, **kwargs):
+        self.config = config
+        self.max_records = max(config.macd_slow, config.macd_fast, config.macd_signal, config.bb_length)
+        if len(self.config.candles_config) == 0:
+            self.config.candles_config = [CandlesConfig(
+                connector=config.candles_connector,
+                trading_pair=config.candles_trading_pair,
+                interval=config.interval,
+                max_records=self.max_records
+            )]
+        super().__init__(config, *args, **kwargs)
+
+    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)
+        # 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
+
+        # Update processed data
+        self.processed_data["signal"] = df["signal"].iloc[-1]
+        self.processed_data["features"] = df

bots/controllers/directional_trading/supertrend_v1.py

@@ -0,0 +1,70 @@
+from typing import List, Optional
+
+import pandas_ta as ta  # noqa: F401
+from pydantic import Field, validator
+
+from hummingbot.client.config.config_data_types import ClientFieldData
+from hummingbot.data_feed.candles_feed.data_types import CandlesConfig
+from hummingbot.strategy_v2.controllers.directional_trading_controller_base import (
+    DirectionalTradingControllerBase,
+    DirectionalTradingControllerConfigBase,
+)
+
+
+class SuperTrendConfig(DirectionalTradingControllerConfigBase):
+    controller_name: str = "supertrend_v1"
+    candles_config: List[CandlesConfig] = []
+    candles_connector: Optional[str] = Field(default=None, client_data=ClientFieldData(prompt_on_new=True, prompt=lambda mi: "Enter the connector for the candles data, leave empty to use the same exchange as the connector: ", ))
+    candles_trading_pair: Optional[str] = Field(default=None, client_data=ClientFieldData(prompt_on_new=True, prompt=lambda mi: "Enter the trading pair for the candles data, leave empty to use the same trading pair as the connector: ", ))
+    interval: str = Field(default="3m", client_data=ClientFieldData(prompt=lambda mi: "Enter the candle interval (e.g., 1m, 5m, 1h, 1d): ", prompt_on_new=False))
+    length: int = Field(default=20, client_data=ClientFieldData(prompt=lambda mi: "Enter the supertrend length: ", prompt_on_new=True))
+    multiplier: float = Field(default=4.0, client_data=ClientFieldData(prompt=lambda mi: "Enter the supertrend multiplier: ", prompt_on_new=True))
+    percentage_threshold: float = Field(default=0.01, client_data=ClientFieldData(prompt=lambda mi: "Enter the percentage threshold: ", prompt_on_new=True))
+
+    @validator("candles_connector", pre=True, always=True)
+    def set_candles_connector(cls, v, values):
+        if v is None or v == "":
+            return values.get("connector_name")
+        return v
+
+    @validator("candles_trading_pair", pre=True, always=True)
+    def set_candles_trading_pair(cls, v, values):
+        if v is None or v == "":
+            return values.get("trading_pair")
+        return v
+
+
+class SuperTrend(DirectionalTradingControllerBase):
+    def __init__(self, config: SuperTrendConfig, *args, **kwargs):
+        self.config = config
+        self.max_records = config.length + 10
+        if len(self.config.candles_config) == 0:
+            self.config.candles_config = [CandlesConfig(
+                connector=config.candles_connector,
+                trading_pair=config.candles_trading_pair,
+                interval=config.interval,
+                max_records=self.max_records
+            )]
+        super().__init__(config, *args, **kwargs)
+
+    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)
+        # Add indicators
+        df.ta.supertrend(length=self.config.length, multiplier=self.config.multiplier, append=True)
+        df["percentage_distance"] = abs(df["close"] - df[f"SUPERT_{self.config.length}_{self.config.multiplier}"]) / df["close"]
+
+        # Generate long and short conditions
+        long_condition = (df[f"SUPERTd_{self.config.length}_{self.config.multiplier}"] == 1) & (df["percentage_distance"] < self.config.percentage_threshold)
+        short_condition = (df[f"SUPERTd_{self.config.length}_{self.config.multiplier}"] == -1) & (df["percentage_distance"] < self.config.percentage_threshold)
+
+        # Choose side
+        df['signal'] = 0
+        df.loc[long_condition, 'signal'] = 1
+        df.loc[short_condition, 'signal'] = -1
+
+        # Update processed data
+        self.processed_data["signal"] = df["signal"].iloc[-1]
+        self.processed_data["features"] = df