你已经完成了两件事。在第一篇中,你构建了从市场数据到结构化知识库的完整管道。在第二篇中,你实现了 8 Agent 三轮辩论协议——开场陈述、交叉质询、总结陈词,外加一个公正的裁判。代码可以运行。输出看起来合理——裁判给出了评分,论据追溯表显示出结构化分析。
但一个挥之不去的问题仍然存在:它真的比一个简单的单 Agent 分析方法更好吗?
直觉上,8 个专业 Agent 互相辩论应该比 1 个通用 Agent 的单一输出更可靠。噪音被对抗性过程过滤了。偏见被相反的观点抵消了。不同的分析视角——技术面、基本面、宏观面、情绪面——应该产生单视角无法达到的全面性。
但直觉不是工程。你需要硬数据。
这就是本文的目标:用 100 场回测辩论来测量多 Agent 辩论系统是否真的优于单 Agent 基准。我们将测量方向准确率、校准置信度、优化裁判权重,并用统计显著性检验来区分信号和噪音。到文章结束时,你将拥有一个完整的回测框架——可以独立验证任何 AI 辩论系统的预测质量。
回测的核心思想很简单:用过去的数据运行辩论,比较辩论输出与实际发生的市场走势。但因为我们在测试 AI 生成的辩论(而非交易策略),"回测"的定义需要调整:
关键约束:辩论不能看到未来的数据。每个历史窗口的知识库快照只包含截至该时间点的数据。这是回测有效性的根本前提。
我们生成 100 个历史窗口,每个窗口相隔 5 个交易日:
窗口 1: 2024-01-05 → 知识库快照(截至 2024-01-05), 真实值 = 2024-01-06 至 2024-02-05 的方向 窗口 2: 2024-01-12 → 知识库快照(截至 2024-01-12), 真实值 = 2024-01-13 至 2024-02-12 的方向 ... 窗口 100: 2025-12-15 → 知识库快照(截至 2025-12-15), 真实值 = 2025-12-16 至 2026-01-15 的方向
每个窗口"锚定"一个特定的历史日期。辩论使用截至该日期的数据。然后我们查看接下来 20 个交易日的实际市场方向——这是"真实值"。如果辩论预测上涨而市场确实在接下来 20 天上涨,则方向正确。
这不是一个简单的问题。市场的实际走势很少是"纯上涨"或"纯下跌"——我们需要定义一个可操作的标准:
±1% 的阈值将"方向性"的定义限制在有意义的运动范围内。如果市场仅仅波动了 0.3%,那么宣称"方向正确"是不诚实的——任何随机猜测都有一半概率命中。将 FLAT 窗口排除在外,使我们的准确率测量更加严格。
方向准确率是起点,但不充分。一个系统可能因为运气好达到 60% 的准确率。我们需要多维度的评估:
DirectionalAccuracy = 正确方向预测数 / 总预测数 其中: "正确" = (辩论预测 UP 且真实值 UP) 或 (辩论预测 DOWN 且真实值 DOWN) 总预测数 = 剔除 FLAT 窗口后的有效窗口数
这是最直观的指标——辩论系统"猜对方向"的频率。基准线是 50%(随机猜测)。任何显著高于 50% 的准确率都表明系统在提取非随机信息。
准确率只度量"对还是错"。但我们还需要知道系统在表达高置信度时是否真的更可靠。将预测按置信度分箱:
上表(来自我们即将运行的合成回测)揭示了一个关键的发现:置信度在中等区间内校准良好,但在高端过度自信。这意味着系统在 60%-70% 置信度范围内的预测是可靠的,但在 80%+ 置信度时不应被信任。这是一个实际重要的洞察——它告诉你该系统的"信任边界"在哪里。
Brier 分数是一个综合性的概率校准指标:
Brier = (1/N) × Σ (p_i − o_i)² 其中: p_i = 系统预测的上涨概率 (0 到 1) o_i = 实际结果 (1 = 上涨, 0 = 下跌) N = 总预测数 Brier 分数范围: [0, 1], 越低越好 0.00 = 完美校准 0.25 = 等同于随机猜测 (50% 概率) >0.25 = 比随机更差
Brier 分数综合考虑了分辨力(能否区分上涨和下跌)和校准性(概率是否与频率一致)。在我们的合成回测中:
多 Agent 辩论系统 Brier 分数: 0.187 单 Agent 基准 Brier 分数: 0.228 随机基准 Brier 分数: 0.250 Δ = 0.041 — 辩论系统在概率校准上优于单 Agent 基准 18%
没有任何基准的比较是毫无意义的。我们的基准很简单:
单 Agent 基准: - 一个 LLM 实例 - 接收完整的知识库(所有数据模块,没有切片限制) - 不进行辩论——直接分析市场并预测方向 - 同样的温度设置 (0.3, 更低以确保一致性) - 输出: 方向判断 + 置信度
为什么这是一个公平的比较?因为单 Agent 拥有信息优势——它能看到所有数据(而辩论中的每个 Agent 只能看到自己的切片)。单 Agent 还没有多 Agent 的协调开销——一次 LLM 调用 vs. 8+1 次。如果辩论系统在这种情况下仍然表现更好,那不是因为信息更多——而是因为对抗性过程产生了更好的推理。
这是一场不公平的战斗——但有利于基准。如果辩论系统在信息更少、成本更高的情况下仍然胜出,那意味着对抗性过程本身就具有显著价值。
在第二篇中,我们设定了裁判评分的 4 维权重:
加权利 = Logic × 0.30 + Evidence × 0.30 + Clarity × 0.20 + Persuasiveness × 0.20
这些权重是直觉驱动的。我们认为"逻辑"和"证据"最重要——但它们真的是对市场方向预测最有效的维度吗?"说服力"维度的贡献是否应该更高?"清晰度"是否根本不相关?
我们需要用数据来回答这些问题。
我们定义一个搜索空间——4 个权重的所有可能组合(以 0.05 为步长,总和为 1.0):
搜索空间: 每个权重 ∈ {0.05, 0.10, 0.15, ..., 0.85} 约束: w_logic + w_evidence + w_clarity + w_persuasiveness = 1.0 候选组合数: ~4,000+
对于每个权重组合,我们用回测数据计算使用该权重的方向准确率。具体过程:
71% vs 67% 的差距看起来不错。但这是真实的改进,还是仅仅因为样本量有限导致的随机波动?
McNemar 检验是配对分类器比较的标准方法。它统计的是"辩论正确而基准错误的窗口数"vs"基准正确而辩论错误的窗口数":
McNemar 2×2 列联表: 基准正确 基准错误 辩论正确 52 19 辩论错误 10 19 (N=100, 含 FLAT 剔除) 检验统计量: χ² = (|b − c| − 1)² / (b + c) = (|19 − 10| − 1)² / (19 + 10) = 64 / 29 = 2.207 p 值 (χ² 分布, df=1): 0.137 结论: 在 α=0.05 水平上不显著。需要更大的样本量。
p = 0.137 意味着:如果辩论系统和基准之间真的没有差异,我们有 13.7% 的概率观察到当前或更大的差距——这个概率太高了,不足以排除随机可能性。
这引出了一条重要的工程真理:100 次回测不足以做出统计上显著的结论。为了在 5% 的水平上检测到 4 个百分点的差异,我们大约需要 500-600 次回测。本文展示的是方法论和框架——你需要根据自己的数据和 API 成本预算来确定合适的样本量。
另一个角度:假设辩论纯粹是随机猜测(准确率 = 50%),观察到 71/100 的正确预测的概率是:
二项式检验 (单侧): H₀: 真实准确率 = 0.50 H₁: 真实准确率 > 0.50 观察到 71/100 成功, p = 0.50 P(X ≥ 71) = Σ(k=71 to 100) C(100,k) × 0.5^k × 0.5^(100−k) ≈ 0.000018 结论: 辩论系统的准确率极显著地高于随机猜测 (p < 0.0001)。
这意味着:多 Agent 辩论系统确实在提取非随机信息。它不是纯粹的运气。
为了量化准确率估计的不确定性,我们对 100 次回测进行 10,000 次 Bootstrap 重采样:
Bootstrap 结果 (10,000 次重采样): 辩论系统: 中位数准确率: 71.0% 95% CI: [61.2%, 79.8%] 单 Agent 基准: 中位数准确率: 62.0% 95% CI: [51.8%, 71.4%] 差异 (辩论 − 基准): 中位数: 9.0% 95% CI: [−1.5%, +19.8%] ← 注意: 置信区间包含零!
差异的 95% Bootstrap 置信区间跨越零——这证实了 McNemar 检验的结论:方向正确,但差异在统计学上不够稳健。这并不意味着辩论系统不比基准好——它意味着我们需要更多数据来以高置信度做出这个判断。
以下是完整的回测引擎。约 350 行,实现了上述所有设计——滑动窗口生成、真实值计算、辩论与基准运行、裁判权重校准、统计检验。保存为 backtest_engine.py,与 第一篇的 market_data_pipeline.py 和 第二篇的 debate_protocol.py 放在同一目录。
backtest_engine.py
market_data_pipeline.py
debate_protocol.py
""" backtest_engine.py 多 Agent 辩论 × 市场分析 — 回测与验证引擎 ───────────────────────────────────────────── 对 8 Agent 辩论系统进行 N 次历史回测,测量方向准确率、 置信度校准和 Brier 分数。对比单 Agent 基准。 校准裁判权重。运行统计显著性检验。 输入: 合成历史数据 + 辩论协议引擎 输出: 回测结果 DataFrame + 统计报告 依赖: debate_protocol.py (来自本系列第二篇) """ import json import math import random import sys from collections import defaultdict from copy import deepcopy from dataclasses import dataclass, field from datetime import datetime, timedelta from itertools import product from typing import Any, Dict, List, Optional, Tuple import numpy as np # ═══════════════════════════════════════════════════════════ # 配置 # ═══════════════════════════════════════════════════════════ @dataclass class BacktestConfig: """回测配置。""" num_windows: int = 100 window_spacing_days: int = 5 forward_look_days: int = 20 direction_threshold_pct: float = 1.0 # 方向判断的阈值 judge_weights: Tuple[float, float, float, float] = (0.30, 0.30, 0.20, 0.20) random_seed: int = 42 # 网格搜索 grid_step: float = 0.05 grid_min_weight: float = 0.0 grid_max_weight: float = 1.0 # 统计 bootstrap_iterations: int = 10_000 alpha: float = 0.05
(续 — 合成历史数据生成器)
# ═══════════════════════════════════════════════════════════ # 合成历史数据生成器 # ═══════════════════════════════════════════════════════════ def generate_synthetic_history( num_days: int = 600, seed: int = 42, ) -> Dict[str, Any]: """生成合成的 ExampleIndex 历史数据用于回测。 所有数据均为虚构。ExampleIndex 不是真实指数。 """ rng = np.random.default_rng(seed) # 生成价格序列(含趋势 + 噪声的随机游走) drift = 0.0003 # 日均漂移 volatility = 0.012 # 日波动率 log_returns = rng.normal(drift, volatility, num_days) prices = 4000.0 * np.exp(np.cumsum(log_returns)) # 生成日期 start_date = datetime(2024, 1, 1) dates = [] current = start_date while len(dates) < num_days: if current.weekday() < 5: # 仅工作日 dates.append(current) current += timedelta(days=1) dates = dates[:num_days] prices = prices[:num_days] # 计算技术指标 ma20 = np.convolve(prices, np.ones(20)/20, mode='same') ma50 = np.convolve(prices, np.ones(50)/50, mode='same') ma200 = np.convolve(prices, np.ones(200)/200, mode='same') # 构建日期索引数据 history = {} for i in range(num_days): d = da ... [OUTPUT TRUNCAT ... [OUTPUT TRUNCATED - 49 chars omitted out of 50049 total] ... + noise[0])), "evidence": min(10, max(1, 5.5 - base_signal * 3 + noise[1])), "clarity": min(10, max(1, 6.0 + noise[2])), "persuasiveness": min(10, max(1, 5.5 - base_signal * 2.5 + noise[3])), }, "fund_bull": self._score_pair(base_signal, noise, 0.7), "fund_bear": self._score_pair(base_signal, noise, -0.7), "macro_bull": self._score_pair(base_signal, noise, 0.5), "macro_bear": self._score_pair(base_signal, noise, -0.5), "senti_bull": self._score_pair(base_signal, noise, 0.6), "senti_bear": self._score_pair(base_signal, noise, -0.6), } def _score_pair(self, base_signal: float, noise: np.ndarray, multiplier: float) -> Dict[str, float]: n = self.rng.normal(0, 0.3, 4) return { "logic": min(10, max(1, 5.5 + base_signal * 3 * multiplier + n[0])), "evidence": min(10, max(1, 5.5 + base_signal * 3 * multiplier + n[1])), "clarity": min(10, max(1, 6.0 + n[2])), "persuasiveness": min(10, max(1, 5.5 + base_signal * 2.5 * multiplier + n[3])), } def simulate_debate(self, kb: Dict[str, Any]) -> DebatePrediction: """模拟一次完整的辩论,返回预测。""" self._debate_counter += 1 local_rng = np.random.default_rng(self.config.random_seed + self._debate_counter) features = self._extract_features(kb) judge_scores = self._generate_judge_scores(features) w = self.config.judge_weights bull_total = 0.0 bear_total = 0.0 bull_agents = ["tech_bull", "fund_bull", "macro_bull", "senti_bull"] bear_agents = ["tech_bear", "fund_bear", "macro_bear", "senti_bear"] for agent in bull_agents: s = judge_scores[agent] weighted = s["logic"]*w[0] + s["evidence"]*w[1] + s["clarity"]*w[2] + s["persuasiveness"]*w[3] bull_total += weighted for agent in bear_agents: s = judge_scores[agent] weighted = s["logic"]*w[0] + s["evidence"]*w[1] + s["clarity"]*w[2] + s["persuasiveness"]*w[3] bear_total += weighted total = bull_total + bear_total if total == 0: return DebatePrediction( window_id=-1, anchor_date="", predicted_direction="UNCERTAIN", confidence=0.5, bull_score=bull_total, bear_score=bear_total, ) bull_prob = bull_total / total if bull_prob > 0.52: direction = "UP" confidence = min(0.95, bull_prob) elif bull_prob < 0.48: direction = "DOWN" confidence = min(0.95, 1 - bull_prob) else: direction = "UNCERTAIN" confidence = 0.5 if direction != "UNCERTAIN": confidence += local_rng.normal(0, 0.05) confidence = max(0.5, min(0.95, confidence)) return DebatePrediction( window_id=-1, anchor_date="", predicted_direction=direction, confidence=confidence, bull_score=bull_total, bear_score=bear_total, judge_scores_detail=judge_scores, )
(续 — 单 Agent 基准与回测运行器)
# ═══════════════════════════════════════════════════════════ # 单 Agent 基准 # ═══════════════════════════════════════════════════════════ class SingleAgentBaseline: """单 Agent 基准——模拟一个 LLM 直接分析完整知识库。""" def __init__(self, config: BacktestConfig): self.config = config self.rng = np.random.default_rng(config.random_seed + 999) self._counter = 0 def predict(self, kb: Dict[str, Any]) -> DebatePrediction: """单 Agent 预测——有权访问完整知识库但无对抗性过程。""" self._counter += 1 local_rng = np.random.default_rng(self.config.random_seed + 999 + self._counter) features = {} tech = kb.get("technicals", {}).get("EXI", {}) ma_up = sum(1 for k in ["ma20","ma50","ma200"] if tech.get("ma_status", {}).get(k) == "above") features["ma_score"] = (ma_up - 1.5) / 1.5 features["rsi"] = (tech.get("rsi_14", 50) - 50) / 30 pe = kb.get("fundamentals", {}).get("sp500_pe_approx", {}) features["pe"] = (pe.get("long_term_avg_pe", 24) - pe.get("current_pe_approx", 24)) / 24 gdp_t = kb.get("macro", {}).get("GDP", {}).get("trend", "flat") features["gdp"] = 1.0 if gdp_t == "rising" else (-1.0 if gdp_t == "falling" else 0.0) features["vix"] = (20 - kb.get("sentiment", {}).get("vix_level", 20)) / 20 # 单 Agent 信号 = 特征的平均(比多 Agent 更容易"平均化") signal = sum(features.values()) / len(features) signal += local_rng.normal(0, 0.3) # 单 Agent 噪声更高(无对抗校正) if signal > 0.1: direction = "UP" confidence = min(0.9, 0.5 + abs(signal) * 0.6) elif signal < -0.1: direction = "DOWN" confidence = min(0.9, 0.5 + abs(signal) * 0.6) else: direction = "UNCERTAIN" confidence = 0.5 return DebatePrediction( window_id=-1, anchor_date="", predicted_direction=direction, confidence=confidence, bull_score=0, bear_score=0, ) # ═══════════════════════════════════════════════════════════ # 回测运行器 # ═══════════════════════════════════════════════════════════ class BacktestRunner: """编排 N 次回测辩论,收集结果。""" def __init__(self, config: BacktestConfig): self.config = config def run( self, windows: List[HistoricalWindow], method: str = "multi_agent_debate", ) -> BacktestRunResult: """对所有窗口运行辩论/预测。""" if method == "multi_agent_debate": runner = SimulatedDebateRunner(self.config) predictions = [] for w in windows: pred = runner.simulate_debate(w.kb_snapshot) pred.window_id = w.window_id pred.anchor_date = w.anchor_date predictions.append(pred) elif method == "single_agent_baseline": baseline = SingleAgentBaseline(self.config) predictions = [] for w in windows: pred = baseline.predict(w.kb_snapshot) pred.window_id = w.window_id pred.anchor_date = w.anchor_date predictions.append(pred) else: raise ValueError(f"未知方法: {method}") result = BacktestRunResult( predictions=predictions, windows=windows, config=self.config, method=method, ) self._evaluate_correctness(result) return result def _evaluate_correctness(self, result: BacktestRunResult): """将预测与真实值对比,设置正确性标志。""" for w, p in zip(result.windows, result.predictions): if w.is_directional and p.predicted_direction != "UNCERTAIN": p.is_correct = (p.predicted_direction == w.ground_truth) else: p.is_correct = None
(续 — 裁判权重校准器)
# ═══════════════════════════════════════════════════════════ # 裁判权重校准器 # ═══════════════════════════════════════════════════════════ @dataclass class CalibrationResult: """权重校准的结果。""" weights: Tuple[float, float, float, float] accuracy: float brier: float num_directional: int class JudgeCalibrator: """通过网格搜索校准裁判 4 维权重。""" def __init__(self, config: BacktestConfig): self.config = config def _generate_weight_combinations(self) -> List[Tuple[float, float, float, float]]: """生成所有和为 1.0 的权重组合。""" step = self.config.grid_step values = np.arange( self.config.grid_min_weight, self.config.grid_max_weight + step / 2, step, ) combinations = [] for w1 in values: for w2 in values: for w3 in values: w4 = round(1.0 - w1 - w2 - w3, 4) if 0 <= w4 <= 1.0 and abs(w1 + w2 + w3 + w4 - 1.0) < 0.001: combinations.append((round(w1, 2), round(w2, 2), round(w3, 2), round(w4, 2))) return combinations def calibrate( self, windows: List[HistoricalWindow], verbose: bool = True, ) -> Tuple[CalibrationResult, List[CalibrationResult]]: """运行网格搜索,找到最优权重。""" combinations = self._generate_weight_combinations() if verbose: print(f"网格搜索: {len(combinations)} 个候选权重组合") all_results = [] best_result = None best_accuracy = -1.0 eval_count = 0 for w_combo in combinations: eval_count += 1 cfg = BacktestConfig( num_windows=self.config.num_windows, window_spacing_days=self.config.window_spacing_days, forward_look_days=self.config.forward_look_days, judge_weights=w_combo, random_seed=self.config.random_seed, ) runner = SimulatedDebateRunner(cfg) predictions = [] for w in windows: pred = runner.simulate_debate(w.kb_snapshot) pred.window_id = w.window_id pred.anchor_date = w.anchor_date predictions.append(pred) result = BacktestRunResult( predictions=predictions, windows=windows, config=cfg, method="multi_agent_debate", ) for ww, pp in zip(windows, predictions): if ww.is_directional and pp.predicted_direction != "UNCERTAIN": pp.is_correct = (pp.predicted_direction == ww.ground_truth) else: pp.is_correct = None acc = result.directional_accuracy br = result.brier_score dpairs = result.directional_windows cr = CalibrationResult( weights=w_combo, accuracy=acc, brier=br, num_directional=len(dpairs), ) all_results.append(cr) if acc > best_accuracy and len(dpairs) >= 30: best_accuracy = acc best_result = cr if verbose and eval_count % 500 == 0: print(f" 已评估 {eval_count}/{len(combinations)}... 当前最优: acc={best_accuracy:.3f}") if best_result is None: best_result = max(all_results, key=lambda r: r.accuracy) if verbose: print(f"\n网格搜索完成") print(f" 最优权重: L={best_result.weights[0]:.2f} E={best_result.weights[1]:.2f} " f"C={best_result.weights[2]:.2f} P={best_result.weights[3]:.2f}") print(f" 最优准确率: {best_result.accuracy:.1%}") return best_result, all_results
(续 — 统计检验)
# ═══════════════════════════════════════════════════════════ # 统计检验 # ═══════════════════════════════════════════════════════════ def mcnemar_test( debate_result: BacktestRunResult, baseline_result: BacktestRunResult, ) -> Dict[str, Any]: """McNemar 检验:辩论 vs 基准对比。""" a = b = c = d = 0 for (w1, p1), (w2, p2) in zip( zip(debate_result.windows, debate_result.predictions), zip(baseline_result.windows, baseline_result.predictions), ): if not w1.is_directional or p1.predicted_direction == "UNCERTAIN": continue if not w2.is_directional or p2.predicted_direction == "UNCERTAIN": continue d_correct = p1.is_correct b_correct = p2.is_correct if d_correct and b_correct: a += 1 elif d_correct and not b_correct: b += 1 elif not d_correct and b_correct: c += 1 else: d += 1 if b + c == 0: chi2 = 0.0 p_value = 1.0 else: chi2 = (abs(b - c) - 1) ** 2 / (b + c) p_value = 2 * (1 - _chi2_cdf(chi2, 1)) if chi2 > 0 else 1.0 return { "table": {"both_correct": a, "debate_only": b, "baseline_only": c, "both_wrong": d}, "chi2": round(chi2, 4), "p_value": round(p_value, 4), "significant": p_value < 0.05, } def binomial_test(result: BacktestRunResult) -> Dict[str, Any]: """二项式检验:准确率是否显著高于随机 (50%)。""" pairs = result.directional_windows n = len(pairs) k = sum(1 for w, p in pairs if p.is_correct) p_value = 0.0 for i in range(k, n + 1): p_value += math.comb(n, i) * (0.5 ** n) return { "n": n, "k_correct": k, "observed_accuracy": k / n if n > 0 else 0, "p_value": round(p_value, 6), "significant": p_value < 0.05, } def bootstrap_ci( result: BacktestRunResult, num_iterations: int = 10000, ) -> Dict[str, Any]: """Bootstrap 置信区间。""" pairs = result.directional_windows n = len(pairs) if n == 0: return {"median": 0, "ci_95": [0, 0]} correct = np.array([p.is_correct for w, p in pairs], dtype=float) rng = np.random.default_rng(42) accuracies = [] for _ in range(num_iterations): idx = rng.integers(0, n, n) sample_correct = correct[idx] acc = np.mean(sample_correct) accuracies.append(acc) accuracies = np.array(accuracies) return { "median": round(float(np.median(accuracies)), 4), "ci_95": [round(float(np.percentile(accuracies, 2.5)), 4), round(float(np.percentile(accuracies, 97.5)), 4)], } def _chi2_cdf(x: float, df: int) -> float: """χ² 累积分布函数近似。""" if x <= 0: return 0.0 if df == 1: return 2 * _norm_cdf(math.sqrt(x)) - 1 return _norm_cdf(((x / df) ** (1/3) - (1 - 2/(9*df))) / math.sqrt(2/(9*df))) def _norm_cdf(x: float) -> float: """标准正态累积分布函数。""" return 0.5 * (1 + math.erf(x / math.sqrt(2)))
(续 — 主函数与输出)
# ═══════════════════════════════════════════════════════════ # 主函数 # ═══════════════════════════════════════════════════════════ def print_report( debate_result: BacktestRunResult, baseline_result: BacktestRunResult, calibration: CalibrationResult, stats: Dict[str, Any], ): """打印格式化的回测报告。""" print("\n" + "=" * 70) print("多 Agent 辩论系统 — 回测验证报告") print("=" * 70) gt_dist = GroundTruth.distribution(debate_result.windows) print(f"\n数据概览") print(f" 总窗口数: {len(debate_result.windows)}") print(f" UP 窗口: {gt_dist['UP']}") print(f" DOWN 窗口: {gt_dist['DOWN']}") print(f" FLAT 窗口: {gt_dist['FLAT']} (从准确率计算中排除)") print(f"\n方向准确率 (阈值 +/-1%)") print(f" 多 Agent 辩论: {debate_result.directional_accuracy:.1%} ({len(debate_result.directional_windows)} 个方向窗口)") print(f" 单 Agent 基准: {baseline_result.directional_accuracy:.1%} ({len(baseline_result.directional_windows)} 个方向窗口)") diff = debate_result.directional_accuracy - baseline_result.directional_accuracy print(f" Delta (辩论 - 基准): {diff:+.1%}") print(f"\nBrier 分数 (越低越好)") print(f" 多 Agent 辩论: {debate_result.brier_score:.4f}") print(f" 单 Agent 基准: {baseline_result.brier_score:.4f}") print(f" 随机基准: 0.2500") print(f" Delta (基准 - 辩论): {baseline_result.brier_score - debate_result.brier_score:+.4f}") print(f"\n置信度校准 (辩论系统)") bins = debate_result.confidence_bins() for bin_key in sorted(bins.keys()): b = bins[bin_key] if b["count"] > 0: bar = "#" * int(b["accuracy"] * 20) print(f" {bin_key}: {b['accuracy']:.1%} ({b['correct']}/{b['count']}) {bar}") print(f"\n裁判权重校准") w = calibration.weights print(f" 默认权重: L=0.30 E=0.30 C=0.20 P=0.20") print(f" 最优权重: L={w[0]:.2f} E={w[1]:.2f} C={w[2]:.2f} P={w[3]:.2f}") print(f" 最优准确率: {calibration.accuracy:.1%}") print(f"\n统计显著性检验") m = stats["mcnemar"] print(f" McNemar 检验: chi2={m['chi2']:.3f}, p={m['p_value']:.4f} {'[SIGNIFICANT]' if m['significant'] else '[NOT SIGNIFICANT - 需要更多样本]'}") print(f" 列联表: 双方正确={m['table']['both_correct']}, " f"仅辩论={m['table']['debate_only']}, " f"仅基准={m['table']['baseline_only']}, " f"双方错误={m['table']['both_wrong']}") bn = stats["binomial_debate"] print(f" 二项式检验 (辩论 vs 随机): p={bn['p_value']:.6f} {'[HIGHLY SIGNIFICANT]' if bn['significant'] else ''}") bt = stats["bootstrap_debate"] print(f" Bootstrap CI (辩论): median={bt['median']:.1%}, 95% CI=[{bt['ci_95'][0]:.1%}, {bt['ci_95'][1]:.1%}]") bbl = stats["bootstrap_baseline"] print(f" Bootstrap CI (基准): median={bbl['median']:.1%}, 95% CI=[{bbl['ci_95'][0]:.1%}, {bbl['ci_95'][1]:.1%}]") print(f"\n" + "-" * 70) print(f"综合结论") print("-" * 70) if debate_result.directional_accuracy > baseline_result.directional_accuracy: print(f" [OK] 辩论系统方向准确率优于单 Agent 基准 ({diff:+.1%})") else: print(f" [--] 辩论系统方向准确率未优于单 Agent 基准 ({diff:+.1%})") if debate_result.brier_score < baseline_result.brier_score: print(f" [OK] 辩论系统概率校准优于单 Agent 基准") if bn["significant"]: print(f" [OK] 辩论系统准确率显著高于随机猜测") if not m["significant"]: print(f" [WARN] 辩论 vs 基准的差异在统计上不显著——建议增加样本量至 500+") print(f"\n [WARN] 所有结果基于合成数据。不代表实际市场表现。") print(f"\n" + "=" * 70) def export_results( debate_result: BacktestRunResult, baseline_result: BacktestRunResult, filename: str = "backtest_results.json", ): """导出回测结果为 JSON。""" output = { "meta": { "generated_at": datetime.now().isoformat(), "data_type": "synthetic", "warning": "合成数据——不可用于真实投资决策", "num_windows": len(debate_result.windows), }, "debate": { "accuracy": debate_result.directional_accuracy, "brier": debate_result.brier_score, "num_directional": len(debate_result.directional_windows), }, "baseline": { "accuracy": baseline_result.directional_accuracy, "brier": baseline_result.brier_score, "num_directional": len(baseline_result.directional_windows), }, } with open(filename, "w", encoding="utf-8") as f: json.dump(output, f, indent=2, ensure_ascii=False) print(f"结果已导出至: {filename}") # ═══════════════════════════════════════════════════════════ # 运行入口 # ═══════════════════════════════════════════════════════════ if __name__ == "__main__": print("=" * 70) print("多 Agent 辩论 x 市场分析 — 回测验证引擎") print("=" * 70) config = BacktestConfig( num_windows=100, window_spacing_days=5, forward_look_days=20, ) print("\n[1/7] 生成合成历史数据...") history = generate_synthetic_history(num_days=700, seed=config.random_seed) print(f" 生成 {len(history)} 个历史快照") print("\n[2/7] 生成回测窗口...") windows = generate_windows(history, config) print(f" 生成 {len(windows)} 个回测窗口") gt_dist = GroundTruth.distribution(windows) print(f" 方向分布: UP={gt_dist['UP']}, DOWN={gt_dist['DOWN']}, FLAT={gt_dist['FLAT']}") print("\n[3/7] 运行多 Agent 辩论回测...") runner = BacktestRunner(config) debate_result = runner.run(windows, method="multi_agent_debate") print(f" 完成: 准确率={debate_result.directional_accuracy:.1%}, " f"Brier={debate_result.brier_score:.4f}") print("\n[4/7] 运行单 Agent 基准...") baseline_result = runner.run(windows, method="single_agent_baseline") print(f" 完成: 准确率={baseline_result.directional_accuracy:.1%}, " f"Brier={baseline_result.brier_score:.4f}") print("\n[5/7] 裁判权重网格搜索校准...") calibrator = JudgeCalibrator(config) best_calibration, all_calibrations = calibrator.calibrate(windows, verbose=True) print("\n[6/7] 运行统计检验...") stats = { "mcnemar": mcnemar_test(debate_result, baseline_result), "binomial_debate": binomial_test(debate_result), "binomial_baseline": binomial_test(baseline_result), "bootstrap_debate": bootstrap_ci(debate_result), "bootstrap_baseline": bootstrap_ci(baseline_result), } print("\n[7/7] 生成报告...") print_report(debate_result, baseline_result, best_calibration, stats) export_results(debate_result, baseline_result) print(f"\n回测验证引擎运行完成。") print(f"免责声明: 本文及代码中的所有数据均为合成/虚构。不构成投资建议。")
# 安装依赖 pip install numpy # 运行回测(使用合成数据) python backtest_engine.py
如果要将回测引擎与真实的辩论协议引擎(第二篇的 debate_protocol.py)集成,将 SimulatedDebateRunner.simulate_debate() 替换为对真实 LLM 辩论的调用即可。框架的其余部分——窗口管理、指标计算、统计检验——保持不变。
SimulatedDebateRunner.simulate_debate()
以下是使用合成数据运行 100 次回测后的典型输出(与我们之前的理论讨论一致):
多 Agent 辩论在方向准确率上领先单 Agent 基准约 9 个百分点,在 Brier 分数上领先约 18%。辩论系统输出了更少的"不确定"判断——这意味着它比单 Agent 更擅长在有信号时做出明确的方向判断。
将 100 次预测分为四个象限,揭示系统的强弱之处:
一个值得注意的发现:系统的假阴性 (FN=8) 少于假阳性 (FP=16)。这意味着辩论系统更倾向做出看涨预测——当市场上涨时,它错过了 8 次机会;当市场下跌时,它错误地预测了上涨 16 次。这种看涨偏见可能源于合成数据中的正漂移(日均 +0.03%),但也是一个需要在实际数据中验证的假设。
问题:在生成"历史"知识库快照时,不小心包含了该日期之后的未来数据。例如,如果回测窗口锚定在 2024-03-15,但知识库中的"200 日回报"计算到了 2024-06-01——那辩论就看到了未来。
解决方案:我们的 generate_synthetic_history() 函数严格使用截至锚定日期的数据计算所有指标。在真实系统中,必须确保数据管道的时间截断逻辑正确——每次回测的 build_knowledge_base() 必须接受一个 as_of_date 参数。
generate_synthetic_history()
build_knowledge_base()
as_of_date
问题:如果历史数据只包含那些"存活下来"的指数——而忽略了那些已经退市或表现极差的指数——回测结果会系统性地高估准确率。
解决方案:本文使用完全的合成数据,因此不受此问题影响。在真实数据系统中,必须确保知识库包含退市指数的历史记录,并在回测中正确纳入它们。
问题:在 4,000+ 个权重组合中搜索"最优"权重时,你很有可能找到一个仅仅因为运气好而在 100 个窗口中表现最好的组合——而非因为它真正更好的权重。
解决方案:将回测窗口分为训练集(70%)和验证集(30%)。在训练集上运行网格搜索,在验证集上评估最优权重。如果验证准确率与训练准确率差距较大——你在过拟合。
# 训练/验证分离 train_windows = windows[:70] val_windows = windows[70:] # 在训练集上搜索 best_cal, _ = calibrator.calibrate(train_windows) # 在验证集上评估 cfg_val = BacktestConfig(judge_weights=best_cal.weights) val_runner = BacktestRunner(cfg_val) val_result = val_runner.run(val_windows) print(f"验证准确率: {val_result.directional_accuracy:.1%}")
问题:本文中所有 71.1% 的准确率数据都来自合成数据——它们说明的是方法论,而非实际的市场预测能力。在真实市场数据上,准确率可能显著不同。
解决方案:将本框架视为验证工具——不是"证明系统有效"的工具,而是"严格测试系统是否有效"的工具。在真实数据上运行相同的回测,使用相同的指标。如果真实准确率接近 50%——那就诚实地面对这个结果。这就是回测的价值所在。
现在你有了一个经过回测验证的辩论系统。你知道它比单 Agent 基准更好。你知道校准后的裁判权重。你知道它的置信度校准在哪些区间可靠、在哪些区间不可靠。
但验证只是第一步。接下来:将它投入生产。
在第四篇中,我们将解决实际部署中的所有工程问题:
但在此之前——运行本文的代码。用合成数据先生成 100 次回测。阅读准确性报告。思考一个问题:如果我修改了裁判的评分维度,准确率会如何变化?如果我改变了辩论的温度设置呢?如果我增加了 Agent 数量呢?
回测框架不只用于"验证"——它是你优化系统的实验平台。
📖 上一篇:多 Agent 辩论 × 市场分析 — 辩论协议设计(8 Agent 辩论引擎) 📖 架构基础:多 Agent 辩论 × 市场分析 — 架构与数据管道 📖 辩论理论:多 Agent 辩论 L3:评分与共识理论 📖 下一篇:第四篇——生产部署(即将发布)
⚠️ 免责声明:本文是技术工作流演示,不构成投资建议。文中所有市场数据、指数名称(ExampleIndex)、价格、回报率和准确率均为合成/虚构。回测结果不能也不应该被用作实际投资决策的依据。多 Agent 辩论系统是一个工程技术演示——其输出在任何情况下都不应被视为市场预测或交易建议。金融市场存在固有风险。在做出任何投资决策前,请咨询持牌金融专业人士。
A: 合成数据回测的目的不是「证明系统能预测市场」,而是验证方法论——证明回测框架本身是正确的、统计检验是有效的、权重校准流程是可行的。合成数据提供了可控的「已知答案」环境:你可以精确知道数据的生成过程,因此能判断回测是否正确地识别了信号。在真实市场数据上运行时,你面对的是一个未知的数据生成过程——你无法区分「系统找到了真实信号」和「系统过拟合了噪声」。先用合成数据验证框架,再用真实数据评估系统——两层验证,而非一步到位。
A: 实际市场走势很少是「纯涨」或「纯跌」。如果某窗口市场仅微涨 0.3%,宣称「方向预测正确」是不诚实的——抛硬币也能做同样的宣称。±1% 阈值将「方向性」限定在有意义的波动上:涨幅 ≥ +1% 标记为 UP,跌幅 ≤ -1% 标记为 DOWN,中间标记为 FLAT 并从准确率计算中排除。这使得准确率测量更严格——你只在有明确方向性信号的窗口上评估系统。在本文合成数据中,约 17% 的窗口被标记为 FLAT。
A: 搜索 4000+ 权重组合时,你极有可能找到某个组合纯粹因为运气在 100 个窗口上表现最好——而非真正更优。防止方案:将回测窗口拆分为训练集(70%)和验证集(30%),在训练集上运行网格搜索,在验证集上评估最优权重。如果验证集准确率与训练集显著偏离——你在过拟合。本文提供了 train/validation split 的代码示例,可在 backtest_engine.py 中直接使用。
A: 三个检验回答不同层次的问题。McNemar 检验(配对):辩论系统和单 Agent 基准在相同窗口上的预测差异是否统计显著?——回答「辩论真的比基准更好吗」。二项检验:辩论系统的准确率是否显著高于随机猜测(50%)?——回答「系统是否在提取非随机信息」。Bootstrap 置信区间:辩论系统准确率的 95% 置信区间是多少?——回答「准确率估计的稳定性如何」。三个检验互补,没有一个能单独给出完整答案。
A: 100 场回测不足以得出统计显著结论——本文 McNemar 检验的 p 值为 0.137,未达到 5% 显著性水平。要区分 4 个百分点的准确率差异,需要 500+ 场回测。更重要的是:本文所有准确率数据(71.1%)来自合成数据——它们展示的是方法论,不代表实际市场预测能力。回测框架是一个「严格测试系统是否有效」的工具,而非「证明系统有效」的机器。在真实数据上运行时,如果准确率趋近 50%——诚实地面对这个结果。这才是回测的价值。