"""Step 20：全级别对 × 全品种的区间套批量回测（多进程）。

数据已补全为 BTC/ETH/SOL 全周期 2172~2543 天，此前所有结论都建立在残缺样本上，
本步用完整数据一次性重跑，并行执行避免逐个等待。

任务单元 = (品种, 小级别, 大级别1, 大级别2)。
每个 worker 独立跑：小级别中枢 -> 快速三买 -> 挂大级别分型 -> 事件驱动回测。
"""
from __future__ import annotations

import argparse
import os
import sys
import warnings
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

import numpy as np
import pandas as pd

warnings.filterwarnings("ignore")
os.environ.setdefault("OMP_NUM_THREADS", "1")
os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")
os.environ.setdefault("MKL_NUM_THREADS", "1")

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))

pd.set_option("display.width", 280)

SL, TP, MAXB = 1.5, 3.0, 48
DEFAULT_PAIRS = "5m:15m:1h,15m:1h:4h,30m:2h:4h,1h:4h:1d,4h:1d:1w"
SYMBOLS = ["BTC", "ETH", "SOL"]
# 1m 有 300 万+ 根，全量跑内存吃紧，默认只取最近这么多根
MAX_ROWS = {"1m": 1_200_000}


def run_one(task: tuple) -> dict | None:
    """单个 (品种, 级别对) 的完整回测。运行在子进程里。"""
    import warnings as _w
    _w.filterwarnings("ignore")
    sys.path.insert(0, str(HERE))
    sys.path.insert(0, str(HERE.parent))

    from chanlun import TF_DF
    from lib.breakout import run_trades
    from lib.data import fetch_ohlcv
    from lib.fast_bsp3 import find_fast_bsp3
    from lib.fx_signal import extract_fx_signals, signals_to_frame
    from lib.nested_bsp import attach_htf_context, htf_fx_timeline
    from lib.nested_level import build_htf_zones

    sym_short, ltf, h1, h2 = task
    symbol = f"{sym_short}/USDT:USDT"
    try:
        df_l = fetch_ohlcv(symbol, ltf, MAX_ROWS.get(ltf, 10**9))
        if df_l is None or len(df_l) < 3000:
            return None
        chan_l = TF_DF(df_l, 1, ltf)
        cdf_l = chan_l.dataframe
        zones = build_htf_zones(cdf_l, ltf, chan=chan_l)
        sig = find_fast_bsp3(cdf_l, zones)
        if sig.empty or len(sig) < 10:
            return None

        for tf, pref in ((h1, "h1"), (h2, "h2")):
            df_h = fetch_ohlcv(symbol, tf, 10**9)
            if df_h is None or len(df_h) < 300:
                continue
            chan_h = TF_DF(df_h, 1, tf)
            s = signals_to_frame(extract_fx_signals(chan_h, chan_h.dataframe))
            sig = attach_htf_context(sig, cdf_l, htf_fx_timeline(s, chan_h.dataframe), pref)

        entries = list(zip(sig["entry_idx"].astype(int), sig["direction"].astype(int)))
        tr = run_trades(cdf_l, entries, SL, TP, MAXB)
        if tr.empty:
            return None
        m = sig.set_index("entry_idx")
        tr = tr.copy()
        tr["symbol"] = sym_short
        tr["ltf"] = ltf
        tr["pair"] = f"{ltf}/{h1}+{h2}"
        tr["date"] = cdf_l["date"].to_numpy()[tr["entry_idx"].to_numpy()]
        for c in ("h1_agree", "h2_agree", "depth", "lag", "width_pct"):
            tr[c] = tr["entry_idx"].map(m[c]) if c in m.columns else np.nan
        return {
            "task": f"{sym_short} {ltf}/{h1}+{h2}",
            "n_zones": len(zones),
            "n_sig": len(sig),
            "trades": tr,
        }
    except Exception as e:  # 单个任务失败不应拖垮整批
        return {"task": f"{sym_short} {ltf}/{h1}+{h2}", "error": repr(e)[:200]}


