# -*- 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 &gt; 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:]))