diff --git a/freqtrade b/freqtrade index ab093ff..2c521aa 160000 --- a/freqtrade +++ b/freqtrade @@ -1 +1 @@ -Subproject commit ab093ff0e1af445f0b8491ea1168c46e1a51b2c0 +Subproject commit 2c521aa002ac905e14032e3604c1a2636e983884 diff --git a/user_data/strategies/DonchianChannelStrategy.py b/user_data/strategies/DonchianChannelStrategy.py new file mode 100644 index 0000000..95c6abe --- /dev/null +++ b/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 + \ No newline at end of file diff --git a/user_data/strategies/ExhaustionGapBullishStrategy.py b/user_data/strategies/ExhaustionGapBullishStrategy.py new file mode 100644 index 0000000..ba1aaa9 --- /dev/null +++ b/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 + \ No newline at end of file diff --git a/user_data/strategies/PairsTradingStrategy.py b/user_data/strategies/PairsTradingStrategy.py new file mode 100644 index 0000000..d0b4f9b --- /dev/null +++ b/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 + diff --git a/user_data/strategies/VVR_VWAP_Strategy.py b/user_data/strategies/VVR_VWAP_Strategy.py new file mode 100644 index 0000000..c9767cd --- /dev/null +++ b/user_data/strategies/VVR_VWAP_Strategy.py @@ -0,0 +1,119 @@ +# pragma pylint: disable=missing-docstring, invalid-name +# flake8: noqa +# isort: skip_file + +from pandas import DataFrame +from freqtrade.strategy import IStrategy +import talib.abstract as ta +from technical import qtpylib + +class VVR_VWAP_Strategy(IStrategy): + """ + Strategy adapted from paper: https://arxiv.org/pdf/2508.01419 + """ + # Strategy interface version - allow new iterations of the strategy interface. + # Check the documentation or the Sample strategy to get the latest version. + INTERFACE_VERSION = 3 + + # Can this strategy go short? + can_short: bool = False + + # Minimal ROI designed for the strategy. + # This attribute will be overridden if the config file contains "minimal_roi". + minimal_roi = {} + + # Optimal stoploss designed for the strategy. + # This attribute will be overridden if the config file contains "stoploss". + stoploss = -0.02 + + # Trailing stoploss + trailing_stop = False + # trailing_only_offset_is_reached = False + # trailing_stop_positive = 0.01 + # trailing_stop_positive_offset = 0.0 # Disabled / not configured + + # Optimal timeframe for the strategy. + timeframe = "5m" + + # Run "populate_indicators()" only for new candle. + process_only_new_candles = True + + # These values can be overridden in the config. + use_exit_signal = True + exit_profit_only = False + ignore_roi_if_entry_signal = False + + # Number of candles the strategy requires before producing valid signals + startup_candle_count: int = 200 + + def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + # 1. EMAs for Trend + dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=12) + dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=26) + + # 2. RSI + dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) + + # 3. MACD + macd = ta.MACD(dataframe) + dataframe["macd"] = macd["macd"] + dataframe["macdsignal"] = macd["macdsignal"] + + # 4. VVR (Volume-to-Volatility Ratio) + epsilon = 1e-6 + dataframe["price_range"] = dataframe["high"] - dataframe["low"] + # Use rolling mean of VVR immediately for comparison + dataframe["vvr"] = dataframe["volume"] / (dataframe["price_range"] + epsilon) + dataframe["vvr_mean"] = dataframe["vvr"].rolling(window=50).mean() + + # 5. Rolling VWAP (approx 1 day on 5m candles = 288 candles) + rolling_window = 288 + vwap_num = (dataframe['volume'] * (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3).rolling(window=rolling_window).sum() + vwap_denom = dataframe['volume'].rolling(window=rolling_window).sum() + dataframe['vwap'] = vwap_num / vwap_denom + + return dataframe + + def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + dataframe.loc[ + ( + # 1. TREND: Long term trend is UP + (dataframe["ema_fast"] > dataframe["ema_slow"]) & + + # 2. MOMENTUM: RSI is not Overbought (Safe to buy) + (dataframe["rsi"] < 55) & + (dataframe["rsi"] > 30) & + + # 3. MACD: Momentum is recovering or positive + (dataframe["macd"] > dataframe["macdsignal"]) & + + # 4. VALUATION: Price is below or near the daily VWAP + (dataframe["close"] < dataframe["vwap"]) & + + # 5. LIQUIDITY: High efficiency (Volume > Volatility) + (dataframe["vvr"] > dataframe["vvr_mean"]) & + + (dataframe["volume"] > 0) + ), + "enter_long", + ] = 1 + + return dataframe + + def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + dataframe.loc[ + ( + # Exit if RSI gets too hot + (dataframe["rsi"] > 75) | + + # OR MACD crosses down + (qtpylib.crossed_below(dataframe["macd"], dataframe["macdsignal"])) | + + # OR Price extends too far above VWAP + (dataframe["close"] > dataframe["vwap"] * 1.03) + ), + "exit_long", + ] = 1 + + return dataframe + \ No newline at end of file