# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement from freqtrade.strategy import IStrategy, merge_informative_pair, IntParameter, DecimalParameter, BooleanParameter 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 hyperopt -c ./user_data/Chan/config/Local_Test.json --strategy CryptoFutures1m5mStrategyV2Hyperopt --strategy-path ./user_data/Chan/strategies --timerange=20260101- --epochs 200 -j 4 --space buy # freqtrade backtesting -c ./user_data/Chan/config/Local_Test.json --strategy CryptoFutures1m5mStrategyV2Hyperopt --strategy-path ./user_data/Chan/strategies --timerange=20260101- class CryptoFutures1m5mStrategyV2Hyperopt(IStrategy): """ SOL/USDT 合约策略 - 1分钟+5分钟双时间框架 V2 Hyperopt优化版 (Short Only) 基于V2优化版添加Hyperopt参数： 1. ATR波动率过滤参数 2. 时间止损参数 3. 趋势确认参数（ADX, RSI, EMA200距离） """ INTERFACE_VERSION = 3 timeframe = '1m' informative_timeframe = '5m' can_short = True lev = 1.0 # 硬止损 stoploss = -0.025 # 追踪止盈 - 固定值 trailing_stop = True trailing_stop_positive = 0.008 trailing_stop_positive_offset = 0.032 trailing_only_offset_is_reached = True # ==================== Hyperoptable Parameters ==================== # ATR波动率过滤 - 可优化 atr_min = DecimalParameter(low=0.03, high=0.15, default=0.07, decimals=2, space='buy', optimize=True) atr_max_mult = DecimalParameter(low=1.5, high=3.5, default=2.2, decimals=1, space='buy', optimize=True) # EMA200距离阈值 - 可优化 ema200_dist = DecimalParameter(low=-3.0, high=-0.5, default=-1.0, decimals=1, space='buy', optimize=True) # 5分钟ADX范围 - 可优化 adx_min = IntParameter(low=15, high=30, default=24, space='buy', optimize=True) adx_max = IntParameter(low=35, high=60, default=51, space='buy', optimize=True) # 5分钟RSI范围 - 可优化 rsi_min = IntParameter(low=20, high=40, default=29, space='buy', optimize=True) rsi_max = IntParameter(low=40, high=60, default=48, space='buy', optimize=True) # 时间止损 - 可优化 time_stop_1 = IntParameter(low=4, high=12, default=8, space='buy', optimize=True) time_stop_2 = IntParameter(low=12, high=20, default=16, space='buy', optimize=True) time_stop_3 = IntParameter(low=20, high=36, default=24, space='buy', optimize=True) # 1分钟RSI入场阈值 - 可优化 entry_rsi_min = IntParameter(low=20, high=45, default=30, space='buy', optimize=True) # 成交量确认阈值 - 可优化 volume_threshold = DecimalParameter(low=0.5, high=1.5, default=0.75, decimals=2, space='buy', optimize=True) # 完全禁用 exit_signal 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斜率（3根K线变化率） 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（5分钟） informative['rsi_5m'] = ta.RSI(informative['close'], timeperiod=14) # ATR（5分钟） informative['atr_5m'] = ta.ATR(informative['high'], informative['low'], informative['close'], timeperiod=14) informative['atr_pct_5m'] = informative['atr_5m'] / informative['close'] * 100 # ATR 长期均值 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 # EMA200斜率 informative['ema200_slope'] = (informative['ema200'] - informative['ema200'].shift(20)) / informative['ema200'].shift(20) * 100 # 大趋势过滤（Short Only）- 使用hyperopt参数 informative['below_ema200'] = informative['ema200_dist_pct'] < self.ema200_dist.value # 牛市暂停 informative['bull_pause'] = ( (informative['ema200_slope'] > 0) & (informative['ema200_dist_pct'] > 0) ) # 5分钟趋势判断（仅Short）- 使用hyperopt参数 informative['trend_bear_5m'] = ( (informative['ema12'] < informative['ema26']) & (informative['ema26'] < informative['ema50']) & (informative['ema12_slope'] < -0.05) & (informative['adx_5m'] > self.adx_min.value) & (informative['adx_5m'] < self.adx_max.value) & (informative['close'] < informative['ema12']) & (informative['rsi_5m'] < self.rsi_max.value) & (informative['rsi_5m'] > self.rsi_min.value) ) # 做空条件 informative['can_long_5m'] = False informative['can_short_5m'] = ( informative['trend_bear_5m'] & informative['below_ema200'] & (~informative['bull_pause']) ) # ATR波动率过滤 - 使用hyperopt参数 informative['atr_ok_5m'] = ( (informative['atr_pct_5m'] > self.atr_min.value) & (informative['atr_pct_5m'] < informative['atr_pct_ma_5m'] * self.atr_max_mult.value) ) # 成交量确认 - 使用hyperopt参数 informative['volume_ma_5m'] = ta.SMA(informative['volume'], timeperiod=20) informative['volume_ok_5m'] = informative['volume'] > informative['volume_ma_5m'] * self.volume_threshold.value # 合并5分钟数据到1分钟 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) # 1分钟MACD斜率 dataframe['macd_slope'] = (dataframe['macd'] - dataframe['macd'].shift(3)) / 3 # 1分钟做空入场信号 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) ) # 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['hour_utc'] = dataframe['date'].dt.hour dataframe['is_bad_hour'] = dataframe['hour_utc'].isin([4, 5, 6, 7]) # 安全转换5分钟布尔列 bool_cols = [ 'can_long_5m_5m', 'can_short_5m_5m', 'trend_bear_5m_5m', 'atr_ok_5m_5m', 'below_ema200_5m', 'bull_pause_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'] # 5分钟MACD方向确认 macd_bear_5m = dataframe['macd_hist_5m_5m'] < 0 # 1分钟MACD方向确认 macd_bear_1m = dataframe['macd_hist'] < 0 # 成交量确认 volume_ok = dataframe['volume_ok_5m_5m'] # 做空入场 - 使用hyperopt参数 dataframe.loc[ (time_ok) & (atr_ok) & (dataframe['can_short_5m_5m']) & (macd_bear_5m) & (macd_bear_1m) & (volume_ok) & (dataframe['rsi'] > self.entry_rsi_min.value) & ( dataframe['top_divergence'] | dataframe['ema_cross_down'] | dataframe['bear_pullback'] ) & (dataframe['volume'] > 0), 'enter_short' ] = 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: """自定义出场逻辑：时间止损 - 使用hyperopt参数""" trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # 时间止损：持仓过久且亏损 if trade_duration > self.time_stop_1.value and current_profit < -0.005: return 'time_stop_1' if trade_duration > self.time_stop_2.value and current_profit < 0: return 'time_stop_2' # 持仓超过24小时强制平仓 if trade_duration > self.time_stop_3.value: return 'time_stop_3' 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