# 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 CryptoFutures1m5mStrategyLongOnly(IStrategy): """ SOL/USDT 合约策略 - 只做多版本 基于V5修改： - 只做多，禁止做空 - 优化做多止损和止盈参数 """ INTERFACE_VERSION = 3 timeframe = '1m' informative_timeframe = '5m' can_short = False # 禁用做空 can_long = True lev = 1.0 stoploss = -0.035 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) ) # 大趋势过滤 informative['above_ema200'] = informative['ema200_dist_pct'] > 1.0 # 做多条件 informative['can_long_5m'] = informative['trend_bull_5m'] & informative['above_ema200'] # 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_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', 'trend_bull_5m_5m', 'atr_ok_5m_5m', 'above_ema200_5m', 'volume_ok_5m_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'] # 只做多 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 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_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