# 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) class CryptoFutures1m5mStrategyShortOnly(IStrategy): """ SOL/USDT 合约策略 - 只做空版本 基于V5修改： - 只做空，禁止做多 - 优化做空止损和止盈参数 """ INTERFACE_VERSION = 3 timeframe = '1m' informative_timeframe = '5m' can_short = True can_long = False # 禁用做多 lev = 1.0 # Trailing设置 - 基于V5 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 # 做空趋势 - 基于V5优化 informative['trend_bear_5m'] = ( (informative['ema12'] < informative['ema26']) & (informative['ema26'] < informative['ema50']) & (informative['ema12_slope'] < -0.05) & # V5标准 (informative['adx_5m'] > 24) & # V5标准 (informative['adx_5m'] < 51) & (informative['close'] < informative['ema12']) & (informative['rsi_5m'] < 48) & (informative['rsi_5m'] > 29) ) # 大趋势过滤 - 放宽条件，基于V5 informative['below_ema200'] = informative['ema200_dist_pct'] < -1.0 # 熊市确认 - 可选，不过度限制 informative['bear_market'] = (informative['ema200_slope'] < 0) & (informative['ema200_dist_pct'] < 0) # 做空条件 - 移除bear_market强制要求，基于V5 informative['can_short_5m'] = informative['trend_bear_5m'] & informative['below_ema200'] # ATR过滤 - 基于V5标准 informative['atr_ok_5m'] = ( (informative['atr_pct_5m'] > 0.07) & (informative['atr_pct_5m'] < informative['atr_pct_ma_5m'] * 2.2) ) # 成交量 - 基于V5标准 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) ) 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]) # 类型转换 bool_cols = ['can_short_5m_5m', 'trend_bear_5m_5m', 'atr_ok_5m_5m', 'below_ema200_5m', 'volume_ok_5m_5m', 'bear_market_5m'] for col in bool_cols: if col in dataframe.columns: dataframe[col] = dataframe[col].astype(bool).fillna(False) 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'] # 只做空 - 基于V5标准 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 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: 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 # 做空时间止损 - 基于V5标准 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' 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