389 lines
17 KiB
Python
389 lines
17 KiB
Python
"""
|
|
策略验收标准 — 盈利优先的三层量化判断
|
|
|
|
设计哲学:
|
|
策略的最终目的是赚钱, 不是为了优化指标而优化指标。
|
|
好的夏普不等于赚钱 (高波动率 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 低的常见原因是止损<ATR, 噪音打掉止损)",
|
|
"尝试反向信号 (若 PF 明显 <1, 反向可能 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),
|
|
)
|