Chan/research/step37_oos_symbols.py

"""Step 37：跨品种样本外——在完全没参与过调参的币种上原样重跑。

到此为止所有参数（级别对、tol、SL/TP、过滤条件）都是在 BTC/ETH/SOL 上挑的，
即便 step36 的时间切分也仍是这三个币。本步换 8 个全新品种，
一个参数都不动，直接套用最终配置，看结论是否还在。

固定配置（不做任何搜索）：
  信号  find_fast_bsp3 默认（require_touch=False, tol=-1）
  过滤  大级别分型同向 + 中枢阶梯顺向推进
  出场  1.5 ATR 止损 / 3.0 ATR 止盈 / 最多 48 根
  成交  信号次根开盘，4bp 手续费 + 1bp 滑点
"""
from __future__ import annotations

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")
for v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
    os.environ.setdefault(v, "1")

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))
pd.set_option("display.width", 320)

SL, TP, MAXB = 1.5, 3.0, 48
FEE, SLIP = 0.0004, 0.0001
PAIRS = [("5m", "30m"), ("15m", "1h"), ("30m", "2h")]
IS_SYMS = ["BTC", "ETH", "SOL"]
OOS_SYMS = ["BNB", "XRP", "DOGE", "ADA", "AVAX", "LINK", "LTC", "TRX"]


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, ltf, htf, group = task
    pair = f"{sym}/USDT:USDT"
    try:
        df_l = fetch_ohlcv(pair, ltf, 10**9)
        if df_l is None or len(df_l) < 3000:
            return {"task": f"{sym} {ltf}", "error": "小级别数据不足"}
        chan_l = TF_DF(df_l, 1, ltf)
        cdf = chan_l.dataframe
        zones = build_htf_zones(cdf, ltf, chan=chan_l).reset_index(drop=True)
        if zones.empty:
            return {"task": f"{sym} {ltf}", "error": "无中枢"}
        z = zones.copy()
        pg, pdn = z["zg"].shift(), z["zd"].shift()
        z["z_above"], z["z_below"] = z["zd"] > pg, z["zg"] < pdn
        z["zone_i"] = np.arange(len(z))

        df_h = fetch_ohlcv(pair, htf, 10**9)
        if df_h is None or len(df_h) < 300:
            return {"task": f"{sym} {ltf}", "error": "大级别数据不足"}
        chan_h = TF_DF(df_h, 1, htf)
        s = signals_to_frame(extract_fx_signals(chan_h, chan_h.dataframe))
        tl = htf_fx_timeline(s, chan_h.dataframe)

        sig = find_fast_bsp3(cdf, zones)
        if sig.empty:
            return {"task": f"{sym} {ltf}", "error": "无信号"}
        sig = sig.merge(z[["zone_i", "z_above", "z_below"]], on="zone_i", how="left")
        sig = attach_htf_context(sig, cdf, tl, "h1")
        entries = list(zip(sig["entry_idx"].astype(int), sig["direction"].astype(int)))
        tr = run_trades(cdf, entries, SL, TP, MAXB, fee=FEE,
                        entry_delay=1, slippage=SLIP)
        if tr.empty:
            return {"task": f"{sym} {ltf}", "error": "无成交"}
        m = sig.set_index("entry_idx")
        tr["symbol"], tr["ltf"], tr["group"] = sym, ltf, group
        tr["date"] = cdf["date"].to_numpy()[tr["entry_idx"].to_numpy()]
        for c in ("h1_agree", "z_above", "z_below", "direction"):
            tr[c if c != "direction" else "dir_sig"] = tr["entry_idx"].map(m[c])
        return {"task": f"{sym} {ltf}", "trades": tr}
    except Exception as e:
        return {"task": f"{sym} {ltf}", "error": repr(e)[:200]}


