# source: https://raw.githubusercontent.com/t734070824/my-projects/6625ecd6d435e484bf5bd5f4bc43a508602275e3/my-btc-trade-system/buy_sell_notify/freqtrade/strategies-v4/TripleTimeframeTrendStrategyV4_1.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement from freqtrade.strategy.interface import IStrategy from typing import Dict, List, Optional from functools import reduce import talib.abstract as ta import numpy as np import pandas as pd from pandas import DataFrame class Github_t734070824_my_projects__TripleTimeframeTrendStrategyV4_1__20250811_095941(IStrategy): """ 三重时间框架趋势跟踪策略 V4.1 - KDJ辅助增强版 V4.1 新增特性： 1. 集成KDJ指标作为辅助确认 2. 震荡市场中KDJ权重增强 3. 利用KDJ精确择时优化进出场 4. KDJ与RSI分歧检测避免假突破 5. 保持V3.5所有智能环境适应功能 """ # 策略参数 INTERFACE_VERSION = 3 # 平衡的ROI设置 minimal_roi = { "0": 0.25, # 25% "30": 0.18, # 30分钟后18% "60": 0.12, # 1小时后12% "120": 0.08, # 2小时后8% "240": 0.05, # 4小时后5% "480": 0.03 # 8小时后3% } # 适中的止损 stoploss = -0.04 # 4% 止损 # 时间框架 timeframe = '1h' # 启用位置堆叠 position_stacking = False # 可以做多做空 can_short = True # 启用使用卖出信号 use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # 平衡的追踪止损设置 trailing_stop = True trailing_stop_positive = 0.02 # 2%后启用追踪止损 trailing_stop_positive_offset = 0.04 # 追踪止损偏移4% trailing_only_offset_is_reached = True # 优化的参数 sma_short_period = 18 # 保持敏感 sma_long_period = 46 # 适中 macd_fast = 11 # 保持敏感 macd_slow = 25 # 适中 macd_signal = 9 # 标准 rsi_period = 14 rsi_overbought = 71 # 70→71 稍微放宽 rsi_oversold = 29 # 30→29 稍微放宽 bb_period = 20 bb_std = 2.0 ichimoku_conversion = 9 ichimoku_base = 26 ichimoku_lagging = 52 atr_period = 14 # V4.1 新增：KDJ参数 kdj_k_period = 9 # K值周期 kdj_d_period = 3 # D值平滑周期 kdj_j_multiplier = 3 # J值计算倍数 J=3K-2D # KDJ阈值设定 kdj_oversold = 20 # 超卖阈值 kdj_overbought = 80 # 超买阈值 kdj_signal_weight = 0.3 # KDJ在震荡市场中的额外权重 # V4.1 更新：基于V3.5表现调整权重 pair_weights = { 'AVAX/USDT': 1.4, # V3.5中表现最佳 'DOGE/USDT': 1.3, # 稳定优秀表现 'LINK/USDT': 1.3, # V3.5中表现突出 'WIF/USDT': 1.2, # 高波动但盈利 'ETH/USDT': 1.2, # 主流币稳定 'XRP/USDT': 1.1, # 中等表现 'SUI/USDT': 1.1, # 表现改善 'DOT/USDT': 1.0, # 标准权重 'BNB/USDT': 1.0, # 长期持仓表现一般 'ADA/USDT': 0.9, # 表现一般 'SOL/USDT': 0.9, # 需要提升 'NEAR/USDT': 0.8, # 表现较弱 'UNI/USDT': 0.7, # 表现不稳定 'BTC/USDT': 0.8, # 大币种但持仓时间长 } # V3.5 继承：市场环境适应参数 market_strength_period = 20 # 市场强度计算周期 bullish_threshold = 0.05 # 牛市判断阈值：5%涨幅 bearish_threshold = -0.05 # 熊市判断阈值：-5%跌幅 # V3.5 继承：动态信号阈值 # 基础阈值 strong_signal_threshold = 3 weak_signal_threshold = 1 # 做空专用阈值 strong_short_threshold = 2 weak_short_threshold = 0 market_volatility_period = 50 trending_threshold = 0.02 def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs = [] for pair in pairs: informative_pairs.extend([ (pair, '1d'), (pair, '4h') ]) return informative_pairs def calculate_kdj(self, dataframe: DataFrame) -> DataFrame: """V4.1 新增：计算KDJ指标""" high = dataframe['high'] low = dataframe['low'] close = dataframe['close'] # 计算RSV (Raw Stochastic Value) lowest_low = low.rolling(window=self.kdj_k_period).min() highest_high = high.rolling(window=self.kdj_k_period).max() rsv = ((close - lowest_low) / (highest_high - lowest_low)) * 100 # 计算K值 (K = SMA(RSV, d_period)) dataframe['kdj_k'] = rsv.ewm(span=self.kdj_d_period).mean() # 计算D值 (D = SMA(K, d_period)) dataframe['kdj_d'] = dataframe['kdj_k'].ewm(span=self.kdj_d_period).mean() # 计算J值 (J = 3K - 2D) dataframe['kdj_j'] = (self.kdj_j_multiplier * dataframe['kdj_k']) - (2 * dataframe['kdj_d']) # KDJ信号判断 dataframe['kdj_bullish'] = (dataframe['kdj_k'] > dataframe['kdj_d']) & (dataframe['kdj_j'] > self.kdj_oversold) dataframe['kdj_bearish'] = (dataframe['kdj_k'] < dataframe['kdj_d']) & (dataframe['kdj_j'] < self.kdj_overbought) # KDJ与RSI分歧检测 rsi_bullish = dataframe['rsi'] > 50 rsi_bearish = dataframe['rsi'] < 50 dataframe['kdj_rsi_divergence'] = ( (dataframe['kdj_bullish'] & rsi_bearish) | (dataframe['kdj_bearish'] & rsi_bullish) ) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: inf_1d = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='1d') inf_4h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='4h') # 1小时指标 dataframe['sma_short'] = ta.SMA(dataframe, timeperiod=self.sma_short_period) dataframe['sma_long'] = ta.SMA(dataframe, timeperiod=self.sma_long_period) macd = ta.MACD(dataframe, fastperiod=self.macd_fast, slowperiod=self.macd_slow, signalperiod=self.macd_signal) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period) bollinger = ta.BBANDS(dataframe, timeperiod=self.bb_period, nbdevup=float(self.bb_std), nbdevdn=float(self.bb_std)) dataframe['bb_lower'] = bollinger['lowerband'] dataframe['bb_middle'] = bollinger['middleband'] dataframe['bb_upper'] = bollinger['upperband'] dataframe['bb_percent'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period) # V4.1 新增：计算KDJ指标 dataframe = self.calculate_kdj(dataframe) # 简化的风险指标 dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20) dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # V3.5 继承：市场环境识别 dataframe = self.calculate_market_environment(dataframe) # 市场状态 dataframe['market_volatility'] = dataframe['close'].rolling(self.market_volatility_period).std() / dataframe['close'].rolling(self.market_volatility_period).mean() dataframe['is_trending'] = dataframe['market_volatility'] > self.trending_threshold # V4.1 优化：包含KDJ的评分系统 dataframe['h1_score'] = self.calculate_enhanced_score(dataframe) # 处理日线数据 if len(inf_1d) > 0: inf_1d['sma_short'] = ta.SMA(inf_1d, timeperiod=self.sma_short_period) inf_1d['sma_long'] = ta.SMA(inf_1d, timeperiod=self.sma_long_period) inf_1d['rsi'] = ta.RSI(inf_1d, timeperiod=self.rsi_period) macd_1d = ta.MACD(inf_1d, fastperiod=self.macd_fast, slowperiod=self.macd_slow, signalperiod=self.macd_signal) inf_1d['macd'] = macd_1d['macd'] inf_1d['macdsignal'] = macd_1d['macdsignal'] inf_1d['macdhist'] = macd_1d['macdhist'] bollinger_1d = ta.BBANDS(inf_1d, timeperiod=self.bb_period, nbdevup=float(self.bb_std), nbdevdn=float(self.bb_std)) inf_1d['bb_lower'] = bollinger_1d['lowerband'] inf_1d['bb_middle'] = bollinger_1d['middleband'] inf_1d['bb_upper'] = bollinger_1d['upperband'] # V4.1：日线也计算KDJ inf_1d = self.calculate_kdj(inf_1d) inf_1d['daily_score'] = self.calculate_enhanced_score(inf_1d) dataframe = pd.merge(dataframe, inf_1d[['date', 'daily_score']], on='date', how='left') dataframe['daily_score'] = dataframe['daily_score'].fillna(method='ffill') # 处理4小时数据 if len(inf_4h) > 0: inf_4h['sma_short'] = ta.SMA(inf_4h, timeperiod=self.sma_short_period) inf_4h['sma_long'] = ta.SMA(inf_4h, timeperiod=self.sma_long_period) inf_4h['rsi'] = ta.RSI(inf_4h, timeperiod=self.rsi_period) macd_4h = ta.MACD(inf_4h, fastperiod=self.macd_fast, slowperiod=self.macd_slow, signalperiod=self.macd_signal) inf_4h['macd'] = macd_4h['macd'] inf_4h['macdsignal'] = macd_4h['macdsignal'] inf_4h['macdhist'] = macd_4h['macdhist'] bollinger_4h = ta.BBANDS(inf_4h, timeperiod=self.bb_period, nbdevup=float(self.bb_std), nbdevdn=float(self.bb_std)) inf_4h['bb_lower'] = bollinger_4h['lowerband'] inf_4h['bb_middle'] = bollinger_4h['middleband'] inf_4h['bb_upper'] = bollinger_4h['upperband'] # V4.1：4小时也计算KDJ inf_4h = self.calculate_kdj(inf_4h) inf_4h['h4_score'] = self.calculate_enhanced_score(inf_4h) dataframe = pd.merge(dataframe, inf_4h[['date', 'h4_score']], on='date', how='left') dataframe['h4_score'] = dataframe['h4_score'].fillna(method='ffill') dataframe['daily_score'] = dataframe['daily_score'].fillna(0) dataframe['h4_score'] = dataframe['h4_score'].fillna(0) dataframe['signal_strength'] = self.calculate_signal_strength(dataframe) return dataframe def calculate_market_environment(self, dataframe: DataFrame) -> DataFrame: """V3.5 继承：市场环境识别""" # 计算价格变化率 dataframe['price_change'] = dataframe['close'].pct_change(periods=self.market_strength_period) # 市场环境分类 dataframe['is_strong_bull'] = dataframe['price_change'] > self.bullish_threshold * 1.5 # 7.5% dataframe['is_bull'] = (dataframe['price_change'] > self.bullish_threshold) & (dataframe['price_change'] <= self.bullish_threshold * 1.5) dataframe['is_strong_bear'] = dataframe['price_change'] < self.bearish_threshold * 1.5 # -7.5% dataframe['is_bear'] = (dataframe['price_change'] < self.bearish_threshold) & (dataframe['price_change'] >= self.bearish_threshold * 1.5) dataframe['is_sideways'] = (dataframe['price_change'] >= self.bearish_threshold) & (dataframe['price_change'] <= self.bullish_threshold) # 市场强度评分 (-2到2) dataframe['market_strength'] = np.where( dataframe['is_strong_bull'], 2, np.where(dataframe['is_bull'], 1, np.where(dataframe['is_strong_bear'], -2, np.where(dataframe['is_bear'], -1, 0) ) ) ) return dataframe def calculate_enhanced_score(self, df: DataFrame) -> pd.Series: """V4.1 增强：包含KDJ的评分计算""" score = pd.Series(0.0, index=df.index) # SMA趋势评分 - 标准权重 sma_trend = np.where( (df['close'] > df['sma_short']) & (df['sma_short'] > df['sma_long']), 2, np.where( (df['close'] < df['sma_short']) & (df['sma_short'] < df['sma_long']), -2, np.where(df['close'] > df['sma_short'], 1, -1) ) ) score += sma_trend # MACD评分 if 'macdhist' in df.columns: macd_score = np.where( (df['macd'] > df['macdsignal']) & (df['macdhist'] > 0), 2, np.where( (df['macd'] < df['macdsignal']) & (df['macdhist'] < 0), -2, np.where(df['macd'] > df['macdsignal'], 1, -1) ) ) else: macd_score = np.where(df['macd'] > df['macdsignal'], 1, -1) score += macd_score # RSI评分 - 适中权重 rsi_score = np.where( df['rsi'] > 60, 1, np.where(df['rsi'] < 40, -1, 0) ) score += rsi_score # V4.1 新增：KDJ评分 if 'kdj_k' in df.columns and 'kdj_d' in df.columns and 'kdj_j' in df.columns: # KDJ金叉死叉评分 kdj_cross_score = np.where( (df['kdj_k'] > df['kdj_d']) & (df['kdj_j'] > self.kdj_oversold), 1, np.where( (df['kdj_k'] < df['kdj_d']) & (df['kdj_j'] < self.kdj_overbought), -1, 0 ) ) # V4.1 重点：在震荡市场中增加KDJ权重 market_env_multiplier = 1.0 if 'is_sideways' in df.columns: market_env_multiplier = np.where(df['is_sideways'], 1 + self.kdj_signal_weight, 1.0) score += kdj_cross_score * market_env_multiplier # 布林带评分 - 降低权重 if 'bb_upper' in df.columns and 'bb_lower' in df.columns: bb_score = np.where( df['close'] > df['bb_upper'], 0.5, np.where(df['close'] < df['bb_lower'], -0.5, 0) ) score += bb_score return score def calculate_signal_strength(self, dataframe: DataFrame) -> pd.Series: """V4.1 增强：包含KDJ的信号强度计算""" strength = pd.Series(0.0, index=dataframe.index) # 三时间框架权重 daily_weight = abs(dataframe['daily_score']) * 0.4 h4_weight = abs(dataframe['h4_score']) * 0.3 h1_weight = abs(dataframe['h1_score']) * 0.3 # 成交量确认 - 简化 volume_confirm = np.where(dataframe['volume_ratio'] > 1.0, 0.5, 0) # V4.1 新增：KDJ确认加分 kdj_confirm = 0 if 'kdj_bullish' in dataframe.columns and 'kdj_bearish' in dataframe.columns: kdj_confirm = np.where( dataframe['kdj_bullish'] | dataframe['kdj_bearish'], 0.3, 0 ) # KDJ与RSI分歧时减分 if 'kdj_rsi_divergence' in dataframe.columns: kdj_confirm = np.where(dataframe['kdj_rsi_divergence'], kdj_confirm * 0.5, kdj_confirm) strength = daily_weight + h4_weight + h1_weight + volume_confirm + kdj_confirm return np.clip(strength, 0, 10) def get_dynamic_thresholds(self, dataframe: DataFrame, row_index: int) -> tuple: """V3.5 继承：根据市场环境动态调整阈值""" if row_index >= len(dataframe): return self.strong_signal_threshold, self.weak_signal_threshold, self.strong_short_threshold, self.weak_short_threshold market_strength = dataframe.iloc[row_index]['market_strength'] # 在强势牛市中放宽做多条件，收紧做空条件 if market_strength >= 1: # 牛市环境 strong_long = max(1, self.strong_signal_threshold - 1) # 3→2 或 2→1 weak_long = max(0, self.weak_signal_threshold - 1) # 1→0 strong_short = self.strong_short_threshold + 1 # 2→3 收紧 weak_short = self.weak_short_threshold + 1 # 0→1 收紧 elif market_strength <= -1: # 熊市环境 strong_long = self.strong_signal_threshold + 1 # 3→4 收紧 weak_long = self.weak_signal_threshold + 1 # 1→2 收紧 strong_short = max(1, self.strong_short_threshold - 1) # 2→1 放宽 weak_short = max(-1, self.weak_short_threshold - 1) # 0→-1 放宽 else: # 横盘市场 strong_long = self.strong_signal_threshold weak_long = self.weak_signal_threshold strong_short = self.strong_short_threshold weak_short = self.weak_short_threshold return strong_long, weak_long, strong_short, weak_short def populate_entry_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: """ V4.1 双向入场条件 - 集成KDJ辅助确认 """ pair = metadata['pair'] pair_weight = self.pair_weights.get(pair, 1.0) # 初始化新版FreqTrade列 dataframe['enter_long'] = 0 dataframe['enter_short'] = 0 dataframe['enter_tag'] = '' # V4.1 新增：为每行计算动态阈值并集成KDJ判断 for i in range(len(dataframe)): if i < self.market_strength_period: # 前期数据不足时使用默认值 continue strong_long_thresh, weak_long_thresh, strong_short_thresh, weak_short_thresh = self.get_dynamic_thresholds(dataframe, i) # === 做多信号 (V4.1 KDJ增强) === row = dataframe.iloc[i] # V4.1：KDJ择时优化 kdj_timing_bonus_long = 0 if 'kdj_bullish' in dataframe.columns: if row['kdj_bullish'] and not row.get('kdj_rsi_divergence', False): kdj_timing_bonus_long = 0.5 # KDJ确认加分 elif row.get('kdj_rsi_divergence', False): kdj_timing_bonus_long = -0.5 # 分歧减分 # 强做多信号 - 使用动态阈值+KDJ确认 strong_long_conditions = [ row['daily_score'] > 0, row['h4_score'] > 0, row['h1_score'] >= (strong_long_thresh - kdj_timing_bonus_long), # KDJ调整阈值 row['volume'] > 0, row['rsi'] < self.rsi_overbought, row['volume_ratio'] > 0.8 ] # 币种权重调整 if pair_weight >= 1.2: strong_long_conditions.append(row['rsi'] < 75) elif pair_weight >= 0.8: strong_long_conditions.append(row['rsi'] < self.rsi_overbought) else: strong_long_conditions.append(row['rsi'] < 68) if all(strong_long_conditions): dataframe.iloc[i, dataframe.columns.get_loc('enter_long')] = 1 dataframe.iloc[i, dataframe.columns.get_loc('enter_tag')] = 'strong_long_v4_1' # 弱做多信号 - KDJ增强版 elif not dataframe.iloc[i]['enter_long']: # 没有强信号时 # 基础弱信号条件 weak_long_basic = [ row['daily_score'] >= 0, row['h4_score'] >= 0, row['h1_score'] >= (weak_long_thresh - kdj_timing_bonus_long), # KDJ调整 row['signal_strength'] >= 3, row['volume'] > 0, row['volume_ratio'] > 0.8 ] # V4.1 新增：震荡市场中KDJ权重增强 sideways_kdj_condition = [ row.get('is_sideways', False), # 震荡市场 row.get('kdj_bullish', False), # KDJ看多 not row.get('kdj_rsi_divergence', False), # 无分歧 row['h1_score'] >= 0, # 1h不太差 row['signal_strength'] >= 2 # 降低要求 ] # V3.5 继承：强势牛市中的特殊做多条件 bull_market_condition = [ row['market_strength'] >= 1, # 牛市环境 row['h1_score'] >= 0, # 1h不太差即可 row['rsi'] < 80, # 放宽RSI限制 row['close'] > row['sma_short'], # 价格在短期均线上方 row['signal_strength'] >= 2 # 降低信号强度要求 ] # 趋势市场中的机会信号 trend_condition = [ row['is_trending'], row['daily_score'] > 1, row['h1_score'] >= weak_long_thresh, row['signal_strength'] >= 4 ] if (all(weak_long_basic) or all(sideways_kdj_condition) or all(bull_market_condition) or all(trend_condition)): # 应用币种权重过滤 rsi_filter = True if pair_weight >= 1.2: rsi_filter = row['rsi'] < 78 elif pair_weight >= 0.8: rsi_filter = row['rsi'] < 73 else: rsi_filter = row['rsi'] < 70 if rsi_filter: dataframe.iloc[i, dataframe.columns.get_loc('enter_long')] = 1 dataframe.iloc[i, dataframe.columns.get_loc('enter_tag')] = 'weak_long_v4_1' # === 做空信号 (V4.1 KDJ增强，避免牛市过度做空) === if not dataframe.iloc[i]['enter_long']: # 避免同时做多做空 # V4.1：KDJ择时优化 kdj_timing_bonus_short = 0 if 'kdj_bearish' in dataframe.columns: if row['kdj_bearish'] and not row.get('kdj_rsi_divergence', False): kdj_timing_bonus_short = 0.5 # KDJ确认加分 elif row.get('kdj_rsi_divergence', False): kdj_timing_bonus_short = -0.5 # 分歧减分 # 强做空信号 - 在牛市中收紧条件 strong_short_conditions = [ row['daily_score'] <= 0, row['h4_score'] <= 0, row['h1_score'] <= -(strong_short_thresh - kdj_timing_bonus_short), # KDJ调整 row['volume'] > 0, row['volume_ratio'] > 0.6 ] # V3.5 继承：牛市中的额外限制 if row['market_strength'] >= 1: # 强势牛市 strong_short_conditions.extend([ row['rsi'] > 75, # RSI必须超买 row['h1_score'] <= -3, # 1h信号必须很弱 row['macdhist'] < 0 # MACD转弱 ]) else: strong_short_conditions.append(row['rsi'] > 35) # 币种权重调整 if pair_weight >= 1.2: strong_short_conditions.append(row['rsi'] > 30) elif pair_weight >= 0.8: strong_short_conditions.append(row['rsi'] > 35) else: strong_short_conditions.append(row['rsi'] > 38) if all(strong_short_conditions): dataframe.iloc[i, dataframe.columns.get_loc('enter_short')] = 1 dataframe.iloc[i, dataframe.columns.get_loc('enter_tag')] = 'strong_short_v4_1' # 弱做空信号 - V4.1 KDJ增强，在牛市中大幅限制 elif not dataframe.iloc[i]['enter_short']: # 没有强做空信号时 # V4.1 新增：震荡市场中KDJ辅助做空 sideways_kdj_short = [ row.get('is_sideways', False), # 震荡市场 row.get('kdj_bearish', False), # KDJ看空 not row.get('kdj_rsi_divergence', False), # 无分歧 row['h1_score'] <= 0, # 1h转弱 row['signal_strength'] >= 2 # 信号要求 ] # V3.5 继承：在强势牛市中几乎禁用弱做空 if row['market_strength'] >= 1: # 牛市中弱做空条件极严格 weak_short_conditions = [ row['rsi'] > 80, # RSI极度超买 row['daily_score'] < -1, # 日线明确转弱 row['h4_score'] < -1, # 4h明确转弱 row['h1_score'] <= -2, # 1h信号较弱 row['close'] < row['bb_lower'], # 价格击穿布林下轨 row['macdhist'] < 0, # MACD转弱 row['signal_strength'] >= 4, row.get('kdj_bearish', False) # V4.1：必须KDJ确认 ] else: # 非牛市中的正常弱做空条件 weak_short_conditions = [ row['daily_score'] <= 0, row['h4_score'] <= 0, row['h1_score'] <= (weak_short_thresh - kdj_timing_bonus_short), row['signal_strength'] >= 2, row['volume'] > 0, row['volume_ratio'] > 0.6, row['rsi'] > 35 ] if all(weak_short_conditions) or all(sideways_kdj_short): dataframe.iloc[i, dataframe.columns.get_loc('enter_short')] = 1 dataframe.iloc[i, dataframe.columns.get_loc('enter_tag')] = 'weak_short_v4_1' return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: """ V4.1 双向退出条件 - KDJ辅助确认优化 """ # 初始化新版FreqTrade出场列 dataframe['exit_long'] = 0 dataframe['exit_short'] = 0 dataframe['exit_tag'] = '' # === 平多仓信号 (exit_long) - V4.1 KDJ增强 === conditions_strong_exit_long = [] conditions_weak_exit_long = [] # 强烈平多仓 - 原逻辑 + KDJ确认 conditions_strong_exit_long.append(dataframe['daily_score'] < -1) conditions_strong_exit_long.append(dataframe['h4_score'] < -1) conditions_strong_exit_long.append(dataframe['h1_score'] <= -self.strong_signal_threshold) # V4.1 新增：KDJ死叉确认 if 'kdj_bearish' in dataframe.columns: conditions_strong_exit_long.append(dataframe['kdj_bearish']) # 保护性平多仓 - KDJ辅助 weak_exit_cond_1_long = ( (dataframe['signal_strength'] < 2) & (dataframe['h1_score'] <= 0) & (dataframe['rsi'] > 65) ) weak_exit_cond_2_long = ( (dataframe['rsi'] > self.rsi_overbought) & (dataframe['close'] > dataframe['bb_upper']) & (dataframe['macdhist'] < 0) ) # V4.1 新增：KDJ顶背离平多仓 if 'kdj_j' in dataframe.columns: weak_exit_cond_3_long = ( (dataframe['kdj_j'] > self.kdj_overbought) & (dataframe['kdj_k'] < dataframe['kdj_d']) & # KDJ死叉 (dataframe['signal_strength'] < 3) ) conditions_weak_exit_long.append(weak_exit_cond_3_long) conditions_weak_exit_long.append(weak_exit_cond_1_long) conditions_weak_exit_long.append(weak_exit_cond_2_long) # === 平空仓信号 (exit_short) - V4.1 KDJ优化 === conditions_strong_exit_short = [] conditions_weak_exit_short = [] # 强烈平空仓 - V4.1 在牛市中更积极平空 conditions_strong_exit_short.append(dataframe['daily_score'] > 0) conditions_strong_exit_short.append(dataframe['h4_score'] > 0) conditions_strong_exit_short.append(dataframe['h1_score'] >= self.strong_short_threshold) # V4.1 新增：KDJ金叉确认 if 'kdj_bullish' in dataframe.columns: conditions_strong_exit_short.append(dataframe['kdj_bullish']) # V3.5 继承：牛市中的快速平空条件 bull_market_exit_short = ( (dataframe['market_strength'] >= 1) & # 牛市环境 (dataframe['h1_score'] >= 0) & # 1h转正 (dataframe['close'] > dataframe['sma_short']) # 价格突破短期均线 ) conditions_strong_exit_short.append(bull_market_exit_short) # 保护性平空仓 weak_exit_cond_1_short = ( (dataframe['signal_strength'] < 2) & (dataframe['h1_score'] >= 0) & # 对称：1h转正 (dataframe['rsi'] < 40) # 放宽：< 35 → < 40 ) weak_exit_cond_2_short = ( (dataframe['rsi'] < self.rsi_oversold) & # 对称：RSI超卖 (dataframe['close'] < dataframe['bb_lower']) & # 对称：触及布林下轨 (dataframe['macdhist'] > 0) # 对称：MACD转强 ) # V4.1 新增：KDJ底背离平空仓 if 'kdj_j' in dataframe.columns: weak_exit_cond_3_short = ( (dataframe['kdj_j'] < self.kdj_oversold) & (dataframe['kdj_k'] > dataframe['kdj_d']) & # KDJ金叉 (dataframe['signal_strength'] < 3) ) conditions_weak_exit_short.append(weak_exit_cond_3_short) conditions_weak_exit_short.append(weak_exit_cond_1_short) conditions_weak_exit_short.append(weak_exit_cond_2_short) # 通用退出过滤 common_filters_exit = [] common_filters_exit.append(dataframe['volume'] > 0) # 执行平多仓信号 common_filters_exit_long = common_filters_exit + [dataframe['rsi'] > self.rsi_oversold] if conditions_strong_exit_long and common_filters_exit_long: strong_exit_long = reduce(lambda x, y: x & y, conditions_strong_exit_long + common_filters_exit_long) dataframe.loc[strong_exit_long, ['exit_long', 'exit_tag']] = (1, 'strong_bearish_v4_1') if conditions_weak_exit_long and common_filters_exit_long: weak_exit_long = reduce(lambda x, y: x | y, conditions_weak_exit_long) weak_exit_long = weak_exit_long & reduce(lambda x, y: x & y, common_filters_exit_long) # 避免覆盖强信号 weak_exit_long = weak_exit_long & (dataframe['exit_long'] != 1) dataframe.loc[weak_exit_long, ['exit_long', 'exit_tag']] = (1, 'protect_profit_long_v4_1') # 执行平空仓信号 - V4.1 common_filters_exit_short = common_filters_exit + [dataframe['rsi'] < self.rsi_overbought] if conditions_strong_exit_short and common_filters_exit_short: strong_exit_short = reduce(lambda x, y: x | y, conditions_strong_exit_short) # 使用OR逻辑 strong_exit_short = strong_exit_short & reduce(lambda x, y: x & y, common_filters_exit_short) dataframe.loc[strong_exit_short, ['exit_short', 'exit_tag']] = (1, 'strong_bullish_v4_1') if conditions_weak_exit_short and common_filters_exit_short: weak_exit_short = reduce(lambda x, y: x | y, conditions_weak_exit_short) weak_exit_short = weak_exit_short & reduce(lambda x, y: x & y, common_filters_exit_short) # 避免覆盖强信号 weak_exit_short = weak_exit_short & (dataframe['exit_short'] != 1) dataframe.loc[weak_exit_short, ['exit_short', 'exit_tag']] = (1, 'protect_profit_short_v4_1') return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time, current_rate: float, current_profit: float, **kwargs) -> float: """基于表现的动态止损 - 支持双向交易""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 1: return self.stoploss last_candle = dataframe.iloc[-1] pair_weight = self.pair_weights.get(pair, 1.0) if 'atr' in last_candle: atr_value = last_candle['atr'] # 根据币种表现调整止损 - 双向一致 if pair_weight >= 1.3: # 优秀币种 - 给更多空间 atr_multiplier = 2.2 max_loss = 0.06 elif pair_weight >= 1.0: # 标准币种 atr_multiplier = 2.0 max_loss = 0.05 else: # 表现差的币种 - 更紧止损 atr_multiplier = 1.8 max_loss = 0.045 atr_stoploss = atr_multiplier * atr_value / current_rate return max(-max_loss, min(-0.02, -atr_stoploss)) return self.stoploss def custom_exit(self, pair: str, trade: 'Trade', current_time, current_rate: float, current_profit: float, **kwargs) -> 'Optional[Union[str, bool]]': """V4.1 优化：基于KDJ的动态止盈""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 1: return None last_candle = dataframe.iloc[-1] signal_strength = last_candle.get('signal_strength', 5) pair_weight = self.pair_weights.get(pair, 1.0) # V4.1 新增：KDJ极值检测 kdj_extreme_exit = False if 'kdj_j' in last_candle: if trade.is_short == False: # 做多持仓 # KDJ J值极度超买时考虑止盈 kdj_extreme_exit = last_candle['kdj_j'] > 90 and current_profit > 0.02 else: # 做空持仓 # KDJ J值极度超卖时考虑止盈 kdj_extreme_exit = last_candle['kdj_j'] < 10 and current_profit > 0.02 # 基于币种表现调整止盈策略 - 双向一致 if pair_weight >= 1.3: # 优秀币种 - 更高目标 first_target = 0.08 second_target = 0.15 elif pair_weight >= 1.0: # 标准币种 first_target = 0.06 second_target = 0.12 else: # 表现差的币种 - 快速止盈 first_target = 0.04 second_target = 0.08 # V4.1：KDJ极值时提前止盈 if kdj_extreme_exit: direction = 'long' if trade.is_short == False else 'short' return f'kdj_extreme_{direction}_v4_1' # 分批止盈 - 双向交易都适用 if current_profit > first_target and not hasattr(trade, 'first_exit_done'): trade.first_exit_done = True direction = 'long' if trade.is_short == False else 'short' return f'first_target_{direction}_v4_1' if current_profit > second_target and not hasattr(trade, 'second_exit_done'): trade.second_exit_done = True direction = 'long' if trade.is_short == False else 'short' return f'second_target_{direction}_v4_1' # 信号恶化提前出场 - 双向交易都适用 if signal_strength < 1.5 and current_profit > 0.015: direction = 'long' if trade.is_short == False else 'short' return f'signal_weak_{direction}_v4_1' return None def position_sizing(self, pair: str, current_time, current_rate: float, proposed_stake: float, min_stake: float, max_stake: float, side: str, **kwargs) -> float: """V4.1 继承V3.5：基于市场环境的仓位调整""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) pair_weight = self.pair_weights.get(pair, 1.0) # V3.5 继承：根据市场环境调整做空仓位 if len(dataframe) > 0: last_candle = dataframe.iloc[-1] market_strength = last_candle.get('market_strength', 0) signal_strength = last_candle.get('signal_strength', 5) # 市场环境调整 if side == 'short': if market_strength >= 1: # 牛市中减小做空仓位 direction_multiplier = 0.6 elif market_strength <= -1: # 熊市中正常做空仓位 direction_multiplier = 1.0 else: # 横盘市场 direction_multiplier = 0.8 else: # 做多 if market_strength >= 1: # 牛市中增加做多仓位 direction_multiplier = 1.1 elif market_strength <= -1: # 熊市中减小做多仓位 direction_multiplier = 0.8 else: # 横盘市场 direction_multiplier = 1.0 # 信号强度调整 if signal_strength >= 7: strength_multiplier = 1.2 elif signal_strength >= 5: strength_multiplier = 1.0 else: strength_multiplier = 0.8 else: direction_multiplier = 0.9 if side == 'short' else 1.0 strength_multiplier = 1.0 # 最终倍数 final_multiplier = pair_weight * strength_multiplier * direction_multiplier # 限制倍数范围 final_multiplier = max(0.5, min(final_multiplier, 1.5)) final_stake = proposed_stake * final_multiplier return max(min_stake, min(final_stake, max_stake)) def leverage(self, pair: str, current_time, current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: """V4.1 继承V3.5：基于市场环境的杠杆调整""" pair_weight = self.pair_weights.get(pair, 1.0) # 基础杠杆根据币种表现 if pair_weight >= 1.3: base_leverage = 4.0 # 优秀币种适中杠杆 elif pair_weight >= 1.0: base_leverage = 3.0 else: base_leverage = 2.5 # 表现差的币种低杠杆 # V3.5 继承：根据市场环境调整杠杆 dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) > 0: last_candle = dataframe.iloc[-1] market_strength = last_candle.get('market_strength', 0) # 在强势趋势中降低杠杆，减少风险 if abs(market_strength) >= 2: # 强势市场 environment_multiplier = 0.8 elif abs(market_strength) >= 1: # 趋势市场 environment_multiplier = 0.9 else: # 横盘市场 environment_multiplier = 1.0 # 做空风险略高，杠杆稍微降低 direction_multiplier = 0.8 if side == 'short' else 1.0 target_leverage = base_leverage * environment_multiplier * direction_multiplier else: # 做空风险略高，杠杆稍微降低 direction_multiplier = 0.8 if side == 'short' else 1.0 target_leverage = base_leverage * direction_multiplier return min(target_leverage, max_leverage)