Add modelling strategies

user_data/strategies/DonchianChannelStrategy.py

@@ -0,0 +1,144 @@
+# pragma pylint: disable=missing-docstring, invalid-name
+# flake8: noqa
+# isort: skip_file
+
+from datetime import datetime, timezone
+from pandas import DataFrame
+from typing import Optional
+
+from freqtrade.strategy import IStrategy, Trade, timeframe_to_minutes
+import talib.abstract as ta
+
+class DonchianChannelStrategy(IStrategy):
+    """
+    Donchian Channel Trend Following Strategy (adapted from Curtis Faith's The Way Of The Turtle)
+
+    Rules implemented:
+      - Entry: 20-period Donchian breakout.
+      - Trend filter: SMA(15) vs SMA(350)
+          * if SMA15 > SMA350 => allow longs
+          * if SMA15 < SMA350 => allow shorts (optional)
+      - Exit: 10-period Donchian exit OR time exit after 80 candles (eg: 80 days on 1d timeframe)
+
+    Note:
+      - Use timeframe = "1d" to match “20-day / 10-day / 80-day” wording.
+      - Donchian levels are shifted by 1 candle to avoid lookahead.
+    """
+
+    INTERFACE_VERSION = 3
+
+    can_short: bool = False
+
+    minimal_roi = {}
+    stoploss = -0.20
+    trailing_stop = False
+
+    timeframe = "1d"
+    process_only_new_candles = True
+
+    use_exit_signal = True
+    exit_profit_only = False
+    ignore_roi_if_entry_signal = False
+
+    startup_candle_count: int = 400  # SMA350 + channel windows
+
+    # Period parameters
+    entry_dc_period = 20
+    exit_dc_period = 10
+    fast_ma_period = 15
+    slow_ma_period = 350
+
+    # Time-exit in candles (80 days if timeframe == "1d")
+    time_exit_candles = 80
+
+    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe["ma_fast"] = ta.SMA(dataframe, timeperiod=self.fast_ma_period)
+        dataframe["ma_slow"] = ta.SMA(dataframe, timeperiod=self.slow_ma_period)
+
+        # Donchian channels - shift(1) prevents using the current candle in the threshold.
+        dataframe["dc_upper_entry"] = dataframe["high"].rolling(self.entry_dc_period).max().shift(1)
+        dataframe["dc_lower_entry"] = dataframe["low"].rolling(self.entry_dc_period).min().shift(1)
+
+        dataframe["dc_upper_exit"] = dataframe["high"].rolling(self.exit_dc_period).max().shift(1)
+        dataframe["dc_lower_exit"] = dataframe["low"].rolling(self.exit_dc_period).min().shift(1)
+
+        dataframe["trend_up"] = dataframe["ma_fast"] > dataframe["ma_slow"]
+        dataframe["trend_down"] = dataframe["ma_fast"] < dataframe["ma_slow"]
+
+        return dataframe
+
+    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        # Long breakout entry
+        dataframe.loc[
+            (
+                (dataframe["trend_up"]) &
+                (dataframe["close"] > dataframe["dc_upper_entry"]) &
+                (dataframe["volume"] > 0)
+            ),
+            "enter_long",
+        ] = 1
+
+        # Short breakout entry (only if enabled)
+        dataframe.loc[
+            (
+                (self.can_short) &
+                (dataframe["trend_down"]) &
+                (dataframe["close"] < dataframe["dc_lower_entry"]) &
+                (dataframe["volume"] > 0)
+            ),
+            "enter_short",
+        ] = 1
+
+        return dataframe
+
+    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        # Donchian exit for longs: break below 10-period low
+        dataframe.loc[
+            (
+                (dataframe["close"] < dataframe["dc_lower_exit"])
+            ),
+            "exit_long",
+        ] = 1
+
+        # Donchian exit for shorts: break above 10-period high
+        dataframe.loc[
+            (
+                (dataframe["close"] > dataframe["dc_upper_exit"])
+            ),
+            "exit_short",
+        ] = 1
+
+        return dataframe
+
+    def custom_exit(
+        self,
+        pair: str,
+        trade: Trade,
+        current_time: datetime,
+        current_rate: float,
+        current_profit: float,
+        **kwargs,
+    ) -> Optional[str]:
+        """
+        Time-based exit: close trade after N candles. (1d timeframe = N days)
+        Convert candle-count to minutes using timeframe_to_minutes(self.timeframe).
+        """
+        tf_min = timeframe_to_minutes(self.timeframe)
+        max_age_min = self.time_exit_candles * tf_min
+
+        # Ensure timezone-aware datetimes
+        open_dt = trade.open_date_utc # gives trade's open date
+        if open_dt.tzinfo is None:
+            open_dt = open_dt.replace(tzinfo=timezone.utc)
+
+        now_dt = current_time
+        if now_dt.tzinfo is None:
+            now_dt = now_dt.replace(tzinfo=timezone.utc)
+
+        age_minutes = (now_dt - open_dt).total_seconds() / 60.0
+
+        if age_minutes >= max_age_min:
+            return f"time_exit_{self.time_exit_candles}c"
+
+        return None
+

