Chan/research/step28_pair_matrix.py

"""Step 28：小级别 × 大级别 全矩阵扫描。

此前配对都是拍脑袋定的（5m配15m、15m配1h），而远距实验里 5m 配 1h 明显好于配 15m，
说明「隔几级去挂分型」本身是个从没调过的参数。本步把它扫完。

效率关键：一笔交易的入场与出场只由小级别决定，大级别只改变过滤标签。
所以每个小级别只回测一次，各大级别的同向标记以列的形式附加上去，
18 个组合的成本接近 4 个。
"""
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")
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
LTFS = ["5m", "15m", "30m", "1h"]
HTFS = ["15m", "30m", "1h", "2h", "4h", "1d", "1w"]
ORDER = {t: i for i, t in enumerate(["1m", "5m", "15m", "30m", "1h", "2h", "4h", "1d", "1w"])}


def run_symbol(sym: str) -> 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

    pair = f"{sym}/USDT:USDT"
    frames = []
    try:
        # 大级别分型时间线：每个周期只算一次
        timelines = {}
        for tf in HTFS:
            df_h = fetch_ohlcv(pair, 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))
            timelines[tf] = htf_fx_timeline(s, chan_h.dataframe)

        for ltf in LTFS:
            df_l = fetch_ohlcv(pair, ltf, 10**9)
            if df_l is None or len(df_l) < 3000:
                continue
            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:
                continue
            sig = find_fast_bsp3(cdf, zones)
            if sig.empty or len(sig) < 20:
                continue

            # 中枢阶梯方向（只依赖小级别，与大级别无关）
            z = zones.copy()
            pg, pdn = z["zg"].shift(), z["zd"].shift()
            z["z_above"] = z["zd"] > pg
            z["z_below"] = z["zg"] < pdn
            z["zone_i"] = np.arange(len(z))
            sig = sig.merge(z[["zone_i", "z_above", "z_below"]], on="zone_i", how="left")

            # 各大级别的同向标记，逐列附加
            for tf in HTFS:
                if ORDER[tf] <= ORDER[ltf] or tf not in timelines:
                    continue
                sig = attach_htf_context(sig, cdf, timelines[tf], f"x{tf}")

            entries = list(zip(sig["entry_idx"].astype(int), sig["direction"].astype(int)))
            tr = run_trades(cdf, entries, SL, TP, MAXB, fee=0.0, entry_delay=1)
            if tr.empty:
                continue
            m = sig.drop_duplicates("entry_idx").set_index("entry_idx")
            tr["symbol"], tr["ltf"] = sym, ltf
            tr["date"] = cdf["date"].to_numpy()[tr["entry_idx"].to_numpy()]
            for c in m.columns:
                if c.endswith("_agree") or c in ("z_above", "z_below", "direction"):
                    tr[c if c != "direction" else "dir_sig"] = tr["entry_idx"].map(m[c])
            frames.append(tr)
        if not frames:
            return None
        return {"sym": sym, "trades": pd.concat(frames, ignore_index=True)}
    except Exception as e:
        return {"sym": sym, "error": repr(e)[:300]}


