387 lines
16 KiB
Python
387 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Walk-Forward 滚动验证
|
||
====================
|
||
将单份报告的逐笔交易按时间切成多个滚动窗口，每个窗口内
|
||
前段做 IS（参数评估段）、后段做 OOS（样本外验证段），
|
||
计算每个窗口的 IS/OOS 指标，再汇总：
|
||
|
||
- Walk-Forward Efficiency (WFE) = Σ OOS净盈利 / Σ IS净盈利
|
||
- OOS 盈利窗口占比
|
||
- IS PF 与 OOS PF 的相关性（高=过拟合低；低=参数不稳健）
|
||
- 滚动 IS/OOS PF 曲线
|
||
|
||
相比单次 IS/OSS 划分，能反映参数在不同行情段的稳健性。
|
||
|
||
可独立运行，也可被 run_analysis.py 导入集成进主报告。
|
||
|
||
用法:
|
||
python walk_forward.py <report.xlsx> [--is-days 60] [--oos-days 30] [--step-days 30]
|
||
python walk_forward.py --demo
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import html
|
||
import os
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
import mt5_report_parser as mp
|
||
|
||
OUT_DIR = "output"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 单段指标（从 mt5_report_parser 复用）
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
from mt5_report_parser import compute_segment_metrics as _seg_metrics
|
||
|
||
|
||
@dataclass
|
||
class WFWindow:
|
||
idx: int
|
||
is_start: pd.Timestamp
|
||
is_end: pd.Timestamp
|
||
oos_start: pd.Timestamp
|
||
oos_end: pd.Timestamp
|
||
is_metrics: Dict[str, float]
|
||
oos_metrics: Dict[str, float]
|
||
|
||
|
||
def walk_forward(
|
||
trades: pd.DataFrame,
|
||
is_days: int = 60,
|
||
oos_days: int = 30,
|
||
step_days: Optional[int] = None,
|
||
) -> List[WFWindow]:
|
||
"""
|
||
滚动切窗（向量化边界查找）。
|
||
trades: 含 open_time, net_profit 的 DataFrame
|
||
is_days / oos_days: IS/OOS 窗口天数
|
||
step_days: 滑动步长（默认 = oos_days，即不重叠的 OOS）
|
||
"""
|
||
if trades is None or trades.empty:
|
||
return []
|
||
t = trades.sort_values("open_time").reset_index(drop=True)
|
||
times = t["open_time"].to_numpy()
|
||
nets = t["net_profit"].astype(float).to_numpy()
|
||
n_trades = len(nets)
|
||
|
||
# 预计算累积和（含前导0，cumsum[i] = sum of nets[:i]）
|
||
cumsum = np.concatenate([[0.0], np.cumsum(nets)])
|
||
|
||
if step_days is None:
|
||
step_days = oos_days
|
||
|
||
start = t["open_time"].min()
|
||
end = t["open_time"].max()
|
||
total_span = (end - start).days
|
||
win_span = is_days + oos_days
|
||
if total_span < win_span:
|
||
return []
|
||
|
||
# 用 numpy 二分查找窗口边界索引（所有窗口一次性确定）
|
||
cur = np.datetime64(start, "ns")
|
||
end_ns = np.datetime64(end, "ns")
|
||
bounds: List[Tuple[int, int, int, int]] = [] # (is_s, is_e, oos_s, oos_e) 为交易索引
|
||
idx = 0
|
||
while True:
|
||
is_e = cur + np.timedelta64(is_days, "D")
|
||
oos_s = is_e
|
||
oos_e = oos_s + np.timedelta64(oos_days, "D")
|
||
if oos_e > end_ns + np.timedelta64(1, "D"):
|
||
break
|
||
i_s = int(np.searchsorted(times, cur, side="left"))
|
||
i_e = int(np.searchsorted(times, is_e, side="left"))
|
||
o_s = int(np.searchsorted(times, oos_s, side="left"))
|
||
o_e = int(np.searchsorted(times, oos_e, side="left"))
|
||
bounds.append((i_s, i_e, o_s, o_e))
|
||
cur += np.timedelta64(step_days, "D")
|
||
idx += 1
|
||
|
||
windows: List[WFWindow] = []
|
||
for i, (i_s, i_e, o_s, o_e) in enumerate(bounds):
|
||
if i_e - i_s == 0 or o_e - o_s == 0:
|
||
continue
|
||
# IS 段指标（用累积和 O(1) 计算）
|
||
is_net_sum = float(cumsum[i_e] - cumsum[i_s])
|
||
is_n = i_e - i_s
|
||
is_seg = nets[i_s:i_e]
|
||
is_gp = float(np.sum(is_seg[is_seg > 0]))
|
||
is_gl = float(-np.sum(is_seg[is_seg <= 0]))
|
||
is_pf = is_gp / is_gl if is_gl > 0 else np.inf
|
||
is_cum = cumsum[i_s + 1:i_e + 1] - cumsum[i_s:i_e]
|
||
is_dd = float(np.min(is_cum - np.maximum.accumulate(is_cum))) if len(is_cum) else 0.0
|
||
is_win = float(np.sum(is_seg > 0) / is_n * 100) if is_n else 0.0
|
||
# OOS 段
|
||
oos_net_sum = float(cumsum[o_e] - cumsum[o_s])
|
||
oos_n = o_e - o_s
|
||
oos_seg = nets[o_s:o_e]
|
||
oos_gp = float(np.sum(oos_seg[oos_seg > 0]))
|
||
oos_gl = float(-np.sum(oos_seg[oos_seg <= 0]))
|
||
oos_pf = oos_gp / oos_gl if oos_gl > 0 else np.inf
|
||
oos_cum = cumsum[o_s + 1:o_e + 1] - cumsum[o_s:o_e]
|
||
oos_dd = float(np.min(oos_cum - np.maximum.accumulate(oos_cum))) if len(oos_cum) else 0.0
|
||
oos_win = float(np.sum(oos_seg > 0) / oos_n * 100) if oos_n else 0.0
|
||
|
||
windows.append(WFWindow(
|
||
idx=i,
|
||
is_start=times[i_s], is_end=times[min(i_e - 1, n_trades - 1)],
|
||
oos_start=times[o_s], oos_end=times[min(o_e - 1, n_trades - 1)],
|
||
is_metrics={
|
||
"n": is_n, "net": is_net_sum, "pf": is_pf,
|
||
"win": is_win, "dd": is_dd, "exp": is_net_sum / is_n if is_n else 0.0,
|
||
},
|
||
oos_metrics={
|
||
"n": oos_n, "net": oos_net_sum, "pf": oos_pf,
|
||
"win": oos_win, "dd": oos_dd, "exp": oos_net_sum / oos_n if oos_n else 0.0,
|
||
},
|
||
))
|
||
|
||
return windows
|
||
|
||
|
||
def wf_summary(windows: List[WFWindow]) -> Dict[str, Any]:
|
||
if not windows:
|
||
return {}
|
||
is_nets = [w.is_metrics["net"] for w in windows]
|
||
oos_nets = [w.oos_metrics["net"] for w in windows]
|
||
is_pfs = [w.is_metrics["pf"] for w in windows if not np.isinf(w.is_metrics["pf"])]
|
||
oos_pfs = [w.oos_metrics["pf"] for w in windows if not np.isinf(w.oos_metrics["pf"])]
|
||
wfe = sum(oos_nets) / sum(is_nets) if sum(is_nets) != 0 else np.nan
|
||
oos_profitable = sum(1 for n in oos_nets if n > 0)
|
||
# IS-OOS PF 相关性
|
||
if len(is_pfs) > 2 and len(oos_pfs) > 2:
|
||
# 配对（取有限值的交集）
|
||
pairs = [(w.is_metrics["pf"], w.oos_metrics["pf"]) for w in windows
|
||
if not np.isinf(w.is_metrics["pf"]) and not np.isinf(w.oos_metrics["pf"])]
|
||
if len(pairs) > 2:
|
||
arr = np.array(pairs)
|
||
corr = float(np.corrcoef(arr[:, 0], arr[:, 1])[0, 1])
|
||
else:
|
||
corr = np.nan
|
||
else:
|
||
corr = np.nan
|
||
return {
|
||
"n_windows": len(windows),
|
||
"wfe": float(wfe),
|
||
"oos_profitable": oos_profitable,
|
||
"oos_profitable_pct": float(oos_profitable / len(windows) * 100),
|
||
"sum_is_net": float(sum(is_nets)),
|
||
"sum_oos_net": float(sum(oos_nets)),
|
||
"avg_is_pf": float(np.mean(is_pfs)) if is_pfs else 0.0,
|
||
"avg_oos_pf": float(np.mean(oos_pfs)) if oos_pfs else 0.0,
|
||
"is_oos_pf_corr": corr,
|
||
"max_oos_dd": float(min(w.oos_metrics["dd"] for w in windows)),
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# SVG
|
||
# --------------------------------------------------------------------------- #
|
||
def svg_wf_curve(windows: List[WFWindow], w: int = 760, h: int = 260) -> str:
|
||
if not windows:
|
||
return ""
|
||
top, bottom, left, right = 24, 36, 46, 16
|
||
plot_h = h - top - bottom
|
||
plot_w = w - left - right
|
||
is_pfs = [w.is_metrics["pf"] for w in windows]
|
||
oos_pfs = [w.oos_metrics["pf"] for w in windows]
|
||
vals = is_pfs + oos_pfs + [1.0]
|
||
vmin = min(vals)
|
||
vmax = max(vals)
|
||
if vmax == vmin:
|
||
vmax = vmin + 1
|
||
vmax = max(vmax, 1.0)
|
||
vmin = min(vmin, 0.0)
|
||
|
||
def xy(i, v):
|
||
x = left + i / max(1, len(windows) - 1) * plot_w
|
||
y = top + plot_h - (v - vmin) / (vmax - vmin) * plot_h
|
||
return x, y
|
||
|
||
def poly(vals_, color):
|
||
pts = " ".join(f"{x:.1f},{y:.1f}" for x, y in [xy(i, v) for i, v in enumerate(vals_)])
|
||
return f'<polyline points="{pts}" fill="none" stroke="{color}" stroke-width="1.6"><title></title></polyline>'
|
||
|
||
# 1.0 参考线
|
||
one_y = top + plot_h - (1.0 - vmin) / (vmax - vmin) * plot_h
|
||
grid = ""
|
||
for frac in (0, 0.25, 0.5, 0.75, 1.0):
|
||
yv = vmin + frac * (vmax - vmin)
|
||
gy = top + plot_h - frac * plot_h
|
||
grid += (f'<line x1="{left}" y1="{gy:.1f}" x2="{w-right}" y2="{gy:.1f}" stroke="#e5e7eb" stroke-width="0.5"/>'
|
||
f'<text x="{left-6}" y="{gy+3:.1f}" font-size="9" fill="#6b7280" text-anchor="end">{yv:.2f}</text>')
|
||
one_line = f'<line x1="{left}" y1="{one_y:.1f}" x2="{w-right}" y2="{one_y:.1f}" stroke="#9ca3af" stroke-dasharray="3,3" stroke-width="0.8"/>'
|
||
|
||
# 窗口标签（仅首/中/尾）
|
||
xlabs = ""
|
||
for i in [0, len(windows) // 2, len(windows) - 1]:
|
||
x = left + i / max(1, len(windows) - 1) * plot_w
|
||
xlabs += f'<text x="{x:.1f}" y="{h-bottom+14:.1f}" font-size="9" fill="#374151" text-anchor="middle">#{windows[i].idx}</text>'
|
||
|
||
return (
|
||
f'<svg viewBox="0 0 {w} {h}" class="chart">'
|
||
f'{grid}{one_line}'
|
||
f'{poly(is_pfs, "#2563eb")}'
|
||
f'{poly(oos_pfs, "#dc2626")}'
|
||
f'{xlabs}'
|
||
f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">滚动窗口 IS vs OOS 盈利因子 (PF=1 为盈亏平衡线)</text>'
|
||
f'<text x="{w-right}" y="32" font-size="10" fill="#2563eb" text-anchor="end">IS PF</text>'
|
||
f'<text x="{w-right}" y="46" font-size="10" fill="#dc2626" text-anchor="end">OOS PF</text>'
|
||
f'<text x="{w//2}" y="{h-4}" font-size="9" fill="#6b7280" text-anchor="middle">窗口序号 →</text>'
|
||
f'</svg>'
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# HTML 片段（供 run_analysis.py 嵌入）
|
||
# --------------------------------------------------------------------------- #
|
||
def wf_html_fragment(windows: List[WFWindow]) -> str:
|
||
if not windows:
|
||
return "<p class='muted'>数据不足以切出 Walk-Forward 窗口。</p>"
|
||
s = wf_summary(windows)
|
||
rows = ""
|
||
for w in windows:
|
||
oos_cls = " class='pos'" if w.oos_metrics["net"] > 0 else " class='neg'"
|
||
rows += (
|
||
f"<tr><td>#{w.idx}</td>"
|
||
f"<td>{w.is_start:%Y-%m-%d}~{w.is_end:%Y-%m-%d}</td>"
|
||
f"<td>{w.oos_start:%Y-%m-%d}~{w.oos_end:%Y-%m-%d}</td>"
|
||
f"<td>{w.is_metrics['n']}</td>"
|
||
f"<td>{w.is_metrics['net']:+.2f}</td>"
|
||
f"<td>{w.is_metrics['pf']:.2f}</td>"
|
||
f"<td>{w.is_metrics['win']:.1f}%</td>"
|
||
f"<td>{w.oos_metrics['n']}</td>"
|
||
f"<td{oos_cls}>{w.oos_metrics['net']:+.2f}</td>"
|
||
f"<td>{w.oos_metrics['pf']:.2f}</td>"
|
||
f"<td>{w.oos_metrics['win']:.1f}%</td>"
|
||
f"<td>{w.oos_metrics['dd']:.2f}</td></tr>"
|
||
)
|
||
table = (
|
||
"<table class='data'><thead><tr>"
|
||
"<th>窗口</th><th>IS区间</th><th>OOS区间</th>"
|
||
"<th>IS笔</th><th>IS净</th><th>IS PF</th><th>IS胜率</th>"
|
||
"<th>OOS笔</th><th>OOS净</th><th>OOS PF</th><th>OOS胜率</th><th>OOS回撤</th>"
|
||
"</tr></thead><tbody>" + rows + "</tbody></table>"
|
||
)
|
||
|
||
corr_str = f"{s['is_oos_pf_corr']:.2f}" if not np.isnan(s['is_oos_pf_corr']) else "—"
|
||
cards = (
|
||
f"<div class='cards'>"
|
||
f"<div class='card'><div class='k'>窗口数</div><div class='v'>{s['n_windows']}</div></div>"
|
||
f"<div class='card'><div class='k'>WFE (ΣOOS净/ΣIS净)</div><div class='v {'pos' if s['wfe']>0 else 'neg'}'>{s['wfe']:+.2f}</div></div>"
|
||
f"<div class='card'><div class='k'>OOS 盈利窗口占比</div><div class='v'>{s['oos_profitable_pct']:.0f}%</div></div>"
|
||
f"<div class='card'><div class='k'>IS-OOS PF 相关性</div><div class='v'>{corr_str}</div></div>"
|
||
f"</div>"
|
||
)
|
||
|
||
interp = ""
|
||
if not np.isnan(s['is_oos_pf_corr']):
|
||
if s['is_oos_pf_corr'] > 0.6:
|
||
interp = "<p class='muted'>IS 与 OOS 的 PF 相关性较高 → 参数对行情段有一定泛化能力。</p>"
|
||
elif s['is_oos_pf_corr'] < 0.2:
|
||
interp = "<p class='muted'>IS 与 OOS 的 PF 相关性低 → IS 表现难以预测 OOS，参数不稳健，警惕过拟合单段行情。</p>"
|
||
else:
|
||
interp = "<p class='muted'>IS/OOS PF 中等相关性，泛化能力中等。</p>"
|
||
if s['wfe'] < 0:
|
||
interp += "<div class='warn'>WFE 为负：OOS 累计亏损。即使 IS 段盈利，策略也未能泛化。</div>"
|
||
elif s['wfe'] > 0.5:
|
||
interp += "<div class='note'>WFE > 0.5：OOS 能保留 IS 的一半以上盈利，泛化较好。</div>"
|
||
|
||
return cards + svg_wf_curve(windows) + table + interp
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 独立 HTML 报告
|
||
# --------------------------------------------------------------------------- #
|
||
def build_standalone_html(name: str, windows: List[WFWindow]) -> str:
|
||
frag = wf_html_fragment(windows)
|
||
return f"""<!doctype html><html lang="zh-CN"><head><meta charset="utf-8">
|
||
<title>Walk-Forward 报告 — {html.escape(name)}</title>
|
||
<style>
|
||
body {{ font-family: -apple-system, "Microsoft YaHei", sans-serif; background:#f8fafc; color:#111827; margin:0; padding:20px;}}
|
||
.wrap {{ max-width:1100px; margin:0 auto;}}
|
||
h1 {{ color:#1e3a8a; border-bottom:3px solid #1e3a8a; padding-bottom:8px;}}
|
||
table.data {{ border-collapse:collapse; width:100%; background:#fff; margin:8px 0 16px; font-size:12px;}}
|
||
table.data th, table.data td {{ border:1px solid #e2e8f0; padding:5px 7px; text-align:left;}}
|
||
table.data th {{ background:#f1f5f9;}}
|
||
td.pos, tr.pos td {{ color:#16a34a;}} td.neg, tr.neg td {{ color:#dc2626;}}
|
||
.muted {{ color:#6b7280; font-size:12px;}}
|
||
.cards {{ display:grid; grid-template-columns:repeat(auto-fit,minmax(200px,1fr)); gap:12px; margin:16px 0;}}
|
||
.card {{ background:#fff; padding:14px; border-radius:8px; border:1px solid #e2e8f0;}}
|
||
.card .k {{ color:#6b7280; font-size:12px;}} .card .v {{ font-size:20px; font-weight:600; margin-top:4px;}}
|
||
.card .v.pos {{ color:#16a34a;}} .card .v.neg {{ color:#dc2626;}}
|
||
.chart {{ width:100%; height:auto; background:#fff; border:1px solid #e2e8f0; border-radius:6px;}}
|
||
.note {{ background:#fef3c7; border-left:4px solid #f59e0b; padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||
.warn {{ background:#fee2e2; border-left:4px solid #dc2626; padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||
</style></head><body><div class="wrap">
|
||
<h1>Walk-Forward 滚动验证报告</h1>
|
||
<p class='muted'>来源: {html.escape(name)}</p>
|
||
{frag}
|
||
</div></body></html>"""
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# demo
|
||
# --------------------------------------------------------------------------- #
|
||
def _demo_trades() -> pd.DataFrame:
|
||
rng = np.random.default_rng(7)
|
||
n = 1200
|
||
start = pd.Timestamp("2024-01-01")
|
||
times = start + pd.to_timedelta(np.arange(n) * 3, unit="h")
|
||
# 让 PF 随时间漂移，模拟参数不稳健
|
||
drift = np.sin(np.arange(n) / 200) * 1.5
|
||
base = rng.normal(-0.1 + drift * 0.1, 3, n)
|
||
return pd.DataFrame({"open_time": times, "net_profit": base})
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# CLI
|
||
# --------------------------------------------------------------------------- #
|
||
def main(argv: List[str]) -> int:
|
||
ap = argparse.ArgumentParser(description="Walk-Forward 滚动验证")
|
||
ap.add_argument("report", nargs="?", help="MT5 回测报告 xlsx 路径")
|
||
ap.add_argument("--is-days", type=int, default=60)
|
||
ap.add_argument("--oos-days", type=int, default=30)
|
||
ap.add_argument("--step-days", type=int, default=None)
|
||
ap.add_argument("--demo", action="store_true")
|
||
args = ap.parse_args(argv)
|
||
|
||
os.makedirs(OUT_DIR, exist_ok=True)
|
||
|
||
if args.demo:
|
||
trades = _demo_trades()
|
||
name = "demo"
|
||
else:
|
||
if not args.report:
|
||
ap.error("需要报告路径或 --demo")
|
||
rep = mp.parse_report(args.report)
|
||
trades = rep.trades
|
||
name = os.path.basename(args.report)
|
||
|
||
windows = walk_forward(trades, args.is_days, args.oos_days, args.step_days)
|
||
if not windows:
|
||
print("数据不足以切出窗口，试试更小的 is-days/oos-days")
|
||
return 1
|
||
s = wf_summary(windows)
|
||
print(f"窗口数: {s['n_windows']} WFE: {s['wfe']:+.3f} "
|
||
f"OOS盈利窗口: {s['oos_profitable_pct']:.0f}% "
|
||
f"IS-OOS PF 相关性: {s['is_oos_pf_corr']}")
|
||
html_doc = build_standalone_html(name, windows)
|
||
out = os.path.join(OUT_DIR, "walk_forward.html")
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(html_doc)
|
||
print(f"HTML 报告: {out}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv[1:]))
|