# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement from freqtrade.strategy import IStrategy, merge_informative_pair from pandas import DataFrame import pandas as pd import talib.abstract as ta import numpy as np from datetime import datetime from typing import Optional from freqtrade.persistence import Trade import warnings # 抑制 pandas FutureWarning 关于 fillna 的隐式降级警告 warnings.filterwarnings('ignore', category=FutureWarning, message='.*Downcasting object dtype arrays.*') pd.set_option('future.no_silent_downcasting', True) # freqtrade backtesting -c ./user_data/Chan/config/Local_Test.json --strategy CryptoFutures1m5mStrategyV5 --strategy-path ./user_data/Chan/strategies --timerange=20250101- class CryptoFutures1m5mStrategyV5(IStrategy): """ SOL/USDT 合约策略 - 1分钟+5分钟双时间框架 V5 多空分离版 基于V3优化： 1. 多空止损完全分离 2. 保持V3的入场逻辑不变 多空参数分离： - 做多止损: -3.5% (更宽松) - 做空止损: -2.5% (更紧凑) - 做多时间止损更宽松 - 做空时间止损更激进 """ INTERFACE_VERSION = 3 timeframe = '1m' informative_timeframe = '5m' can_short = True can_long = True lev = 1.0 # 统一止损（兜底） stoploss = -0.035 # Trailing设置 trailing_stop = True trailing_stop_positive = 0.008 trailing_stop_positive_offset = 0.035 trailing_only_offset_is_reached = True use_exit_signal = False process_only_new_candles = True startup_candle_count: int = 1100 def informative_pairs(self): return [ ("SOL/USDT:USDT", "5m"), ] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # ==================== 5分钟指标 ==================== inf_tf = self.informative_timeframe informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=inf_tf) # EMA趋势 informative['ema12'] = ta.EMA(informative['close'], timeperiod=12) informative['ema26'] = ta.EMA(informative['close'], timeperiod=26) informative['ema50'] = ta.EMA(informative['close'], timeperiod=50) # EMA12斜率 informative['ema12_slope'] = (informative['ema12'] - informative['ema12'].shift(3)) / informative['ema12'].shift(3) * 100 # MACD macd, macd_signal, macd_hist = ta.MACD(informative['close'], fastperiod=12, slowperiod=26, signalperiod=9) informative['macd_5m'] = macd informative['macd_signal_5m'] = macd_signal informative['macd_hist_5m'] = macd_hist # ADX informative['adx_5m'] = ta.ADX(informative['high'], informative['low'], informative['close'], timeperiod=14) # RSI informative['rsi_5m'] = ta.RSI(informative['close'], timeperiod=14) # ATR informative['atr_5m'] = ta.ATR(informative['high'], informative['low'], informative['close'], timeperiod=14) informative['atr_pct_5m'] = informative['atr_5m'] / informative['close'] * 100 informative['atr_pct_ma_5m'] = informative['atr_pct_5m'].rolling(window=100).mean() # EMA200 informative['ema200'] = ta.EMA(informative['close'], timeperiod=200) informative['ema200_dist_pct'] = (informative['close'] - informative['ema200']) / informative['ema200'] * 100 informative['ema200_slope'] = (informative['ema200'] - informative['ema200'].shift(20)) / informative['ema200'].shift(20) * 100 # ==================== 多空趋势 ==================== # 做多趋势 informative['trend_bull_5m'] = ( (informative['ema12'] > informative['ema26']) & (informative['ema26'] > informative['ema50']) & (informative['ema12_slope'] > 0.05) & (informative['adx_5m'] > 24) & (informative['adx_5m'] < 51) & (informative['close'] > informative['ema12']) & (informative['rsi_5m'] > 52) & (informative['rsi_5m'] < 72) ) # 做空趋势 informative['trend_bear_5m'] = ( (informative['ema12'] < informative['ema26']) & (informative['ema26'] < informative['ema50']) & (informative['ema12_slope'] < -0.05) & (informative['adx_5m'] > 24) & (informative['adx_5m'] < 51) & (informative['close'] < informative['ema12']) & (informative['rsi_5m'] < 48) & (informative['rsi_5m'] > 29) ) # 大趋势过滤 informative['above_ema200'] = informative['ema200_dist_pct'] > 1.0 informative['below_ema200'] = informative['ema200_dist_pct'] < -1.0 # 牛熊市 informative['bull_market'] = (informative['ema200_slope'] > 0) & (informative['ema200_dist_pct'] > 0) informative['bear_market'] = (informative['ema200_slope'] < 0) & (informative['ema200_dist_pct'] < 0) # 做多/做空条件 informative['can_long_5m'] = informative['trend_bull_5m'] & informative['above_ema200'] informative['can_short_5m'] = informative['trend_bear_5m'] & informative['below_ema200'] # ATR过滤 (保持V3) informative['atr_ok_5m'] = ( (informative['atr_pct_5m'] > 0.07) & (informative['atr_pct_5m'] < informative['atr_pct_ma_5m'] * 2.2) ) # 成交量 informative['volume_ma_5m'] = ta.SMA(informative['volume'], timeperiod=20) informative['volume_ok_5m'] = informative['volume'] > informative['volume_ma_5m'] * 0.75 # 合并 dataframe = merge_informative_pair(dataframe, informative, self.timeframe, inf_tf, ffill=True) # ==================== 1分钟指标 ==================== macd_1m, signal_1m, hist_1m = ta.MACD(dataframe['close'], fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd'] = macd_1m dataframe['macd_signal'] = signal_1m dataframe['macd_hist'] = hist_1m dataframe['ema9'] = ta.EMA(dataframe['close'], timeperiod=9) dataframe['ema21'] = ta.EMA(dataframe['close'], timeperiod=21) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['vol_ma20'] = ta.SMA(dataframe['volume'], timeperiod=20) dataframe['macd_slope'] = (dataframe['macd'] - dataframe['macd'].shift(3)) / 3 # ==================== 做空信号 ==================== dataframe['price_high_5'] = dataframe['high'].rolling(window=5).max() dataframe['macd_high_5'] = dataframe['macd'].rolling(window=5).max() dataframe['top_divergence'] = ( (dataframe['high'] >= dataframe['price_high_5'] * 0.999) & (dataframe['macd'] < dataframe['macd_high_5']) & (dataframe['macd_slope'] < 0) & (dataframe['macd'] < dataframe['macd_signal']) & (dataframe['volume'] > dataframe['vol_ma20'] * 0.6) ) dataframe['ema_cross_down'] = ( (dataframe['ema9'] < dataframe['ema21']) & (dataframe['ema9'].shift(1) >= dataframe['ema21'].shift(1)) & (dataframe['rsi'] < 55) & (dataframe['rsi'] > 35) & (dataframe['volume'] > dataframe['vol_ma20'] * 1.0) ) dataframe['is_bear_candle'] = (dataframe['close'] < dataframe['open']) & ((dataframe['open'] - dataframe['close']) / dataframe['open'] > 0.008) dataframe['bear_pullback'] = ( dataframe['is_bear_candle'].shift(2) & (dataframe['close'].shift(1) > dataframe['open'].shift(1)) & (dataframe['high'] < dataframe['high'].shift(2)) & (dataframe['close'] < dataframe['open']) & (dataframe['close'] < dataframe['ema9']) ) # ==================== 做多信号 ==================== dataframe['price_low_5'] = dataframe['low'].rolling(window=5).min() dataframe['macd_low_5'] = dataframe['macd'].rolling(window=5).min() dataframe['bottom_divergence'] = ( (dataframe['low'] <= dataframe['price_low_5'] * 1.001) & (dataframe['macd'] > dataframe['macd_low_5']) & (dataframe['macd_slope'] > 0) & (dataframe['macd'] > dataframe['macd_signal']) & (dataframe['volume'] > dataframe['vol_ma20'] * 0.6) ) dataframe['ema_cross_up'] = ( (dataframe['ema9'] > dataframe['ema21']) & (dataframe['ema9'].shift(1) <= dataframe['ema21'].shift(1)) & (dataframe['rsi'] > 45) & (dataframe['rsi'] < 70) & (dataframe['volume'] > dataframe['vol_ma20'] * 1.0) ) dataframe['is_bull_candle'] = (dataframe['close'] > dataframe['open']) & ((dataframe['close'] - dataframe['open']) / dataframe['open'] > 0.008) dataframe['bull_pullback'] = ( dataframe['is_bull_candle'].shift(2) & (dataframe['close'].shift(1) < dataframe['open'].shift(1)) & (dataframe['low'] > dataframe['low'].shift(2)) & (dataframe['close'] > dataframe['open']) & (dataframe['close'] > dataframe['ema9']) ) # 时间过滤 dataframe['hour_utc'] = dataframe['date'].dt.hour dataframe['is_bad_hour'] = dataframe['hour_utc'].isin([4, 5, 6, 7]) # 类型转换 bool_cols = ['can_long_5m_5m', 'can_short_5m_5m', 'trend_bull_5m_5m', 'trend_bear_5m_5m', 'atr_ok_5m_5m', 'above_ema200_5m', 'below_ema200_5m', 'bull_market_5m', 'bear_market_5m', 'volume_ok_5m_5m'] for col in bool_cols: if col in dataframe.columns: dataframe[col] = dataframe[col].astype(bool).fillna(False) num_cols = ['atr_pct_5m_5m', 'rsi_5m_5m', 'macd_hist_5m_5m', 'atr_pct_ma_5m_5m', 'ema200_dist_pct_5m', 'ema200_slope_5m'] for col in num_cols: if col in dataframe.columns: dataframe[col] = dataframe[col].astype(float).fillna(0.0) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: time_ok = ~dataframe['is_bad_hour'] atr_ok = dataframe['atr_ok_5m_5m'] volume_ok = dataframe['volume_ok_5m_5m'] # 做空入场 (完全保持V3) macd_bear_5m = dataframe['macd_hist_5m_5m'] < 0 macd_bear_1m = dataframe['macd_hist'] < 0 dataframe.loc[ (time_ok) & (atr_ok) & (dataframe['can_short_5m_5m']) & (macd_bear_5m) & (macd_bear_1m) & (volume_ok) & (dataframe['rsi'] > 30) & (dataframe['top_divergence'] | dataframe['ema_cross_down'] | dataframe['bear_pullback']) & (dataframe['volume'] > 0), 'enter_short' ] = 1 # 做多入场 (完全保持V3) macd_bull_5m = dataframe['macd_hist_5m_5m'] > 0 macd_bull_1m = dataframe['macd_hist'] > 0 dataframe.loc[ (time_ok) & (atr_ok) & (dataframe['can_long_5m_5m']) & (macd_bull_5m) & (macd_bull_1m) & (volume_ok) & (dataframe['rsi'] < 70) & (dataframe['rsi'] > 40) & (dataframe['bottom_divergence'] | dataframe['ema_cross_up'] | dataframe['bull_pullback']) & (dataframe['volume'] > 0), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'exit_long'] = 0 dataframe.loc[:, 'exit_short'] = 0 return dataframe def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> str | bool | None: """多空分离的时间止损""" trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # 做空时间止损 - 更激进 if trade.trade_direction == 'short': if trade_duration > 8 and current_profit < -0.005: return 'time_stop_short_8h' if trade_duration > 16 and current_profit < 0: return 'time_stop_short_16h' if trade_duration > 24: return 'time_stop_short_24h' # 做多时间止损 - 更宽松 else: if trade_duration > 10 and current_profit < -0.006: return 'time_stop_long_10h' if trade_duration > 20 and current_profit < 0: return 'time_stop_long_20h' if trade_duration > 30: return 'time_stop_long_30h' return None def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: hour_utc = current_time.utcnow().hour if current_time.tzinfo is None else current_time.hour if hour_utc in {4, 5, 6, 7}: return False return True def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: return self.lev