""" 策略验收标准 — 盈利优先的三层量化判断 设计哲学: 策略的最终目的是赚钱, 不是为了优化指标而优化指标。 好的夏普不等于赚钱 (高波动率 0 收益也能算出夏普), 但赚钱的策略 一定有正向期望和大于 1 的盈利因子。因此本验收采用"盈利优先"三层结构: L1 盈利性 (必须, 任一失败即拒收): - OOS 盈利因子 ≥ 1.3 (总盈利比总亏损多 30%) - OOS 净收益率 > 0% (样本外真的赚到钱) - OOS 每笔期望 > 0 (平均每单为正期望) L2 风险可控 (应该): - OOS 最大回撤 ≤ 20% (放宽原 15%, 赚钱策略可能回撤更大) L3 健壮性 (建议, 参考性指标): - OOS 夏普比率 ≥ 0.7 (放宽原 1.0, 作为参考而非硬指标) - OOS 总交易数 ≥ 30 (统计显著性) - IS/OOS 衰减比 ≥ 0.5 (非过拟合) 判定逻辑: L1 全过是必要条件, L1 任一失败 → 立即拒收, L2/L3 不再掩盖 L1 问题。 总判定 passed = L1 全过 AND L2 全过 AND L3 全过 (保持严格)。 但分层让 AI agent 能一眼看出"问题严重程度", 优先解决 L1。 用法: from acceptance import StrategyAcceptance checker = StrategyAcceptance() report = checker.check(wf_result) print(report.summary()) if report.passed: print("策略通过验收, 可交付") """ from __future__ import annotations from dataclasses import dataclass, field from typing import Optional import numpy as np # 默认验收标准 — 盈利优先三层结构 ACCEPTANCE_DEFAULTS = { # L1 盈利性 (必须): 不赚钱的策略直接拒收, 不让 Sharpe/衰减比等健壮性 # 指标掩盖"稳定亏损"假象 (例: IS/OOS 都亏, 衰减比反而很高) "oos_profit_factor_min": 1.3, # OOS 盈利因子下限 (总盈利/总亏损) "oos_return_min_pct": 0.0, # OOS 平均收益率下限 (真赚到钱) "oos_expectancy_min": 0.0, # OOS 每笔期望值下限 (平均每单为正) # L2 风险可控 (应该): 放宽阈值, 让能赚钱但回撤稍大的策略通过 "oos_max_drawdown_max_pct": 20.0, # OOS 最大回撤上限 (放宽原 15%) # L3 健壮性 (建议): 参考性指标, 不再作为拒收主力 "oos_sharpe_min": 0.7, # OOS 夏普比率下限 (放宽原 1.0, 降为参考) "oos_total_trades_min": 30, # OOS 总交易数下限 (统计显著性) "is_oos_decay_min": 0.5, # IS/OOS 衰减比下限 (非过拟合) } # 三层定义 (用于输出和 AI agent 决策) LAYERS = { "L1": { "name": "盈利性", "level": "必须", "description": "策略是否真的赚钱, 任一失败即拒收", }, "L2": { "name": "风险可控", "level": "应该", "description": "回撤是否在可接受范围", }, "L3": { "name": "健壮性", "level": "建议", "description": "统计显著性和非过拟合, 参考性指标", }, } @dataclass class CriterionResult: """单条验收标准的结果""" name: str # 标准名称 layer: str # 所属层级: "L1" / "L2" / "L3" threshold: float # 阈值 actual: float # 实际值 passed: bool # 是否通过 description: str = "" # 人类可读说明 suggestions: list = field(default_factory=list) # 失败时的修复建议 # 失败诊断建议表 — 按层分级 # L1 失败: 根本性调整 (换策略逻辑/换市场/换周期), 不是小修小补 # L2 失败: 调整风控 (止损/仓位/追踪) # L3 失败: 统计性问题 (样本不足/过拟合) _SUGGESTIONS = { # ── L1 盈利性: 不赚钱的根本原因 ────────────────────── "OOS 盈利因子": [ "策略逻辑本身可能不盈利, 重新审视入场/出场条件是否真的捕捉到正向期望", "切换策略类型 (趋势跟踪在震荡市会持续亏 PF<1, 均值回归在趋势市会亏)", "切换品种或周期 (当前 XAUUSD M5 可能不适合本策略逻辑)", "检查止损过紧是否被频繁扫损 (PF 低的常见原因是止损1)", ], "OOS 净收益率": [ "策略在样本外是亏损的, 首先检查 IS 是否也亏损 (若 IS 也亏, 是策略逻辑问题而非过拟合)", "评估交易成本是否吃掉所有收益 (尝试降低 fees/slippage 看是否转正, 若转正说明逻辑太边缘)", "减少交易频率 (提高入场门槛, 只做高确信度信号, 降低成本侵蚀)", "切换到更顺势的品种/周期 (震荡市做趋势必亏)", "反向信号 (若平均收益明显为负, 反向可能为正)", ], "OOS 每笔期望": [ "每笔平均收益为负, 优先检查胜率×盈亏比是否 >1 (期望 = 胜率×平均盈利 - 败率×平均亏损)", "若胜率高但期望为负, 是盈亏比失衡 (盈利单太小, 放大止盈或加追踪止损)", "若胜率低且期望为负, 是入场质量差 (增加过滤条件提高胜率)", "止损过紧会同时降低胜率和期望, 评估是否被噪音扫损 (放宽止损或换 ATR 止损)", "切换策略逻辑 (期望为负通常意味着信号方向与市场不匹配)", ], # ── L2 风险可控 ────────────────────────────────────── "OOS 最大回撤": [ "收紧止损比例 (如 2%→1.5%)", "启用 ATR 动态止损 (基于波动率自适应)", "启用追踪止损锁定利润", "降低仓位 (减少单笔风险暴露)", "增加趋势过滤, 避免逆势交易", ], # ── L3 健壮性 ──────────────────────────────────────── "OOS 夏普比率": [ "放宽止损/止盈比例 (当前可能止损过紧导致频繁止损)", "增加趋势过滤条件 (如 ADX>25 才交易, 避免震荡市亏损)", "切换到更大周期 (如 H1→H4, 减少噪音)", "增加信号过滤条件 (如要求成交量放大才入场)", "注意: 夏普已降为参考指标, L1 全过时即使夏普略低也可接受", ], "OOS 总交易数": [ "缩短指标周期 (如 RSI 14→7, 增加信号频率)", "降低入场阈值 (如 RSI<30 → RSI<35)", "切换到更小周期 (如 H1→M15)", "放宽信号过滤条件", "检查 warmup_bars 是否过大 (导致 OOS 窗口被吃掉)", ], "IS/OOS 衰减比": [ "参数过拟合: 缩小参数空间 (如 fast=5,10,15 → fast=8,10,12)", "增加 walk-forward 窗口数 (更小的 train_size/test_size)", "简化策略逻辑 (减少可调参数数量)", "增加正则化: 用中位数参数而非最优参数", "检查是否使用了未来数据 (python -m app.main check --strategy XXX --dynamic)", ], } @dataclass class AcceptanceReport: """验收报告 — 分层呈现""" passed: bool # 总体是否通过 criteria: list = field(default_factory=list) # 各标准结果 (按 L1→L2→L3 顺序) l1_passed: bool = True # L1 盈利性是否全过 (关键标志) l2_passed: bool = True # L2 风险可控是否全过 l3_passed: bool = True # L3 健壮性是否全过 def summary(self) -> str: # 顶部结论 (突出 L1) if not self.l1_passed: verdict = "❌ 未通过 (L1 盈利性未达标 — 策略不赚钱, 无需看 L2/L3)" elif not self.l2_passed: verdict = "❌ 未通过 (L2 风险可控未达标)" elif not self.l3_passed: verdict = "❌ 未通过 (L3 健壮性参考指标未达标)" else: verdict = "✅ 通过 (策略在样本外真的赚到钱, 且风险可控)" lines = [ f"验收结果: {verdict}", f"{'─' * 70}", ] # 按层分组输出 current_layer = None for c in self.criteria: if c.layer != current_layer: current_layer = c.layer layer_info = LAYERS[c.layer] layer_passed = (c.layer == "L1" and self.l1_passed) or \ (c.layer == "L2" and self.l2_passed) or \ (c.layer == "L3" and self.l3_passed) status = "✓" if layer_passed else "✗" lines.append( f" [{c.layer} {layer_info['name']}] {layer_info['level']} " f"— {layer_info['description']} {status}" ) lines.append(f" {'─' * 66}") status = "✓" if c.passed else "✗" lines.append( f" {c.name:<18} 阈值={c.threshold:>8.2f} " f"实际={c.actual:>8.2f} {status}" ) if c.description: lines.append(f" └ {c.description}") # 失败时附加修复建议 if not c.passed and c.suggestions: lines.append(f" └ 修复建议:") for i, s in enumerate(c.suggestions, 1): lines.append(f" {i}. {s}") lines.append(f"{'─' * 70}") return "\n".join(lines) def to_dict(self) -> dict: """转为可 JSON 序列化的字典""" return { "passed": self.passed, "l1_passed": self.l1_passed, "l2_passed": self.l2_passed, "l3_passed": self.l3_passed, "verdict": self._verdict_text(), "criteria": [ { "name": c.name, "layer": c.layer, "layer_name": LAYERS[c.layer]["name"], "level": LAYERS[c.layer]["level"], "threshold": c.threshold, "actual": c.actual, "passed": c.passed, "description": c.description, "suggestions": c.suggestions if not c.passed else [], } for c in self.criteria ], } def _verdict_text(self) -> str: if not self.l1_passed: return "L1 盈利性未达标 — 策略不赚钱, 无需看 L2/L3" if not self.l2_passed: return "L2 风险可控未达标" if not self.l3_passed: return "L3 健壮性参考指标未达标" return "全部通过 — 策略在样本外真的赚到钱, 且风险可控" class StrategyAcceptance: """ 策略验收检查器 — 盈利优先三层 参数: criteria: 验收标准字典, None 时使用 ACCEPTANCE_DEFAULTS 用法: checker = StrategyAcceptance() report = checker.check(wf_result) if report.passed: print("策略通过验收, 可以交付") elif not report.l1_passed: print("策略不赚钱, 必须重新设计而非调参") """ def __init__(self, criteria: Optional[dict] = None): self.criteria = criteria or ACCEPTANCE_DEFAULTS.copy() def check(self, wf_result) -> AcceptanceReport: """ 检查 walk-forward 结果是否满足所有验收标准 参数: wf_result: WalkForwardResult 对象 返回: AcceptanceReport (分层结构) """ results = [] l1_passed = True l2_passed = True l3_passed = True # ════════════════════════════════════════════════════ # L1 盈利性 (必须) — 优先检查, 不赚钱直接拒收 # ════════════════════════════════════════════════════ # L1.1 OOS 盈利因子 (Profit Factor = 总盈利/总亏损) pf = wf_result.oos_profit_factor_avg if np.isnan(pf): pf = 0.0 threshold = self.criteria["oos_profit_factor_min"] passed = pf >= threshold results.append(CriterionResult( "OOS 盈利因子", "L1", threshold, pf, passed, f"总盈利/总亏损 = {pf:.2f}, " f"{'≥' if passed else '<'} 阈值 {threshold} " f"({'正期望' if passed else '亏损或勉强盈利'})", suggestions=[] if passed else _SUGGESTIONS["OOS 盈利因子"], )) l1_passed &= passed # L1.2 OOS 净收益率 (真赚到钱) oos_return = wf_result.oos_return_avg if np.isnan(oos_return): oos_return = -999.0 threshold = self.criteria["oos_return_min_pct"] passed = oos_return > threshold results.append(CriterionResult( "OOS 净收益率", "L1", threshold, oos_return, passed, f"OOS 平均收益 {oos_return:.2f}% " f"{'盈利' if passed else '亏损'}", suggestions=[] if passed else _SUGGESTIONS["OOS 净收益率"], )) l1_passed &= passed # L1.3 OOS 每笔期望值 (平均每单为正) exp_val = wf_result.oos_expectancy_avg if np.isnan(exp_val): exp_val = -999.0 threshold = self.criteria["oos_expectancy_min"] passed = exp_val > threshold results.append(CriterionResult( "OOS 每笔期望", "L1", threshold, exp_val, passed, f"每笔平均期望 {exp_val:.4f} " f"{'为正' if passed else '为负/为零'}", suggestions=[] if passed else _SUGGESTIONS["OOS 每笔期望"], )) l1_passed &= passed # ════════════════════════════════════════════════════ # L2 风险可控 (应该) # ════════════════════════════════════════════════════ # L2.1 OOS 最大回撤 (放宽到 20%) oos_dd = abs(wf_result.oos_max_drawdown_avg) if np.isnan(oos_dd): oos_dd = 999.0 threshold = self.criteria["oos_max_drawdown_max_pct"] passed = oos_dd <= threshold results.append(CriterionResult( "OOS 最大回撤", "L2", threshold, oos_dd, passed, f"{'低于' if passed else '超过'}最大回撤限制 {threshold}%", suggestions=[] if passed else _SUGGESTIONS["OOS 最大回撤"], )) l2_passed &= passed # ════════════════════════════════════════════════════ # L3 健壮性 (建议, 参考性指标) # ════════════════════════════════════════════════════ # L3.1 OOS 夏普比率 (放宽到 0.7, 降为参考) oos_sharpe = wf_result.oos_sharpe_avg if np.isnan(oos_sharpe): oos_sharpe = -999.0 threshold = self.criteria["oos_sharpe_min"] passed = oos_sharpe >= threshold results.append(CriterionResult( "OOS 夏普比率", "L3", threshold, oos_sharpe, passed, f"{'达到' if passed else '未达到'}参考夏普要求 {threshold} " f"(已降为参考指标, L1 全过时可接受略低)", suggestions=[] if passed else _SUGGESTIONS["OOS 夏普比率"], )) l3_passed &= passed # L3.2 OOS 总交易数 (统计显著性) total_trades = wf_result.oos_trades_total threshold = self.criteria["oos_total_trades_min"] passed = total_trades >= threshold results.append(CriterionResult( "OOS 总交易数", "L3", threshold, float(total_trades), passed, f"{'达到' if passed else '不足'}最小交易数 {threshold} (统计显著性)", suggestions=[] if passed else _SUGGESTIONS["OOS 总交易数"], )) l3_passed &= passed # L3.3 IS/OOS 衰减比 (过拟合检测) decay = wf_result.decay_ratio if np.isnan(decay): decay = 0.0 threshold = self.criteria["is_oos_decay_min"] passed = decay >= threshold overfit_str = "非过拟合" if passed else "过拟合风险" results.append(CriterionResult( "IS/OOS 衰减比", "L3", threshold, decay, passed, f"OOS/IS = {decay:.1%}, {overfit_str} " f"(注意: 若 IS 也是亏损, 高衰减比不代表策略好)", suggestions=[] if passed else _SUGGESTIONS["IS/OOS 衰减比"], )) l3_passed &= passed return AcceptanceReport( passed=bool(l1_passed and l2_passed and l3_passed), criteria=results, l1_passed=bool(l1_passed), l2_passed=bool(l2_passed), l3_passed=bool(l3_passed), )