# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove这些 imports --- import numpy as np import pandas as pd from datetime import datetime, timedelta, timezone from pandas import DataFrame from typing import Dict, Optional, Union, Tuple from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, # @informative decorator # Hyperopt Parameters BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, # timeframe helpers timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, timeframe_to_seconds, timeframe_to_msecs, ) import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib class EnhancedMAVolumeStrategy(IStrategy): """ 增强版均线金叉 + 成交量放大策略 核心优化功能： 1. 动态止损 - 基于趋势强度和确认条件，减少噪音触发 2. 高胜率exit信号 - 保留有效退出条件，过滤低质量信号 3. 盈利单加权管理 - 对盈利交易进行加仓和延长持仓 4. 持仓时间优化 - 避免长期亏损拖累账户 策略逻辑： 买入：均线金叉 + 成交量放大 + 趋势确认 卖出：高胜率信号 + 利润保护 + 时间管理 止损：动态调整，基于市场条件 """ INTERFACE_VERSION = 3 # 策略时间框架 timeframe = "15m" # 是否可以做空 can_short: bool = False # ROI设置 - 更积极的利润目标 minimal_roi = { "2880": 0.02, # 48小时后2%收益 "1440": 0.025, # 24小时后2.5%收益 "720": 0.03, # 12小时后3%收益 "480": 0.035, # 8小时后3.5%收益 "360": 0.04, # 6小时后4%收益 "240": 0.045, # 4小时后4.5%收益 "180": 0.05, # 3小时后5%收益 "120": 0.06, # 2小时后6%收益 "60": 0.07, # 1小时后7%收益 "30": 0.08, # 30分钟后8%收益 "0": 0.1 # 立即10%收益 } # 基础止损设置 - 动态调整 stoploss = -0.1 # 10%基础止损 # 追踪止损 - 更保守的设置 trailing_stop = True trailing_stop_positive = 0.02 # 2%开始追踪 trailing_stop_positive_offset = 0.025 # 2.5%偏移 trailing_only_offset_is_reached = True # 只处理新K线 process_only_new_candles = True # 策略参数 - 禁用低胜率exit_signal use_exit_signal = False # 完全禁用exit_signal，使用自定义退出 exit_profit_only = False ignore_roi_if_entry_signal = False # 策略启动所需的K线数量 startup_candle_count: int = 100 # 仓位调整参数 position_adjustment_enable = True max_entry_position_adjustment = 3 # 最多加仓3次 # 订单类型 order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False } # 订单时间 order_time_in_force = { "entry": "GTC", "exit": "GTC" } # 策略参数 - 优化后的参数 # 均线参数 fast_ma_period = IntParameter(5, 15, default=8, space="buy") slow_ma_period = IntParameter(15, 30, default=21, space="buy") # 成交量参数 volume_ma_period = IntParameter(10, 20, default=15, space="buy") volume_threshold = DecimalParameter(1.1, 2.0, default=1.3, space="buy") # 支撑位参数 support_ma_period = IntParameter(30, 60, default=50, space="sell") # 动态止损参数 dynamic_stoploss_enabled = BooleanParameter(default=True, space="sell") trend_strength_threshold = IntParameter(1, 3, default=2, space="sell") # 持仓时间管理参数 max_hold_hours = IntParameter(24, 72, default=48, space="sell") profit_hold_extension = IntParameter(12, 48, default=24, space="sell") def informative_pairs(self): return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 计算技术指标 - 增强版 """ # 移动平均线 - 用于金叉死叉判断 dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=self.fast_ma_period.value) dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=self.slow_ma_period.value) # 长期均线 - 用于支撑位判断 dataframe["sma_support"] = ta.SMA(dataframe, timeperiod=self.support_ma_period.value) # 成交量指标 dataframe["volume_ma"] = dataframe["volume"].rolling(window=self.volume_ma_period.value).mean() dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_ma"] # 均线金叉死叉信号 dataframe["ma_golden_cross"] = ( (dataframe["ema_fast"] > dataframe["ema_slow"]) & (dataframe["ema_fast"].shift(1) <= dataframe["ema_slow"].shift(1)) ) dataframe["ma_death_cross"] = ( (dataframe["ema_fast"] < dataframe["ema_slow"]) & (dataframe["ema_fast"].shift(1) >= dataframe["ema_slow"].shift(1)) ) # 成交量放大信号 dataframe["volume_surge"] = dataframe["volume_ratio"] >= self.volume_threshold.value # 价格位置指标 dataframe["price_above_fast_ma"] = dataframe["close"] > dataframe["ema_fast"] dataframe["price_above_slow_ma"] = dataframe["close"] > dataframe["ema_slow"] dataframe["price_above_support"] = dataframe["close"] > dataframe["sma_support"] # 趋势强度 - 增强版 dataframe["trend_strength"] = ( (dataframe["ema_fast"] > dataframe["ema_slow"]).astype(int) + (dataframe["close"] > dataframe["ema_fast"]).astype(int) + (dataframe["close"] > dataframe["ema_slow"]).astype(int) + (dataframe["ema_fast"] > dataframe["ema_fast"].shift(1)).astype(int) + (dataframe["ema_slow"] > dataframe["ema_slow"].shift(1)).astype(int) ) # 支撑位强度 - 增强版 dataframe["support_strength"] = ( (dataframe["close"] > dataframe["sma_support"]).astype(int) + (dataframe["low"] > dataframe["sma_support"]).astype(int) + (dataframe["sma_support"] > dataframe["sma_support"].shift(1)).astype(int) ) # 趋势确认指标 dataframe["trend_confirmation"] = ( (dataframe["trend_strength"] >= 3) & (dataframe["support_strength"] >= 2) & (dataframe["volume_ratio"] >= 1.1) ) # RSI - 辅助判断 dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) # MACD - 趋势确认 macd = ta.MACD(dataframe) dataframe["macd"] = macd["macd"] dataframe["macdsignal"] = macd["macdsignal"] dataframe["macdhist"] = macd["macdhist"] # 布林带 - 价格位置 bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2) dataframe["bb_lowerband"] = bollinger["lower"] dataframe["bb_middleband"] = bollinger["mid"] dataframe["bb_upperband"] = bollinger["upper"] dataframe["bb_percent"] = ( (dataframe["close"] - dataframe["bb_lowerband"]) / (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) ) # 波动率指标 dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["volatility"] = dataframe["atr"] / dataframe["close"] # 价格动量 dataframe["price_momentum"] = ( (dataframe["close"] - dataframe["close"].shift(5)) / dataframe["close"].shift(5) ) # 成交量动量 dataframe["volume_momentum"] = ( (dataframe["volume"] - dataframe["volume"].shift(5)) / dataframe["volume"].shift(5) ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 入场条件：均线金叉 + 成交量放大 + 趋势确认 """ # 基础条件 base_conditions = ( (dataframe["volume"] > 0) & (dataframe["ema_fast"].notna()) & (dataframe["ema_slow"].notna()) & (dataframe["volume_ma"].notna()) ) # 主要入场条件：均线金叉 ma_golden_cross = ( dataframe["ma_golden_cross"] | # 标准金叉 (dataframe["ema_fast"] > dataframe["ema_slow"] * 0.999) # 接近金叉 ) # 成交量确认 volume_confirmation = ( dataframe["volume_surge"] | # 成交量放大 (dataframe["volume_ratio"] >= 1.1) # 成交量稍微放大即可 ) # 价格位置确认 price_position = ( dataframe["price_above_fast_ma"] | # 价格在快线之上 (dataframe["close"] > dataframe["ema_slow"] * 0.95) # 价格接近慢线 ) # 趋势确认 - 增强版 trend_confirmation = ( dataframe["trend_confirmation"] | # 强趋势确认 (dataframe["trend_strength"] >= 2) | # 中等趋势 (dataframe["price_momentum"] > 0) # 价格动量向上 ) # 辅助条件 additional_conditions = ( (dataframe["rsi"] < 80) & # RSI不过度超买 (dataframe["bb_percent"] < 0.98) & # 不在布林带上轨附近 (dataframe["volatility"] < 0.05) # 波动率不过高 ) # 综合入场条件 dataframe.loc[ base_conditions & ma_golden_cross & volume_confirmation & price_position & trend_confirmation & additional_conditions, "enter_long" ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 出场条件：完全禁用，使用自定义退出逻辑 """ # 完全禁用populate_exit_trend，使用custom_exit return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ 智能动态止损 - 减少噪音触发，基于趋势强度和确认条件 """ if not self.dynamic_stoploss_enabled.value: return self.stoploss # 获取当前数据 dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) == 0: return self.stoploss last_candle = dataframe.iloc[-1].squeeze() # 获取关键指标 trend_strength = last_candle.get("trend_strength", 0) support_strength = last_candle.get("support_strength", 0) volume_ratio = last_candle.get("volume_ratio", 1.0) ema_fast = last_candle.get("ema_fast", 0) ema_slow = last_candle.get("ema_slow", 0) current_price = last_candle.get("close", 0) volatility = last_candle.get("volatility", 0.02) rsi = last_candle.get("rsi", 50) # 计算持仓时间 trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # 小时 # 如果有利润，保护利润 if current_profit > 0.05: # 有5%以上利润 return -0.02 # 2%止损（保护大部分利润） elif current_profit > 0.03: # 有3%以上利润 return -0.025 # 2.5%止损 elif current_profit > 0.02: # 有2%以上利润 return -0.03 # 3%止损 elif current_profit > 0.01: # 有1%以上利润 return -0.035 # 3.5%止损 # 根据趋势强度和确认条件调整止损 if (trend_strength >= self.trend_strength_threshold.value and support_strength >= 2 and current_price > ema_fast and volume_ratio >= 0.8 and rsi > 40): # 趋势强且确认 return -0.08 # 8%止损（趋势强，放宽止损） elif (trend_strength >= 2 and current_price > ema_slow and volume_ratio >= 0.7 and rsi > 35): # 趋势中等且确认 return -0.09 # 9%止损（趋势中等） elif (trend_strength >= 1 and current_price > ema_slow * 0.95 and volume_ratio >= 0.6): # 趋势弱但确认 return -0.095 # 9.5%止损 elif trade_duration > 24: # 持仓超过24小时 return -0.07 # 7%止损（长期持仓收紧） else: return self.stoploss # -0.1 (10%止损) def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]: """ 自定义退出逻辑 - 高胜率信号 + 利润保护 + 时间管理 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) == 0: return None last_candle = dataframe.iloc[-1].squeeze() # 获取关键指标 ema_fast = last_candle.get("ema_fast", 0) ema_slow = last_candle.get("ema_slow", 0) sma_support = last_candle.get("sma_support", 0) current_price = last_candle.get("close", 0) trend_strength = last_candle.get("trend_strength", 0) support_strength = last_candle.get("support_strength", 0) volume_ratio = last_candle.get("volume_ratio", 1.0) rsi = last_candle.get("rsi", 50) volatility = last_candle.get("volatility", 0.02) # 计算持仓时间 trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # 小时 # 1. 利润目标退出 - 高胜率信号 if current_profit >= 0.05: # 5%利润目标 return "profit_target_5%" elif current_profit >= 0.04: # 4%利润目标 return "profit_target_4%" elif current_profit >= 0.03: # 3%利润目标 return "profit_target_3%" # 2. 高胜率技术信号退出 # 均线死叉 + 多重确认 if (last_candle.get("ma_death_cross", False) and current_profit > 0.01 and # 有利润 current_price < ema_slow * 0.98 and # 价格明显低于慢线 volume_ratio < 0.8 and # 成交量萎缩 trend_strength <= 1): # 趋势转弱 return "high_confidence_death_cross" # 跌破重要支撑 + 多重确认 if (current_price < sma_support * 0.95 and # 跌破支撑5% current_profit > 0.005 and # 有微利 support_strength <= 0 and # 支撑转弱 volume_ratio < 0.7 and # 成交量萎缩 rsi < 45): # RSI转弱 return "high_confidence_support_break" # 3. 持仓时间管理 - 避免长期亏损 if trade_duration > self.max_hold_hours.value: # 超过最大持仓时间 if current_profit > 0: # 有利润 return "max_hold_time_profit" else: # 无利润，强制退出 return "max_hold_time_loss" # 4. 盈利单延长持仓时间 if current_profit > 0.02: # 有2%以上利润 extended_hold_time = self.max_hold_hours.value + self.profit_hold_extension.value if trade_duration > extended_hold_time: # 超过延长持仓时间 return "extended_hold_time_exit" # 5. 趋势转弱退出 - 高胜率信号 if (trend_strength <= 0 and current_profit > 0.01 and current_price < ema_fast * 0.97 and volume_ratio < 0.8 and rsi < 40): # 多重确认趋势转弱 return "trend_weak_exit" # 6. 波动率过高退出 if (volatility > 0.08 and current_profit > 0.015 and trade_duration > 6): # 高波动率且有利润 return "high_volatility_exit" # 7. 连续亏损强制退出 if (current_profit < -0.03 and trend_strength <= 0 and current_price < ema_slow * 0.95 and trade_duration > 12): # 12小时 return "force_exit_loss" return None def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ 动态仓位调整 - 根据信号强度调整仓位 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) == 0: return proposed_stake last_candle = dataframe.iloc[-1].squeeze() # 获取信号强度指标 trend_strength = last_candle.get("trend_strength", 0) volume_ratio = last_candle.get("volume_ratio", 1.0) support_strength = last_candle.get("support_strength", 0) trend_confirmation = last_candle.get("trend_confirmation", False) # 计算信号强度分数 signal_strength = ( trend_strength * 0.3 + # 趋势强度权重30% min(volume_ratio, 3.0) * 0.25 + # 成交量权重25% support_strength * 0.25 + # 支撑强度权重25% (trend_confirmation * 2) * 0.2 # 趋势确认权重20% ) # 根据信号强度调整仓位 if signal_strength >= 3.0: # 极强信号 return proposed_stake * 1.8 # 增加80%仓位 elif signal_strength >= 2.5: # 强信号 return proposed_stake * 1.5 # 增加50%仓位 elif signal_strength >= 2.0: # 中等信号 return proposed_stake * 1.2 # 增加20%仓位 elif signal_strength >= 1.5: # 弱信号 return proposed_stake * 1.0 # 标准仓位 else: # 很弱信号 return proposed_stake * 0.7 # 减少30%仓位 def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs) -> Optional[float]: """ 加仓逻辑 - 对高胜率盈利单进行加仓 """ dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if len(dataframe) == 0: return None last_candle = dataframe.iloc[-1].squeeze() # 获取当前指标 trend_strength = last_candle.get("trend_strength", 0) volume_ratio = last_candle.get("volume_ratio", 1.0) ema_fast = last_candle.get("ema_fast", 0) ema_slow = last_candle.get("ema_slow", 0) current_price = last_candle.get("close", 0) trend_confirmation = last_candle.get("trend_confirmation", False) support_strength = last_candle.get("support_strength", 0) # 计算持仓时间 trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # 小时 # 加仓条件1：有利润且趋势强劲 + 多重确认 if (current_profit > 0.02 and # 有2%以上利润 trend_strength >= 3 and # 趋势强劲 trend_confirmation and # 趋势确认 current_price > ema_fast and # 价格在快线之上 current_price > ema_slow and # 价格在慢线之上 volume_ratio >= 1.3 and # 成交量放大 support_strength >= 2 and # 支撑强劲 trade_duration < 12): # 持仓时间不太长 # 计算加仓金额（不超过原仓位的60%） stake_amount = trade.stake_amount * 0.6 return stake_amount # 加仓条件2：接近利润目标且趋势持续 elif (current_profit > 0.015 and # 有1.5%以上利润 trend_strength >= 2 and # 趋势持续 current_price > ema_fast * 0.99 and # 价格接近快线 volume_ratio >= 1.2 and # 成交量放大 support_strength >= 1 and # 支撑确认 trade_duration < 8): # 持仓时间较短 # 计算加仓金额（不超过原仓位的40%） stake_amount = trade.stake_amount * 0.4 return stake_amount # 加仓条件3：高胜率信号且微利 elif (current_profit > 0.01 and # 有1%以上利润 trend_strength >= 3 and # 趋势强劲 trend_confirmation and # 趋势确认 volume_ratio >= 1.5 and # 成交量大幅放大 current_price > ema_fast * 1.01 and # 价格明显高于快线 trade_duration < 6): # 持仓时间很短 # 计算加仓金额（不超过原仓位的30%） stake_amount = trade.stake_amount * 0.3 return stake_amount return None 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: """ 杠杆设置 - 基于信号强度的动态杠杆 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) == 0: return min(proposed_leverage, 2.0) last_candle = dataframe.iloc[-1].squeeze() # 获取信号强度 trend_strength = last_candle.get("trend_strength", 0) trend_confirmation = last_candle.get("trend_confirmation", False) # 根据信号强度调整杠杆 if trend_strength >= 4 and trend_confirmation: return min(proposed_leverage, 3.0) # 强信号，3倍杠杆 elif trend_strength >= 3: return min(proposed_leverage, 2.5) # 中等信号，2.5倍杠杆 else: return min(proposed_leverage, 2.0) # 弱信号，2倍杠杆