# 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 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 CryptoFutures1m5mStrategyV6 --strategy-path ./user_data/Chan/strategies --timerange=20250101- class CryptoFutures1m5mStrategyV6(IStrategy): """ SOL/USDT 合约策略 - V6 强化做空版 基于V5优化： 1. 做空条件更严格 - 需要更强的趋势确认 2. 做空ATR过滤更严格 - 避免震荡市 3. 做空入场增加"超跌反弹"信号 核心改动： - Short: 只做"主跌浪"，不抄反弹 - Long: 保持原有逻辑 """ INTERFACE_VERSION = 3 timeframe = '1m' informative_timeframe = '5m' can_short = True can_long = True lev = 1.0 stoploss = -0.030 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: 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) 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) ) # 做空趋势 - 更严格！需要更强的ADX informative['trend_bear_5m'] = ( (informative['ema12'] < informative['ema26']) & (informative['ema26'] < informative['ema50']) & (informative['ema12_slope'] < -0.08) & # 更陡的斜率 (informative['adx_5m'] > 28) & # 更强的趋势确认 (informative['adx_5m'] < 50) & (informative['close'] < informative['ema12']) & (informative['rsi_5m'] < 45) & # 更低RSI (informative['rsi_5m'] > 25) ) # 大趋势过滤 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'] & informative['bear_market'] # 必须确认熊市 ) # ==================== ATR过滤 - 做空更严格 ==================== # 做多ATR - 保持宽松 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.8) # 更强成交量确认 ) # EMA死叉 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'] # 做空入场 - 更严格的熊市条件 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'] > 28) & # 更低RSI (dataframe['top_divergence'] | dataframe['ema_cross_down'] | dataframe['bear_pullback']) & (dataframe['volume'] > 0), 'enter_short' ] = 1 # 做多入场 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 > 6 and current_profit < -0.004: return 'time_stop_short_6h' if trade_duration > 12 and current_profit < 0: return 'time_stop_short_12h' if trade_duration > 20: return 'time_stop_short_20h' # 做多 - 保持宽松 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