def stat(g: pd.DataFrame, label: str, minn: int = 25) -> dict:
    r = g["ret"].to_numpy()
    if len(r) < minn:
        return {}
    w, o = r[r > 0], r[r <= 0]
    sd = r.std(ddof=1)
    t10 = r[r <= np.quantile(r, 0.90)]
    yrs = max((g["date"].max() - g["date"].min()).days / 365.25, 0.25)
    return {"分组": label, "笔数": len(r), "年笔数": round(len(r) / yrs),
            "胜率": f"{(r > 0).mean() * 100:.1f}%",
            "中位": f"{np.median(r) * 100:+.3f}%",
            "PF": f"{w.sum() / abs(o.sum()):.2f}" if len(o) else "inf",
            "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}",
            "剔10%PF": f"{t10[t10 > 0].sum() / abs(t10[t10 <= 0].sum()):.2f}"}


def main() -> None:
    tasks = ([(s, l, h, "样本内") for s in IS_SYMS for l, h in PAIRS]
             + [(s, l, h, "样本外") for s in OOS_SYMS for l, h in PAIRS])
    print(f"[跨品种样本外] {len(tasks)} 个任务\n", flush=True)
    res, skip = [], []
    with ProcessPoolExecutor(max_workers=5) 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()
            if r is None or "error" in (r or {}):
                skip.append(f"{(r or {}).get('task', '?')}: {(r or {}).get('error', '')}")
                continue
            res.append(r)
            print(f"  [{i}/{len(tasks)}] {r['task']} — {len(r['trades'])} 笔", flush=True)
    if skip:
        print(f"\n  跳过 {len(skip)} 个：")
        for s in skip:
            print(f"    {s}")
    if not res:
        return
    allt = pd.concat([r["trades"] for r in res], ignore_index=True)
    allt["date"] = pd.to_datetime(allt["date"])
    allt["push"] = np.where(allt["dir_sig"] == 1, allt["z_above"], allt["z_below"])
    allt["push"] = allt["push"].fillna(False).astype(bool)
    fin = allt[(allt["h1_agree"] == 1) & allt["push"]].copy()
    fin.to_csv(HERE / "out" / "step37_oos.csv", index=False)

    print("\n" + "=" * 116)
    print("########## 1. 样本内（调参用的三个币）vs 样本外（八个全新币）##########")
    rows = [stat(g, k) for k, g in fin.groupby("group")]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))
    print("  参数一个没动。若样本外仍显著为正，说明不是在三个币上过拟合。")

    print("\n########## 2. 样本外逐个品种 ##########")
    oos = fin[fin["group"] == "样本外"]
    rows = [stat(g, s, minn=20) for s, g in oos.groupby("symbol")]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 3. 样本外分级别 ##########")
    rows = [stat(g, f"样本外 {k}", minn=20) for k, g in oos.groupby("ltf")]
    rows += [stat(g, f"样本内 {k}", minn=20)
             for k, g in fin[fin["group"] == "样本内"].groupby("ltf")]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 4. 样本外多空 ##########")
    rows = [stat(oos[oos["dir_sig"] == d], n, minn=20)
            for d, n in ((1, "做多"), (-1, "做空"))]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 5. 样本外分年 ##########")
    o = oos.copy()
    o["y"] = o["date"].dt.year
    rows = [stat(g, str(y), minn=20) for y, g in o.groupby("y")]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 6. 资金曲线（固定名义1倍，不复利，已扣 5bp）##########")
    for k, g in fin.groupby("group"):
        g = g.sort_values("date")
        r = g["ret"].to_numpy()
        yrs = (g["date"].max() - g["date"].min()).days / 365.25
        eq = 1 + np.cumsum(r)
        dd = (np.maximum.accumulate(eq) - eq).max()
        print(f"  {k}: n={len(r):>4} 年{len(r) / yrs:>3.0f}笔  "
              f"年化 {(eq[-1] - 1) / yrs * 100:+6.1f}%  回撤 {dd * 100:5.1f}%  "
              f"Sharpe {r.mean() / r.std(ddof=1) * np.sqrt(len(r) / yrs):5.2f}")


if __name__ == "__main__":
    main()