# source: https://raw.githubusercontent.com/t734070824/my-projects/6625ecd6d435e484bf5bd5f4bc43a508602275e3/my-btc-trade-system/buy_sell_notify/freqtrade/strategies-v3/TripleTimeframeTrendStrategyV3_5.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__TripleTimeframeTrendStrategyV3_5__20250811_095941(IStrategy): """ 三重时间框架趋势跟踪策略 V3.5 - 智能市场环境适应 V3.5 优化重点： 1. 解决强势市场中做多信号缺失问题 2. 添加市场环境识别机制 3. 动态调整多空信号阈值 4. 优化weak_short在上涨市场中的表现 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 # V3.5 更新：基于最新表现优化权重 pair_weights = { 'DOGE/USDT': 1.4, # 最近表现最佳 'AVAX/USDT': 1.3, # 长期优秀表现 'LINK/USDT': 1.2, # 稳定表现 'BNB/USDT': 1.2, # 最近表现好 'XRP/USDT': 1.1, # 调整：最近表现一般 'WIF/USDT': 1.0, # 高波动，中等权重 'ETH/USDT': 1.1, # 主流稳定 'SUI/USDT': 1.0, # 最近表现改善 'SOL/USDT': 1.0, # 降低：最近表现差 'NEAR/USDT': 0.9, # 表现一般 'DOT/USDT': 0.8, # 最近表现差 'UNI/USDT': 0.7, # 表现不稳定 'BTC/USDT': 0.8, # 大幅持仓时间过长 'ADA/USDT': 0.7, # 最近表现差 } # 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 # 做空专用阈值 (保持V3.4设置) 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 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) # 简化的风险指标 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 dataframe['h1_score'] = self.calculate_balanced_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'] inf_1d['daily_score'] = self.calculate_balanced_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'] inf_4h['h4_score'] = self.calculate_balanced_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_balanced_score(self, df: DataFrame) -> pd.Series: """平衡版评分计算""" 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 # 布林带评分 - 降低权重 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: """简化的信号强度计算""" 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 # 增加1h权重 # 成交量确认 - 简化 volume_confirm = np.where(dataframe['volume_ratio'] > 1.0, 0.5, 0) strength = daily_weight + h4_weight + h1_weight + volume_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: """ V3.5 双向入场条件 - 智能市场环境适应 """ pair = metadata['pair'] pair_weight = self.pair_weights.get(pair, 1.0) # 初始化新版FreqTrade列 dataframe['enter_long'] = 0 dataframe['enter_short'] = 0 dataframe['enter_tag'] = '' # V3.5 新增：为每行计算动态阈值 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) # === 做多信号 (V3.5 动态调整) === row = dataframe.iloc[i] # 强做多信号 - 使用动态阈值 strong_long_conditions = [ row['daily_score'] > 0, row['h4_score'] > 0, row['h1_score'] >= strong_long_thresh, # 动态阈值 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_v3_5' # 弱做多信号 - 使用动态阈值和市场环境 elif not dataframe.iloc[i]['enter_long']: # 没有强信号时 # 基础弱信号条件 weak_long_basic = [ row['daily_score'] >= 0, row['h4_score'] >= 0, row['h1_score'] >= weak_long_thresh, # 动态阈值 row['signal_strength'] >= 3, row['volume'] > 0, row['volume_ratio'] > 0.8 ] # 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(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_v3_5' # === 做空信号 (V3.5 优化，避免牛市过度做空) === if not dataframe.iloc[i]['enter_long']: # 避免同时做多做空 # 强做空信号 - 在牛市中收紧条件 strong_short_conditions = [ row['daily_score'] <= 0, row['h4_score'] <= 0, row['h1_score'] <= -strong_short_thresh, # 动态阈值 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_v3_5' # 弱做空信号 - 在牛市中大幅限制 elif not dataframe.iloc[i]['enter_short']: # 没有强做空信号时 # 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 ] else: # 非牛市中的正常弱做空条件 weak_short_conditions = [ row['daily_score'] <= 0, row['h4_score'] <= 0, row['h1_score'] <= weak_short_thresh, # 动态阈值 row['signal_strength'] >= 2, row['volume'] > 0, row['volume_ratio'] > 0.6, row['rsi'] > 35 ] if all(weak_short_conditions): dataframe.iloc[i, dataframe.columns.get_loc('enter_short')] = 1 dataframe.iloc[i, dataframe.columns.get_loc('enter_tag')] = 'weak_short_v3_5' return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: """ V3.5 双向退出条件 - 继承V3.4逻辑 """ # 初始化新版FreqTrade出场列 dataframe['exit_long'] = 0 dataframe['exit_short'] = 0 dataframe['exit_tag'] = '' # === 平多仓信号 (exit_long) === conditions_strong_exit_long = [] conditions_weak_exit_long = [] # 强烈平多仓 - 原V2.1逻辑 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) # 保护性平多仓 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) ) conditions_weak_exit_long.append(weak_exit_cond_1_long) conditions_weak_exit_long.append(weak_exit_cond_2_long) # === 平空仓信号 (exit_short) - V3.5 优化 === conditions_strong_exit_short = [] conditions_weak_exit_short = [] # 强烈平空仓 - V3.5 在牛市中更积极平空 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) # 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转强 ) 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_v3_5') 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_v3_5') # 执行平空仓信号 - V3.5 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_v3_5') 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_v3_5') 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]]': """基于表现的动态止盈 - 支持双向交易""" 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) # 基于币种表现调整止盈策略 - 双向一致 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 # 分批止盈 - 双向交易都适用 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}_v3_5' 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}_v3_5' # 信号恶化提前出场 - 双向交易都适用 if signal_strength < 1.5 and current_profit > 0.015: direction = 'long' if trade.is_short == False else 'short' return f'signal_weak_{direction}_v3_5' 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: """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: """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)