def stats(g: pd.DataFrame, label: str) -> dict:
    r = g["ret"].to_numpy()
    if len(r) == 0:
        return {}
    win, loss = r[r > 0], r[r <= 0]
    sd = r.std(ddof=1) if len(r) > 1 else 0.0
    return {
        "分组": label, "笔数": len(r),
        "胜率": f"{(r > 0).mean() * 100:.1f}%",
        "均收益": f"{r.mean() * 100:+.3f}%",
        "中位": f"{np.median(r) * 100:+.3f}%",
        "PF": f"{win.sum() / abs(loss.sum()):.2f}" if len(loss) and loss.sum() else "inf",
        "偏度": f"{pd.Series(r).skew():.2f}" if len(r) > 2 else "—",
        "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}" if sd else "—",
    }


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--pairs", default=DEFAULT_PAIRS)
    ap.add_argument("--symbols", default="BTC,ETH,SOL")
    ap.add_argument("--workers", type=int, default=6)
    ap.add_argument("--tag", default="main")
    args = ap.parse_args()

    pairs = [tuple(p.split(":")) for p in args.pairs.split(",") if p]
    syms = [s.strip() for s in args.symbols.split(",") if s.strip()]
    tasks = [(s, *p) for p in pairs for s in syms]
    print(f"[批量] {len(tasks)} 个任务，{args.workers} 进程并行\n", flush=True)

    results, errors = [], []
    with ProcessPoolExecutor(max_workers=args.workers) as ex:
        futs = {ex.submit(run_one, t): t for t in tasks}
        for i, f in enumerate(as_completed(futs), 1):
            r = f.result()
            t = futs[f]
            if r is None:
                print(f"  [{i}/{len(tasks)}] {t[0]} {t[1]}/{t[2]} — 样本不足", flush=True)
                continue
            if "error" in r:
                errors.append(r)
                print(f"  [{i}/{len(tasks)}] {r['task']} — 失败 {r['error']}", flush=True)
                continue
            results.append(r)
            print(f"  [{i}/{len(tasks)}] {r['task']} — 中枢 {r['n_zones']} "
                  f"信号 {r['n_sig']} 交易 {len(r['trades'])}", flush=True)

    if not results:
        print("\n无有效结果")
        return

    allt = pd.concat([r["trades"] for r in results], ignore_index=True)
    out = HERE / "out" / f"step20_{args.tag}_trades.csv"
    out.parent.mkdir(exist_ok=True)
    allt.to_csv(out, index=False)

    print(f"\n{'=' * 110}")
    print("########## 一、各级别对（全部信号 vs 大级别同向）##########")
    rows = []
    for pair, g in allt.groupby("pair"):
        rows.append(stats(g, f"{pair} 全部"))
        a1 = g["h1_agree"] == 1
        if a1.sum() >= 20:
            rows.append(stats(g[a1], f"{pair} +大级别同向"))
        if (~a1).sum() >= 20:
            rows.append(stats(g[~a1], f"{pair} 反向(对照)"))
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 二、各品种（大级别同向）##########")
    rows = []
    for (sym, pair), g in allt.groupby(["symbol", "pair"]):
        gg = g[g["h1_agree"] == 1]
        if len(gg) >= 20:
            rows.append(stats(gg, f"{sym} {pair}"))
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 三、合并总览 ##########")
    a1 = allt["h1_agree"] == 1
    a2 = allt["h2_agree"] == 1
    rows = [stats(allt, "全部"), stats(allt[a1], "+大级别1同向"),
            stats(allt[a1 & a2], "+双大级别同向"), stats(allt[~a1], "反向(对照)")]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 四、分年（大级别同向）##########")
    sub = allt[a1].copy()
    sub["year"] = pd.to_datetime(sub["date"]).dt.year
    rows = [stats(g, str(y)) for y, g in sub.groupby("year") if len(g) >= 25]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 五、尾部依赖（大级别同向）##########")
    r = sub["ret"].to_numpy()
    for k in (0, 2, 5, 10):
        v = r if k == 0 else r[r <= np.quantile(r, 1 - k / 100)]
        win, loss = v[v > 0], v[v <= 0]
        sd = v.std(ddof=1)
        print(f"  剔除最赚{k:>2}%: n={len(v):>5}  PF={win.sum() / abs(loss.sum()):.2f}  "
              f"t={v.mean() / (sd / np.sqrt(len(v))):+.2f}  "
              f"中位={np.median(v) * 100:+.3f}%")

    if errors:
        print(f"\n失败任务 {len(errors)} 个")
    print(f"\n明细已写入 {out}")


if __name__ == "__main__":
    main()