# source: https://raw.githubusercontent.com/FelixWayne0318/cryptosignal/8a1cf8ed0cabfcb6003b37ca4b9d5744d836348d/user_data/strategies/AlgVexStrategy.py # cs_ext/backtest/freqtrade_bridge.py """ Freqtrade 回测桥接层 - V8完整实现 使用 CryptoSignal 作为信号引擎的 Freqtrade 策略。 核心架构： - Freqtrade 提供历史K线（DataFrame） - 本策略将 DataFrame 转换为 CryptoSignal 的 klines 格式 - 调用四步决策系统分析，返回方向/强度/风险参数 - 支持做多/做空，自定义止损/止盈 - v8.0.6新增：自动导出诊断数据供Web Dashboard使用 配置文件：config/signal_thresholds.json """ import json import os import atexit from typing import Dict, Any, List, Optional from pandas import DataFrame import numpy as np # v7.4.9: Freqtrade 可选导入（graceful degradation） try: from freqtrade.strategy.interface import IStrategy FREQTRADE_AVAILABLE = True except ImportError: # Freqtrade 未安装时，创建一个占位基类 IStrategy = object FREQTRADE_AVAILABLE = False print( "[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: " "freqtrade未安装，策略类将无法使用。请安装: pip install freqtrade" ) from ats_core.env.bootstrap import bootstrap_env bootstrap_env() # 导入 CryptoSignal 分析函数 try: from ats_core.pipeline.analyze_symbol import analyze_symbol_with_preloaded_klines except ImportError: analyze_symbol_with_preloaded_klines = None print( "[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: " "无法导入 analyze_symbol_with_preloaded_klines，请根据实际项目结构修改导入路径。" ) # v8.0.6新增：导入诊断收集器用于导出数据 try: from ats_core.diagnostics.collector import get_collector DIAGNOSTICS_AVAILABLE = True except ImportError: get_collector = None DIAGNOSTICS_AVAILABLE = False print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] INFO: 诊断模块不可用，将不导出诊断数据") # v8.0.6新增：全局变量标记是否已注册atexit _ATEXIT_REGISTERED = False # v8.0.7.3新增：全局变量标记是否已导出诊断数据 _DIAGNOSTICS_EXPORTED = False def _export_diagnostics_on_exit(): """ 在进程退出时导出诊断数据 v8.0.6新增：确保诊断数据被保存供Web Dashboard使用 v8.0.7.3修改：检查是否已导出，避免重复 """ global _DIAGNOSTICS_EXPORTED # v8.0.7.3: 如果已通过显式导出完成，则跳过 if _DIAGNOSTICS_EXPORTED: print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] INFO: 诊断数据已通过显式导出完成，跳过atexit导出") return if not DIAGNOSTICS_AVAILABLE or get_collector is None: return try: collector = get_collector() counters = collector.get_counters() if counters.get("total_samples", 0) == 0: print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] INFO: 没有诊断数据需要导出") return # 导出到固定位置供Web Dashboard读取 export_dir = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "reports", "diagnostics", "latest" ) os.makedirs(export_dir, exist_ok=True) export_path = os.path.join(export_dir, "diagnostics.json") actual_path = collector.export(export_path) _DIAGNOSTICS_EXPORTED = True # v8.0.7.3: 标记已导出 print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] ✅ 诊断数据已导出(atexit): {actual_path}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 总样本数: {counters.get('total_samples', 0)}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 最终信号: {counters.get('final_accept', 0)}") except Exception as e: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 导出诊断数据失败: {e}") def load_config() -> Dict[str, Any]: """ 从 config/signal_thresholds.json 加载配置 """ config_paths = [ "config/signal_thresholds.json", "../config/signal_thresholds.json", os.path.join(os.path.dirname(__file__), "../../config/signal_thresholds.json"), ] for path in config_paths: if os.path.exists(path): with open(path, 'r', encoding='utf-8') as f: return json.load(f) print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 未找到配置文件，使用默认配置") return {} class Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255(IStrategy): """ 使用 CryptoSignal 四步决策系统的 Freqtrade 策略。 Features: - 调用四步决策系统（Direction → Timing → Risk → Quality） - 配置驱动，阈值从 signal_thresholds.json 读取 - 支持做多/做空 - 自定义止损/止盈基于四步系统计算 """ # ========== Freqtrade 基础配置 ========== # 从配置文件读取时间周期 timeframe = "1h" can_short: bool = True # ROI 设置（四步系统会计算具体止盈价位） minimal_roi = { "0": 10 # 基本禁用默认ROI，让四步系统管理 } # 止损设置（作为最大回撤保护，四步系统会计算具体止损） stoploss = -0.15 # 使用自定义止损 use_custom_stoploss = True # 仓位管理 position_adjustment_enable = False def __init__(self, config: dict) -> None: super().__init__(config) # 加载 CryptoSignal 配置 self._cs_config = load_config() self._backtest_config = self._cs_config.get("v8_integration", {}).get("backtest", {}) # 从配置读取回测引擎参数 engine_config = self._backtest_config.get("engine", {}) self._lookback_bars = engine_config.get("lookback_bars", 300) # 缓存分析结果 self._signal_cache: Dict[str, Dict] = {} # v7.4.9: 调试模式（打印详细日志） self._debug_mode = engine_config.get("debug_mode", True) # 默认开启用于问题排查 self._last_error = None # 用于避免重复错误日志 # v8.0.7新增：退出策略配置（从配置文件读取，零硬编码） exit_strategy_config = engine_config.get("exit_strategy", {}) self._exit_threshold_multiplier = exit_strategy_config.get("exit_threshold_multiplier", 0.7) self._exit_consecutive_bars = exit_strategy_config.get("exit_consecutive_bars", 2) # v8.0.7新增：追踪止损配置 trailing_config = engine_config.get("trailing_stop", {}) self._trailing_activation_pct = trailing_config.get("activation_profit_pct", 0.01) self._trailing_stop_pct = trailing_config.get("trailing_stop_pct", 0.003) self._breakeven_pct = trailing_config.get("breakeven_profit_pct", 0.005) # v8.0.6新增：注册退出时导出诊断数据 global _ATEXIT_REGISTERED if not _ATEXIT_REGISTERED and DIAGNOSTICS_AVAILABLE: atexit.register(_export_diagnostics_on_exit) _ATEXIT_REGISTERED = True print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] ✅ 已注册诊断数据导出钩子") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 初始化完成, lookback_bars={self._lookback_bars}, debug={self._debug_mode}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] v8.0.7退出策略: exit_multiplier={self._exit_threshold_multiplier}, consecutive_bars={self._exit_consecutive_bars}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] v8.0.7追踪止损: activation={self._trailing_activation_pct*100:.1f}%, trailing={self._trailing_stop_pct*100:.2f}%, breakeven={self._breakeven_pct*100:.1f}%") def _convert_df_to_klines(self, dataframe: DataFrame) -> List[Dict]: """ 将 Freqtrade DataFrame 转换为 CryptoSignal klines 格式 v7.4.8修复：四步系统期望字典格式，而非列表格式 CryptoSignal klines 格式: [{"timestamp": ..., "open": ..., "high": ..., ...}, ...] Args: dataframe: Freqtrade 的 OHLCV DataFrame Returns: klines 字典列表 """ klines = [] for idx in range(len(dataframe)): row = dataframe.iloc[idx] # 获取时间戳 # v7.4.10修复：处理Pandas Timestamp对象 if hasattr(row.name, 'timestamp'): open_time = int(row.name.timestamp() * 1000) else: date_val = row.get('date', 0) if hasattr(date_val, 'timestamp'): # Pandas Timestamp object open_time = int(date_val.timestamp() * 1000) elif isinstance(date_val, (int, float)): open_time = int(date_val) else: open_time = 0 # 构造 kline 字典格式（与内置BacktestEngine保持一致） kline = { "timestamp": open_time, "open": float(row['open']), "high": float(row['high']), "low": float(row['low']), "close": float(row['close']), "volume": float(row['volume']), "close_time": open_time + 3600000, # close_time (假设1小时) "quote_volume": 0.0, "trades": 0, "taker_buy_base": 0.0, "taker_buy_quote": 0.0 } klines.append(kline) return klines def _call_cryptosignal(self, dataframe: DataFrame, metadata: Dict[str, Any]) -> Dict[str, Any]: """ 调用 CryptoSignal 四步决策系统分析 Args: dataframe: 完整的历史K线 DataFrame metadata: Freqtrade 元数据（包含 pair） Returns: 分析结果字典： { "direction": "long" / "short" / "none", "final_strength": float, "probability": float, "entry_price": float, "stop_loss": float, "take_profit": float, "risk_reward_ratio": float } """ default_result = { "direction": "none", "final_strength": 0.0, "probability": 0.5, "entry_price": None, "stop_loss": None, "take_profit": None, "risk_reward_ratio": 0.0 } if analyze_symbol_with_preloaded_klines is None: return default_result pair = metadata.get("pair", "") # 将 Freqtrade pair 格式转换为 CryptoSignal symbol 格式 # Freqtrade: "BTC/USDT:USDT" -> CryptoSignal: "BTCUSDT" symbol = pair.replace("/", "").replace(":USDT", "") # 转换 DataFrame 为 klines 格式 k1h = self._convert_df_to_klines(dataframe) # 生成 4 小时 K 线（简化处理：每4根1小时K线合并） k4h = self._resample_to_4h(k1h) # v7.4.8修复：提供必要的价格数据（与内置回测引擎保持一致） mark_price = None spot_price = None if k1h: latest_kline = k1h[-1] mark_price = float(latest_kline["close"]) # close price（字典格式） spot_price = mark_price try: result = analyze_symbol_with_preloaded_klines( symbol=symbol, k1h=k1h, k4h=k4h, # v7.4.8修复：传递空列表而非None，避免因子计算错误 oi_data=[], # 空OI数据 spot_k1h=None, elite_meta=None, k15m=None, k1d=None, orderbook=None, mark_price=mark_price, # 标记价格 funding_rate=0.0, # 默认资金费率 spot_price=spot_price, # 现货价格 btc_klines=None, eth_klines=None, kline_cache=None, market_meta=None ) except Exception as e: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] analyze_symbol_with_preloaded_klines error: {e}") import traceback traceback.print_exc() return default_result # v7.4.9修复：检查分析结果是否成功 if not result.get("success", True): # 默认True保持向后兼容 error_msg = result.get("error", "Unknown error") # 只在首次出现时打印（避免日志洪水） if not hasattr(self, '_last_error') or self._last_error != error_msg: self._last_error = error_msg print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] {symbol} 分析失败: {error_msg}") return default_result # 解析分析结果 - 优先使用四步系统决策 four_step = result.get("four_step_decision", {}) # v7.4.9调试日志：追踪四步系统决策 if self._debug_mode: decision = four_step.get("decision", "N/A") reject_stage = four_step.get("reject_stage", "N/A") reject_reason = four_step.get("reject_reason", "N/A") step1 = four_step.get("step1_direction", {}) strength = step1.get("final_strength", 0) print(f"[DEBUG] {symbol} 四步决策: decision={decision}, " f"strength={strength:.1f}, reject={reject_stage}({reject_reason})") # 方案1：四步系统决策（优先） if four_step.get("decision") == "ACCEPT": action = four_step.get("action", "") direction = "long" if action == "LONG" else "short" if action == "SHORT" else "none" step1 = four_step.get("step1_direction", {}) final_strength = step1.get("final_strength", 0) # 使用四步系统的价格信息 entry_price = four_step.get("entry_price") stop_loss = four_step.get("stop_loss") take_profit = four_step.get("take_profit") risk_reward_ratio = four_step.get("risk_reward_ratio", 0) probability = min(0.5 + final_strength / 20, 0.95) # 强度范围约0-20 # 方案2：后备 - 旧系统判定 else: is_prime = result.get("is_prime", False) side_long = result.get("side_long", None) if not is_prime or side_long is None: return default_result direction = "long" if side_long else "short" final_strength = result.get("prime_strength", 0) entry_price = result.get("entry_price") stop_loss = result.get("stop_loss") take_profit = result.get("take_profit") risk_reward_ratio = result.get("risk_reward_ratio", 0) probability = min(0.5 + final_strength / 200, 0.95) return { "direction": direction, "final_strength": final_strength, "probability": probability, "entry_price": entry_price, "stop_loss": stop_loss, "take_profit": take_profit, "risk_reward_ratio": risk_reward_ratio, # 保存完整结果用于后续分析 "_full_result": result } def _resample_to_4h(self, k1h: List[Dict]) -> List[Dict]: """ 将1小时K线重采样为4小时K线 v7.4.8修复：使用字典格式（与_convert_df_to_klines保持一致） Args: k1h: 1小时K线字典列表 Returns: 4小时K线字典列表 """ k4h = [] for i in range(0, len(k1h) - 3, 4): batch = k1h[i:i+4] if len(batch) < 4: continue k4h_candle = { "timestamp": batch[0]["timestamp"], # open_time "open": batch[0]["open"], "high": max(k["high"] for k in batch), "low": min(k["low"] for k in batch), "close": batch[-1]["close"], "volume": sum(k["volume"] for k in batch), "close_time": batch[-1].get("close_time", batch[-1]["timestamp"] + 14400000), "quote_volume": sum(k.get("quote_volume", 0) for k in batch), "trades": sum(k.get("trades", 0) for k in batch), "taker_buy_base": 0.0, "taker_buy_quote": 0.0 } k4h.append(k4h_candle) return k4h def populate_indicators(self, dataframe: DataFrame, metadata: Dict[str, Any]) -> DataFrame: """ 计算指标（CryptoSignal 内部处理，这里仅做数据准备） """ # 添加辅助列用于信号追踪 dataframe['cs_signal_strength'] = 0.0 dataframe['cs_direction'] = 'none' return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: Dict[str, Any]) -> DataFrame: """ 生成入场信号 使用滚动窗口方式调用 CryptoSignal 分析，生成做多/做空信号。 """ dataframe["enter_long"] = 0 dataframe["enter_short"] = 0 pair = metadata.get("pair", "") # v7.4.9: 信号统计 stats = { "total_analyzed": 0, "direction_long": 0, "direction_short": 0, "direction_none": 0, "strength_above_threshold": 0, "enter_long": 0, "enter_short": 0, "max_strength": 0.0, "avg_strength": 0.0, "strength_sum": 0.0 } entry_threshold = self._backtest_config.get("engine", {}).get("entry_strength_threshold", 7.0) print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] {pair} 开始分析, 总行数={len(dataframe)}, lookback={self._lookback_bars}, 阈值={entry_threshold}") # 滚动窗口分析 for idx in range(self._lookback_bars, len(dataframe)): # 获取历史窗口 window = dataframe.iloc[idx - self._lookback_bars:idx + 1].copy() # 调用 CryptoSignal 分析 signal = self._call_cryptosignal(window, metadata) direction = signal["direction"] strength = signal["final_strength"] # 更新统计 stats["total_analyzed"] += 1 stats["strength_sum"] += strength if strength > stats["max_strength"]: stats["max_strength"] = strength if direction == "long": stats["direction_long"] += 1 elif direction == "short": stats["direction_short"] += 1 else: stats["direction_none"] += 1 # 缓存信号用于退出判断 cache_key = f"{pair}_{idx}" self._signal_cache[cache_key] = signal # 记录信号强度 dataframe.at[dataframe.index[idx], 'cs_signal_strength'] = strength dataframe.at[dataframe.index[idx], 'cs_direction'] = direction # 生成入场信号 # 注意：四步系统的final_strength范围约0-20，阈值应与四步系统配置一致 if direction == "long" and strength >= entry_threshold: dataframe.at[dataframe.index[idx], "enter_long"] = 1 stats["enter_long"] += 1 stats["strength_above_threshold"] += 1 elif direction == "short" and strength >= entry_threshold: dataframe.at[dataframe.index[idx], "enter_short"] = 1 stats["enter_short"] += 1 stats["strength_above_threshold"] += 1 # v7.4.9: 打印信号统计摘要 if stats["total_analyzed"] > 0: stats["avg_strength"] = stats["strength_sum"] / stats["total_analyzed"] print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] {pair} 信号统计:") print(f" - 总分析: {stats['total_analyzed']}") print(f" - 方向: Long={stats['direction_long']}, Short={stats['direction_short']}, None={stats['direction_none']}") print(f" - 强度: 最大={stats['max_strength']:.2f}, 平均={stats['avg_strength']:.2f}") print(f" - 入场信号: Long={stats['enter_long']}, Short={stats['enter_short']}, 总计={stats['enter_long']+stats['enter_short']}") # v8.0.7.3: 每个交易对处理完后增量导出诊断数据（防止进程退出时丢失） self._incremental_export_diagnostics(pair) return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: Dict[str, Any]) -> DataFrame: """ 生成退出信号 v8.0.7优化（基于回测数据驱动）： - 退出条件1: 信号反转（需连续N根K线确认，避免临时波动） - 退出条件2: 信号强度大幅衰减（使用配置化的阈值乘数） - 数据依据: exit_signal 75%亏损率，需更严格的退出确认 配置参数（从signal_thresholds.json读取）： - exit_threshold_multiplier: 退出阈值乘数（默认0.7） - exit_consecutive_bars: 连续确认K线数（默认2） """ dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 pair = metadata.get("pair", "") # v8.0.7: 从配置读取阈值（零硬编码） entry_threshold = self._backtest_config.get("engine", {}).get("entry_strength_threshold", 7.0) # v8.0.7: 使用配置化的退出阈值乘数（从0.5提升到0.7） exit_threshold = entry_threshold * self._exit_threshold_multiplier # v8.0.7: 连续确认K线数（避免临时波动导致过早退出） consecutive_required = self._exit_consecutive_bars if self._debug_mode: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] {pair} 退出配置: entry={entry_threshold}, exit={exit_threshold:.1f}, consecutive={consecutive_required}") # v8.0.7: 需要足够的历史数据进行连续确认 for idx in range(self._lookback_bars + consecutive_required, len(dataframe)): current_strength = dataframe.iloc[idx]['cs_signal_strength'] current_direction = dataframe.iloc[idx]['cs_direction'] # v8.0.7: 检查连续N根K线的方向（用于确认信号反转） reversal_confirmed_long_to_short = True reversal_confirmed_short_to_long = True strength_decay_confirmed = True for i in range(consecutive_required): check_idx = idx - i check_direction = dataframe.iloc[check_idx]['cs_direction'] check_strength = dataframe.iloc[check_idx]['cs_signal_strength'] # 检查是否连续short（做多退出条件） if check_direction != "short": reversal_confirmed_long_to_short = False # 检查是否连续long（做空退出条件） if check_direction != "long": reversal_confirmed_short_to_long = False # 检查强度是否连续低于阈值 if check_strength >= exit_threshold: strength_decay_confirmed = False # 获取入场时的方向（从更早的K线） entry_direction = dataframe.iloc[idx - consecutive_required]['cs_direction'] entry_strength = dataframe.iloc[idx - consecutive_required]['cs_signal_strength'] # 退出做多条件 if entry_direction == "long" and entry_strength >= entry_threshold: # 条件1: 信号连续反转为short（需连续N根确认） if reversal_confirmed_long_to_short: dataframe.at[dataframe.index[idx], "exit_long"] = 1 # 条件2: 强度连续衰减低于阈值 elif strength_decay_confirmed: dataframe.at[dataframe.index[idx], "exit_long"] = 1 # 退出做空条件 if entry_direction == "short" and entry_strength >= entry_threshold: # 条件1: 信号连续反转为long（需连续N根确认） if reversal_confirmed_short_to_long: dataframe.at[dataframe.index[idx], "exit_short"] = 1 # 条件2: 强度连续衰减低于阈值 elif strength_decay_confirmed: dataframe.at[dataframe.index[idx], "exit_short"] = 1 return dataframe def custom_stoploss(self, pair: str, trade, current_time, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> float: """ 自定义止损逻辑 v8.0.7优化（基于回测数据驱动）： - 追踪止损: trailing_stop 100%盈利率，应更早激活 - 保本止损: 盈利0.5%时启用，防止盈利变亏损 - 数据依据: trailing_stop_loss 12笔全盈利+$565.44 配置参数（从signal_thresholds.json读取）： - activation_profit_pct: 激活追踪止损的盈利阈值（默认1%） - trailing_stop_pct: 追踪止损百分比（默认0.3%） - breakeven_profit_pct: 保本止损激活阈值（默认0.5%） """ # 获取开仓时的信号缓存 entry_tag = trade.enter_tag or "" # v8.0.7: 使用配置化的追踪止损参数（零硬编码） # 阶段1: 盈利超过激活阈值时启动追踪止损 if current_profit > self._trailing_activation_pct: # 返回负数表示追踪止损百分比 return -self._trailing_stop_pct # 阶段2: 盈利超过保本阈值时启用保本止损 if current_profit > self._breakeven_pct: # 返回极小的止损，本质上是保本 return -0.001 return self.stoploss # 使用默认止损 def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ 确认交易入场 可以在这里添加额外的过滤逻辑 """ # 风险回报比检查已在四步系统Step3完成 # 这里仅记录日志，不重复拒绝已通过四步系统的信号 for key in reversed(list(self._signal_cache.keys())): if key.startswith(pair): signal = self._signal_cache[key] rr_ratio = signal.get("risk_reward_ratio", 0) if rr_ratio > 0 and rr_ratio < 1.5: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: {pair} RR={rr_ratio:.2f} < 1.5, 但已通过四步系统") break return True def confirm_trade_exit(self, pair: str, trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time, **kwargs) -> bool: """ 确认交易退出并关联交易结果到诊断数据 v8.0.7.7新增: 在交易退出时自动关联交易结果到诊断记录 解决问题: diagnostics accepted_records 中 trade_result 为 null Args: pair: 交易对 trade: 交易对象 order_type: 订单类型 amount: 数量 rate: 退出价格 time_in_force: 有效期 exit_reason: 退出原因 current_time: 当前时间 """ # v8.0.7.7: 关联交易结果到诊断数据 if DIAGNOSTICS_AVAILABLE and get_collector is not None: try: collector = get_collector() # 获取交易信息 # v8.0.7.10修复: 正确处理Freqtrade的pair格式 # Freqtrade格式: "BTC/USDT:USDT" → 需要提取 "BTCUSDT" symbol = pair.replace("/", "").split(":")[0] # "BTC/USDT:USDT" → "BTCUSDT" # v8.0.7.10修复: 时间戳格式标准化 # Freqtrade的trade.open_date_utc格式: "2025-11-13 13:00:00+00:00" # diagnostics记录的格式: "2025-11-13T12:00:00" # 需要转换为统一格式，并考虑Freqtrade的1小时延迟入场 if hasattr(trade, 'open_date_utc'): # trade.open_date_utc 是交易实际开仓时间 # 但信号是在前一根K线产生的，所以需要减去1小时来匹配 from datetime import timedelta signal_time = trade.open_date_utc - timedelta(hours=1) entry_time = signal_time.strftime("%Y-%m-%dT%H:%M:%S") # 统一格式，无时区 else: entry_time = str(trade.open_date) exit_time = current_time.isoformat() if hasattr(current_time, 'isoformat') else str(current_time) entry_price = float(trade.open_rate) exit_price = float(rate) # 计算PnL百分比 is_short = trade.is_short if hasattr(trade, 'is_short') else False if is_short: pnl_pct = (entry_price - exit_price) / entry_price * 100 else: pnl_pct = (exit_price - entry_price) / entry_price * 100 # 调用collector关联交易结果 success = collector.update_trade_result( symbol=symbol, entry_time=entry_time, exit_time=exit_time, entry_price=entry_price, exit_price=exit_price, pnl_pct=pnl_pct, exit_reason=exit_reason ) if success: if self._debug_mode: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] ✅ 交易结果已关联: {pair} pnl={pnl_pct:.2f}% reason={exit_reason}") # v8.0.7.8关键修复: 关联成功后立即重新导出诊断数据 # 原因: populate_entry_trend阶段导出的数据不包含trade_result # 必须在confirm_trade_exit后重新导出才能包含交易结果 self._export_diagnostics_with_trade_result() elif self._debug_mode: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] ⚠️ 未找到匹配的诊断记录: {pair} symbol={symbol} entry_time={entry_time}") except Exception as e: if self._debug_mode: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 关联交易结果失败: {e}") return True # 总是允许退出 def _export_diagnostics_with_trade_result(self) -> None: """ v8.0.7.8新增: 在交易结果关联后重新导出诊断数据 解决问题: confirm_trade_exit在populate_entry_trend之后执行, 但_incremental_export_diagnostics在populate_entry_trend中调用, 导致导出的文件不包含trade_result数据。 此方法在confirm_trade_exit成功关联trade_result后调用, 确保文件包含最新的交易结果。 """ if not DIAGNOSTICS_AVAILABLE or get_collector is None: return try: collector = get_collector() # 导出到固定位置供Web Dashboard读取 export_dir = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "reports", "diagnostics", "latest" ) os.makedirs(export_dir, exist_ok=True) export_path = os.path.join(export_dir, "diagnostics.json") collector.export(export_path) if self._debug_mode: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 📁 诊断数据已重新导出(含trade_result)") except Exception as e: if self._debug_mode: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 重新导出诊断数据失败: {e}") def _incremental_export_diagnostics(self, pair: str) -> None: """ 增量导出诊断数据（每处理完一个交易对后调用） v8.0.7.3新增：解决atexit在Freqtrade回测中不可靠的问题 Args: pair: 当前处理完的交易对 """ global _DIAGNOSTICS_EXPORTED if not DIAGNOSTICS_AVAILABLE or get_collector is None: return try: collector = get_collector() counters = collector.get_counters() total_samples = counters.get("total_samples", 0) # 只有在有新数据时才导出 if total_samples == 0: return # 导出到固定位置供Web Dashboard读取 export_dir = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "reports", "diagnostics", "latest" ) os.makedirs(export_dir, exist_ok=True) export_path = os.path.join(export_dir, "diagnostics.json") collector.export(export_path) _DIAGNOSTICS_EXPORTED = True # 每5个交易对打印一次摘要（避免日志过多） if not hasattr(self, '_diag_export_count'): self._diag_export_count = 0 self._diag_export_count += 1 if self._diag_export_count % 5 == 1: dq_count = getattr(collector, '_dq_rejection_count', 0) print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 📊 诊断数据已增量导出 (samples={total_samples}, dq_reject={dq_count})") except Exception as e: # 诊断导出失败不应影响主流程 if self._debug_mode: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 增量导出诊断数据失败: {e}") def export_diagnostics(self, force: bool = False) -> Optional[str]: """ 显式导出诊断数据 v8.0.7.3新增：提供不依赖atexit的显式导出机制 Args: force: 是否强制导出（忽略已导出标记） Returns: 导出文件路径，失败时返回None """ global _DIAGNOSTICS_EXPORTED if not DIAGNOSTICS_AVAILABLE or get_collector is None: print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 诊断模块不可用，无法导出") return None if _DIAGNOSTICS_EXPORTED and not force: print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] INFO: 诊断数据已导出，跳过（使用force=True强制重新导出）") return None try: collector = get_collector() counters = collector.get_counters() if counters.get("total_samples", 0) == 0: print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] INFO: 没有诊断数据需要导出") return None # 导出到固定位置供Web Dashboard读取 export_dir = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "reports", "diagnostics", "latest" ) os.makedirs(export_dir, exist_ok=True) export_path = os.path.join(export_dir, "diagnostics.json") actual_path = collector.export(export_path) _DIAGNOSTICS_EXPORTED = True total_samples = counters.get('total_samples', 0) final_accept = counters.get('final_accept', 0) dq_rejections = getattr(collector, '_dq_rejection_count', 0) print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] ✅ 诊断数据已显式导出: {actual_path}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 总样本数: {total_samples}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 数据质量拒绝: {dq_rejections}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 最终信号: {final_accept}") return actual_path except Exception as e: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 显式导出诊断数据失败: {e}") import traceback traceback.print_exc() return None @classmethod def force_export_diagnostics(cls) -> Optional[str]: """ 类方法：强制导出诊断数据（可在外部直接调用） v8.0.7.3新增：供回测脚本调用 用法: from user_data.strategies.Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255 import Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255 Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255.force_export_diagnostics() """ global _DIAGNOSTICS_EXPORTED if not DIAGNOSTICS_AVAILABLE or get_collector is None: print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 诊断模块不可用，无法导出") return None try: collector = get_collector() counters = collector.get_counters() if counters.get("total_samples", 0) == 0: print("[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] INFO: 没有诊断数据需要导出") return None # 导出到固定位置供Web Dashboard读取 export_dir = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "reports", "diagnostics", "latest" ) os.makedirs(export_dir, exist_ok=True) export_path = os.path.join(export_dir, "diagnostics.json") actual_path = collector.export(export_path) _DIAGNOSTICS_EXPORTED = True total_samples = counters.get('total_samples', 0) final_accept = counters.get('final_accept', 0) dq_rejections = getattr(collector, '_dq_rejection_count', 0) print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] ✅ 诊断数据已强制导出: {actual_path}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 总样本数: {total_samples}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 数据质量拒绝: {dq_rejections}") print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] 最终信号: {final_accept}") return actual_path except Exception as e: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_174255] WARNING: 强制导出诊断数据失败: {e}") import traceback traceback.print_exc() return None