# source: https://raw.githubusercontent.com/FelixWayne0318/cryptosignal/4c7ad17b9279f4173463d9bfd2949e052c992deb/user_data/strategies/AlgVexStrategy.py # cs_ext/backtest/freqtrade_bridge.py """ Freqtrade 回测桥接层 - V8完整实现 使用 CryptoSignal 作为信号引擎的 Freqtrade 策略。 核心架构： - Freqtrade 提供历史K线（DataFrame） - 本策略将 DataFrame 转换为 CryptoSignal 的 klines 格式 - 调用四步决策系统分析，返回方向/强度/风险参数 - 支持做多/做空，自定义止损/止盈 配置文件：config/signal_thresholds.json """ import json import os 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_145935] 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_145935] WARNING: " "无法导入 analyze_symbol_with_preloaded_klines，请根据实际项目结构修改导入路径。" ) # v8.0.4: 导入诊断模块 try: from ats_core.diagnostics.collector import get_collector, reset_collector from ats_core.diagnostics.reporter import DiagnosticsReporter DIAGNOSTICS_AVAILABLE = True except ImportError: DIAGNOSTICS_AVAILABLE = False get_collector = None reset_collector = None DiagnosticsReporter = None 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_145935] WARNING: 未找到配置文件，使用默认配置") return {} class Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935(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 # 用于避免重复错误日志 print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] 初始化完成, lookback_bars={self._lookback_bars}, debug={self._debug_mode}") 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_145935] 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_145935] {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 分析，生成做多/做空信号。 v8.0.4: 集成诊断数据收集 """ dataframe["enter_long"] = 0 dataframe["enter_short"] = 0 dataframe["enter_long_tag"] = "" dataframe["enter_short_tag"] = "" pair = metadata.get("pair", "") # v8.0.4: 在回测开始时重置诊断收集器 if DIAGNOSTICS_AVAILABLE and reset_collector is not None: try: reset_collector() print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] 诊断收集器已重置 ({pair})") except Exception as e: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] 诊断收集器重置失败: {e}") # 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_145935] {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 # 生成详细的 enter_tag enter_tag = f"4S_L_s{strength:.1f}" if signal.get("risk_reward_ratio"): enter_tag += f"_rr{signal['risk_reward_ratio']:.1f}" dataframe.at[dataframe.index[idx], "enter_long_tag"] = enter_tag stats["enter_long"] += 1 stats["strength_above_threshold"] += 1 elif direction == "short" and strength >= entry_threshold: dataframe.at[dataframe.index[idx], "enter_short"] = 1 # 生成详细的 enter_tag enter_tag = f"4S_S_s{strength:.1f}" if signal.get("risk_reward_ratio"): enter_tag += f"_rr{signal['risk_reward_ratio']:.1f}" dataframe.at[dataframe.index[idx], "enter_short_tag"] = enter_tag 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_145935] {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.4: 打印诊断统计摘要并导出报告 if DIAGNOSTICS_AVAILABLE and get_collector is not None: try: collector = get_collector() if collector.enabled: counters = collector.get_counters() print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] {pair} 四步决策诊断统计:") print(f" - 总决策点: {counters.get('total_samples', 0)}") print(f" - Step1通过: {counters.get('step1_pass', 0)}, 拒绝: {counters.get('step1_reject', 0)}") print(f" - Step2通过: {counters.get('step2_pass', 0)}, 拒绝: {counters.get('step2_reject', 0)}") print(f" - Step3通过: {counters.get('step3_pass', 0)}, 拒绝: {counters.get('step3_reject', 0)}") print(f" - Step4通过: {counters.get('step4_pass', 0)}, 拒绝: {counters.get('step4_reject', 0)}") print(f" - 最终信号: {counters.get('final_accept', 0)}") # v8.0.4: 导出诊断报告到文件（供主进程读取） if counters.get('total_samples', 0) > 0 and DiagnosticsReporter is not None: try: reporter = DiagnosticsReporter(collector) # 导出JSON完整报告 export_path = reporter.export_report() print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] 诊断报告已导出: {export_path}") # v8.0.5: 导出JSONL原始记录 jsonl_path = reporter.export_jsonl() if jsonl_path: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] JSONL原始记录已导出: {jsonl_path}") except Exception as export_err: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] 诊断报告导出失败: {export_err}") except Exception as e: print(f"[Github_FelixWayne0318_cryptosignal__AlgVexStrategy__20251126_145935] 诊断统计打印失败: {e}") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: Dict[str, Any]) -> DataFrame: """ 生成退出信号 v8.0.4 优化：减少频繁退出，让追踪止损和止盈发挥作用 退出条件： 1. 强信号反转（反向信号强度超过阈值） 2. 连续2根K线信号衰减到极低水平 """ dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 pair = metadata.get("pair", "") # 获取阈值配置 entry_threshold = self._backtest_config.get("engine", {}).get("entry_strength_threshold", 7.0) # v8.0.4: 提高退出门槛，只有强反转信号才退出 reversal_threshold = entry_threshold * 0.8 # 反转信号必须达到入场阈值的80% decay_threshold = entry_threshold * 0.2 # 极低衰减阈值 for idx in range(self._lookback_bars + 2, len(dataframe)): current_strength = dataframe.iloc[idx]['cs_signal_strength'] current_direction = dataframe.iloc[idx]['cs_direction'] prev_strength = dataframe.iloc[idx - 1]['cs_signal_strength'] prev_direction = dataframe.iloc[idx - 1]['cs_direction'] prev2_strength = dataframe.iloc[idx - 2]['cs_signal_strength'] # 退出做多条件 if prev_direction == "long": # 强反转信号：空头信号强度超过阈值 if current_direction == "short" and current_strength >= reversal_threshold: dataframe.at[dataframe.index[idx], "exit_long"] = 1 # 连续2根K线信号极度衰减 elif current_strength < decay_threshold and prev_strength < decay_threshold: dataframe.at[dataframe.index[idx], "exit_long"] = 1 # 退出做空条件 if prev_direction == "short": # 强反转信号：多头信号强度超过阈值 if current_direction == "long" and current_strength >= reversal_threshold: dataframe.at[dataframe.index[idx], "exit_short"] = 1 # 连续2根K线信号极度衰减 elif current_strength < decay_threshold and prev_strength < decay_threshold: 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.4 渐进式追踪止损： - 盈利 >3%: 保护1.5%利润 - 盈利 >2%: 保护1%利润 - 盈利 >1%: 保护0.5%利润 - 盈利 >0.5%: 保本 - 其他: 使用默认止损 """ # 渐进式追踪止损 if current_profit > 0.03: # 盈利超过3%，保护1.5%利润 return -0.015 elif current_profit > 0.02: # 盈利超过2%，保护1%利润 return -0.01 elif current_profit > 0.01: # 盈利超过1%，保护0.5%利润 return -0.005 elif current_profit > 0.005: # 盈利超过0.5%，保本 return -0.001 # 默认使用较紧的止损保护资金 return -0.04 # 4%止损，比默认15%更紧 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_145935] WARNING: {pair} RR={rr_ratio:.2f} < 1.5, 但已通过四步系统") break return True