user_data/strategies/ExhaustionGapBullishStrategy.py

@@ -0,0 +1,126 @@
+# pragma pylint: disable=missing-docstring, invalid-name
+# flake8: noqa
+# isort: skip_file
+
+from datetime import datetime, timezone
+from pandas import DataFrame
+from typing import Optional
+
+from freqtrade.strategy import IStrategy, Trade, timeframe_to_minutes
+import talib.abstract as ta
+
+class ExhaustionGapBullishStrategy(IStrategy):
+    """
+    Bullish Exhaustion Gap Strategy for 5m crypto (24/7) 
+
+    Gap Idea:
+      - Current candle opens below the previous candle's low by X% (a fast drop / dislocation),
+        AND then closes bullish (close > open).
+    Entry:
+      - Enter long on the NEXT candle after the signal candle (more realistic).
+    Exit:
+      - Small mean-reversion exit: RSI gets hot OR price recovers above EMA.
+      - Optional: takeprofit + time stop via custom_exit.
+    """
+
+    INTERFACE_VERSION = 3
+
+    can_short: bool = False
+
+    timeframe = "5m"
+    process_only_new_candles = True
+
+    minimal_roi = {}
+    stoploss = -0.03
+    trailing_stop = False
+
+    use_exit_signal = True
+    exit_profit_only = False
+    ignore_roi_if_entry_signal = False
+
+    startup_candle_count: int = 200
+
+    gap_open_below_prev_low_pct = 0.004   # 0.4%
+    min_bull_body_pct = 0.001             # 0.1%
+
+    ema_period = 50
+    rsi_period = 14
+
+    takeprofit_pct = 0.01         
+    max_hold_candles = 24                # 24 * 5m[timeframe] = 120 minutes
+
+    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe["ema"] = ta.EMA(dataframe, timeperiod=self.ema_period)
+        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=self.rsi_period)
+
+        dataframe["low_prev"] = dataframe["low"].shift(1)
+        dataframe["close_prev"] = dataframe["close"].shift(1)
+
+        dataframe["open_below_prev_low_pct"] = (dataframe["low_prev"] - dataframe["open"]) / dataframe["low_prev"]
+
+        # Bullish candle strength
+        dataframe["bull_body_pct"] = (dataframe["close"] - dataframe["open"]) / dataframe["open"]
+
+        dataframe["bull_gap_signal"] = (
+            (dataframe["open_below_prev_low_pct"] > self.gap_open_below_prev_low_pct) &
+            (dataframe["bull_body_pct"] > self.min_bull_body_pct) &
+            (dataframe["close"] > dataframe["open"]) &
+            (dataframe["volume"] > 0)
+        ).astype("int")
+
+        return dataframe
+
+    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        # Enter next candle after signal
+        signal_prev = dataframe["bull_gap_signal"].shift(1).fillna(0).astype("int")
+
+        dataframe.loc[
+            (signal_prev == 1),
+            "enter_long",
+        ] = 1
+
+        return dataframe
+
+    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        # Exit when price recovers above EMA or RSI gets hot
+        dataframe.loc[
+            (
+                (dataframe["close"] > dataframe["ema"]) |
+                (dataframe["rsi"] > 65)
+            ),
+            "exit_long",
+        ] = 1
+
+        return dataframe
+
+    def custom_exit(
+        self,
+        pair: str,
+        trade: Trade,
+        current_time: datetime,
+        current_rate: float,
+        current_profit: float,
+        **kwargs,
+    ) -> Optional[str]:
+        # Fixed TP
+        if current_profit >= self.takeprofit_pct:
+            return f"tp_{int(self.takeprofit_pct * 100)}pct"
+
+        # Time stop
+        tf_min = timeframe_to_minutes(self.timeframe)
+        max_age_min = self.max_hold_candles * tf_min
+
+        open_dt = trade.open_date_utc
+        if open_dt.tzinfo is None:
+            open_dt = open_dt.replace(tzinfo=timezone.utc)
+
+        now_dt = current_time
+        if now_dt.tzinfo is None:
+            now_dt = now_dt.replace(tzinfo=timezone.utc)
+
+        age_minutes = (now_dt - open_dt).total_seconds() / 60.0
+        if age_minutes >= max_age_min:
+            return f"time_stop_{self.max_hold_candles}c"
+
+        return None
+

