(feat) adapt templates to hummingbot library

utils/enum_encoder.py

@@ -1,53 +0,0 @@
-import json
-from decimal import Decimal
-from enum import Enum
-
-import yaml
-
-
-class EnumEncoderDecoder:
-
-    def __init__(self, *enum_classes):
-        self.enum_classes = {enum_class.__name__: enum_class for enum_class in enum_classes}
-
-    def recursive_encode(self, value):
-        if isinstance(value, dict):
-            return {key: self.recursive_encode(val) for key, val in value.items()}
-        elif isinstance(value, list):
-            return [self.recursive_encode(val) for val in value]
-        elif isinstance(value, Enum):
-            return {"__enum__": True, "class": type(value).__name__, "value": value.name}
-        elif isinstance(value, Decimal):
-            return {"__decimal__": True, "value": str(value)}
-        else:
-            return value
-
-    def recursive_decode(self, value):
-        if isinstance(value, dict):
-            if value.get("__enum__"):
-                enum_class = self.enum_classes.get(value['class'])
-                if enum_class:
-                    return enum_class[value["value"]]
-            elif value.get("__decimal__"):
-                return Decimal(value["value"])
-            else:
-                return {key: self.recursive_decode(val) for key, val in value.items()}
-        elif isinstance(value, list):
-            return [self.recursive_decode(val) for val in value]
-        else:
-            return value
-
-    def encode(self, d):
-        return json.dumps(self.recursive_encode(d))
-
-    def decode(self, s):
-        return self.recursive_decode(json.loads(s))
-
-
-    def yaml_dump(self, d, file_path):
-        with open(file_path, 'w') as file:
-            yaml.dump(self.recursive_encode(d), file)
-
-    def yaml_load(self, file_path):
-        with open(file_path, 'r') as file:
-            return self.recursive_decode(yaml.safe_load(file))

utils/file_templates.py

@@ -1,52 +1,63 @@
 from typing import Dict
 
 
-def directional_strategy_template(strategy_cls_name: str) -> str:
+def directional_trading_controller_template(strategy_cls_name: str) -> str:
     strategy_config_cls_name = f"{strategy_cls_name}Config"
     sma_config_text = "{self.config.sma_length}"
-    return f"""import pandas_ta as ta
-from pydantic import BaseModel, Field
+    return f"""import time
+from typing import Optional
 
-from quants_lab.strategy.directional_strategy_base import DirectionalStrategyBase
+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 {strategy_config_cls_name}(BaseModel):
-    name: str = "{strategy_cls_name.lower()}"
-    exchange: str = Field(default="binance_perpetual")
-    trading_pair: str = Field(default="ETH-USDT")
-    interval: str = Field(default="1h")
+class {strategy_config_cls_name}(DirectionalTradingControllerConfigBase):
+    strategy_name: str = "{strategy_cls_name.lower()}"
     sma_length: int = Field(default=20, ge=10, le=200)
     # ... Add more fields here
 
 
-class {strategy_cls_name}(DirectionalStrategyBase[{strategy_config_cls_name}]):
+class {strategy_cls_name}(DirectionalTradingControllerBase):
 
-    def get_raw_data(self):
-        # The method get candles will search for the data in the folder data/candles
-        # If the data is not there, you can use the candles downloader to get the data
-        df = self.get_candles(
-            exchange=self.config.exchange,
-            trading_pair=self.config.trading_pair,
-            interval=self.config.interval,
-        )
-        return df
+    def __init__(self, config: {strategy_config_cls_name}):
+        super().__init__(config)
+        self.config = config
 
-    def preprocessing(self, df):
+
+    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_length, append=True)
         # ... Add more indicators here
         # ... Check https://github.com/twopirllc/pandas-ta#indicators-by-category for more indicators
         # ... Use help(ta.indicator_name) to get more info
-        return df
 
-    def predict(self, df):
         # Generate long and short conditions
         long_cond = (df['close'] > df[f'SMA_{sma_config_text}'])
         short_cond = (df['close'] < df[f'SMA_{sma_config_text}'])
 
         # Choose side
-        df['side'] = 0
-        df.loc[long_cond, 'side'] = 1
-        df.loc[short_cond, 'side'] = -1
+        df['signal'] = 0
+        df.loc[long_cond, 'signal'] = 1
+        df.loc[short_cond, 'signal'] = -1
         return df
 """
 
@@ -70,7 +81,7 @@ def get_optuna_suggest_str(field_name: str, properties: Dict):
         "integer": "trial.suggest_int",
         "string": "trial.suggest_categorical",
     }
-    config_num = f"('{field_name}', {properties.get('minimum', '_')}, {properties.get('maximum', '_')})"
+    config_num = f"('{field_name}', {properties.get('minimum', '_')}, {properties.get('maximum', '_')}, step=0.01)"
     config_cat = f"('{field_name}', ['{properties.get('default', '_')}',])"
     optuna_trial_str = map_by_type[properties["type"]] + config_num if properties["type"] != "string" \
         else map_by_type[properties["type"]] + config_cat
@@ -92,10 +103,10 @@ from hummingbot.core.data_type.common import PositionMode, TradeType, OrderType
 from hummingbot.data_feed.candles_feed.candles_factory import CandlesConfig
 from hummingbot.smart_components.strategy_frameworks.data_types import TripleBarrierConf, OrderLevel
 from hummingbot.smart_components.strategy_frameworks.directional_trading import DirectionalTradingBacktestingEngine
+from hummingbot.smart_components.utils import ConfigEncoderDecoder
 from optuna import TrialPruned   
 
-from quants_lab.strategy.controllers.{strategy_module} import {strategy_cls.__name__}, {strategy_config.__name__}
-from utils.enum_encoder import EnumEncoderDecoder
+from quants_lab.controllers.{strategy_module} import {strategy_cls.__name__}, {strategy_config.__name__}
 
 
 def objective(trial):
@@ -137,7 +148,7 @@ def objective(trial):
                                                      start=start, end=end)
 
         strategy_analysis = backtesting_results["results"]
-        encoder_decoder = EnumEncoderDecoder(TradeType, OrderType, PositionMode)
+        encoder_decoder = ConfigEncoderDecoder(TradeType, OrderType, PositionMode)
 
         trial.set_user_attr("net_pnl_quote", strategy_analysis["net_pnl_quote"])
         trial.set_user_attr("net_pnl_pct", strategy_analysis["net_pnl"])