def stat(r: np.ndarray) -> dict:
    if len(r) < 30:
        return {}
    w, o = r[r > 0], r[r <= 0]
    sd = r.std(ddof=1)
    return {
        "笔数": len(r), "胜率": (r > 0).mean() * 100,
        "均收益": r.mean() * 100, "中位": np.median(r) * 100,
        "PF": w.sum() / abs(o.sum()) if len(o) else np.inf,
        "偏度": pd.Series(r).skew(),
        "t值": r.mean() / (sd / np.sqrt(len(r))),
    }


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbols", default="BTC,ETH,SOL")
    ap.add_argument("--reuse", action="store_true")
    args = ap.parse_args()

    cache = HERE / "out" / "step28_matrix.csv"
    if args.reuse and cache.exists():
        allt = pd.read_csv(cache, parse_dates=["date"])
        print(f"[复用] {len(allt)} 笔\n")
    else:
        syms = [s.strip() for s in args.symbols.split(",") if s.strip()]
        print(f"[全矩阵] {len(syms)} 个品种 × {len(LTFS)} 小级别 × {len(HTFS)} 大级别\n",
              flush=True)
        res = []
        with ProcessPoolExecutor(max_workers=len(syms)) as ex:
            futs = {ex.submit(run_symbol, s): s for s in syms}
            for f in as_completed(futs):
                r = f.result()
                if r is None or "error" in (r or {}):
                    print(f"  {futs[f]} 失败 {(r or {}).get('error', '')}", flush=True)
                    continue
                res.append(r)
                print(f"  {r['sym']} 完成 — {len(r['trades'])} 笔", flush=True)
        if not res:
            return
        allt = pd.concat([r["trades"] for r in res], ignore_index=True)
        allt.to_csv(cache, index=False)

    allt["date"] = pd.to_datetime(allt["date"])
    allt["ret_net"] = allt["gross"] - FEE - SLIP
    allt["push"] = np.where(allt["dir_sig"] == 1, allt["z_above"], allt["z_below"])
    allt["push"] = allt["push"].fillna(False).astype(bool)

    print("=" * 118)
    print("########## 1. PF 矩阵：小级别(行) × 大级别分型(列)，仅同向 ##########")
    for metric in ("PF", "t值", "中位"):
        grid = {}
        for ltf, g in allt.groupby("ltf"):
            for tf in HTFS:
                col = f"x{tf}_agree"
                if col not in g.columns:
                    continue
                s = stat(g[g[col] == 1]["ret_net"].to_numpy())
                if s:
                    grid.setdefault(ltf, {})[tf] = s[metric]
        if not grid:
            continue
        df = pd.DataFrame(grid).T
        df = df.reindex(index=[t for t in LTFS if t in df.index],
                        columns=[t for t in HTFS if t in df.columns])
        print(f"\n-- {metric} --")
        print(df.round(2).to_string())

    print("\n########## 2. 叠加中枢阶梯方向过滤后的 PF ##########")
    grid = {}
    for ltf, g in allt[allt.push].groupby("ltf"):
        for tf in HTFS:
            col = f"x{tf}_agree"
            if col not in g.columns:
                continue
            s = stat(g[g[col] == 1]["ret_net"].to_numpy())
            if s:
                grid.setdefault(ltf, {})[tf] = s["PF"]
    df = pd.DataFrame(grid).T
    df = df.reindex(index=[t for t in LTFS if t in df.index],
                    columns=[t for t in HTFS if t in df.columns])
    print(df.round(2).to_string())

    print("\n########## 3. 全部组合排行（按 t 值，含笔数下限）##########")
    rows = []
    for ltf, g in allt.groupby("ltf"):
        for tf in HTFS:
            col = f"x{tf}_agree"
            if col not in g.columns:
                continue
            for pf_name, sub in (("同向", g[g[col] == 1]),
                                 ("同向+阶梯", g[(g[col] == 1) & g.push])):
                s = stat(sub["ret_net"].to_numpy())
                if s and s["笔数"] >= 80:
                    rows.append({"组合": f"{ltf}/{tf} {pf_name}", **s})
    r = pd.DataFrame(rows).sort_values("t值", ascending=False)
    for c in ("胜率", "均收益", "中位", "PF", "偏度", "t值"):
        r[c] = r[c].round(2)
    print(r.head(25).to_string(index=False))

    print("\n########## 4. 每个小级别的最佳搭档 ##########")
    best = []
    for ltf in LTFS:
        sub = r[r["组合"].str.startswith(f"{ltf}/")]
        if not sub.empty:
            best.append(sub.iloc[0])
    if best:
        print(pd.DataFrame(best).to_string(index=False))


if __name__ == "__main__":
    main()