user_data/strategies/PairsTradingStrategy.py

@@ -0,0 +1,84 @@
+# pragma pylint: disable=missing-docstring, invalid-name
+# flake8: noqa
+# isort: skip_file
+
+from pandas import DataFrame
+
+from freqtrade.strategy import IStrategy
+import statsmodels.api as sm
+
+class PairsTradingStrategy(IStrategy):
+    """
+    Strategy adapted from Gatev, Goetzmann & Rouwenhorst (2006)
+    Implements cointegration-based pairs trading with z-score thresholds.
+    """
+
+    INTERFACE_VERSION = 3
+    can_short: bool = False
+
+    minimal_roi = {}
+    stoploss = -0.05
+    trailing_stop = False
+
+    timeframe = "4h"
+    process_only_new_candles = True
+
+    use_exit_signal = True
+    exit_profit_only = False
+    ignore_roi_if_entry_signal = False
+
+    startup_candle_count: int = 200
+
+    # Define the two pairs to trade
+    pair_a = "BTC/USDT"
+    pair_b = "ETH/USDT"
+
+    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        """
+        Compute hedge ratio, spread, and z-score between two assets.
+        """
+        df_a = self.dp.get_pair_dataframe(self.pair_a, self.timeframe)
+        df_b = self.dp.get_pair_dataframe(self.pair_b, self.timeframe)
+
+        # Hedge ratio via linear regression
+        model = sm.OLS(df_a['close'], sm.add_constant(df_b['close']))
+        result = model.fit()
+        beta = result.params[1]
+
+        # Spread
+        dataframe['spread'] = df_a['close'] - beta * df_b['close']
+
+        # Z-score
+        mean = dataframe['spread'].rolling(window=50).mean()
+        std = dataframe['spread'].rolling(window=50).std()
+        dataframe['zscore'] = (dataframe['spread'] - mean) / std
+
+        return dataframe
+
+    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        """
+        Entry rules: trade when spread diverges significantly.
+        """
+        dataframe.loc[
+            (dataframe['zscore'] > 2),
+            'enter_short'
+        ] = 1
+
+        dataframe.loc[
+            (dataframe['zscore'] < -2),
+            'enter_long'
+        ] = 1
+
+        return dataframe
+
+    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        """
+        Exit rules: close when spread mean-reverts.
+        """
+        dataframe.loc[
+            (dataframe['zscore'].abs() < 0.5),
+            ['exit_long', 'exit_short']
+        ] = 1
+
+        return dataframe
+