Chan/research/step56_fast_bsp1.py

"""实时版一类买卖点（fast B1/S1）能否翻正。

step55 已判定引擎的 B1/S1 不可用（5m PF 0.24/0.20、胜率 18.6%/14.8%、
t −17/−22），且病根是滞后 8~9 根带来的几何劣势，不是参数。B3 当年同病
（PF 0.66），靠重新定义成实时判据翻到 1.59（B4）。本脚本对一类做同样的尝试。

对照组三条，缺一不可：
  引擎 B1/S1   step55 的数，说明「不改判据」是什么下场
  fast B1/S1   本次
  fast B3/S3   同一份数据、同一套出场跑 B4，确认管线本身能跑出正数
               —— 少了它，fast B1 若为负就分不清是判据不行还是管线接错了
"""
from __future__ import annotations

import argparse
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")
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))

OUT = HERE / "out" / "step56_fast_bsp1.feather"
SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48


def collect(sym: str, rows: int, tf: str) -> pd.DataFrame | None:
    from chanlun import TF_DF
    from chanlun.analysis.fast_bsp import (
        ensure_timestamp,
        find_fast_bsp3,
        zones_from_zs_list,
    )
    from lib.data import fetch_ohlcv
    from lib.exit_model import cfg_name, walk_exits
    from lib.fast_bsp1 import find_fast_bsp1, zones_with_enter_area

    try:
        df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows)
        if df is None or len(df) < 5_000:
            return None
        chan = TF_DF(df, 1, tf)
        cdf = ensure_timestamp(chan.dataframe)
        zs_list = chan.cal_bi_zs_list_pure(chan.bi_list)
        if not zs_list:
            return None

        atr = cdf["atr"].to_numpy(float)
        cl = cdf["close"].to_numpy(float)
        cfg = cfg_name(SL, RUNNER, MAXB, RSTOP)
        parts = []

        d1 = {}
        s1 = find_fast_bsp1(cdf, zones_with_enter_area(zs_list, cdf), diag=d1)
        if not s1.empty:
            s1["kind"] = "fastB1"
            parts.append(s1)
        # 同数据同出场跑一遍 B4，作为「管线能出正数」的存在性证明
        s3 = find_fast_bsp3(cdf, zones_from_zs_list(zs_list, cdf))
        if not s3.empty:
            s3["kind"] = "fastB3"
            parts.append(s3)
        if not parts:
            return None

        r = pd.concat(parts, ignore_index=True)
        r = r[(r.entry_idx < len(cdf) - 2)
              & np.isfinite(atr[r.entry_idx.values])
              & (atr[r.entry_idx.values] > 0)].reset_index(drop=True)
        if r.empty:
            return None
        res = walk_exits(cdf, r[["entry_idx", "direction"]], [SL], [RUNNER],
                         [MAXB], scale_at=SCALE_AT, runners=(RUNNER,),
                         runner_stops=(RSTOP,))
        if len(res) != len(r):
            return None
        for k in ("g", "r", "c", "b"):
            r[k] = res[f"{cfg}_{k}"].to_numpy()
        r["sym"], r["tf"] = sym, tf
        r["atr_pct"] = atr[r.entry_idx.values] / cl[r.entry_idx.values]
        r["date"] = cdf["date"].to_numpy()[r.entry_idx.values]
        r["diag"] = str(d1)
        return r
    except Exception as e:                                    # noqa: BLE001
        print(f"  {sym} {tf} 失败: {type(e).__name__}: {e}", flush=True)
        return None


def stats(g: pd.DataFrame) -> dict:
    from lib.exit_model import fee_of, taker_notional

    net = g.g.values - fee_of(g.r.values, g.c.values)
    gR = g.g.values / (SL * g.atr_pct.values)
    R = net / (SL * g.atr_pct.values)
    tn = taker_notional(g.r.values, g.c.values)
    w, o = net[net > 0].sum(), -net[net <= 0].sum()
    return {
        "笔数": len(g), "胜率": f"{(net > 0).mean()*100:.1f}%",
        "毛R": round(gR.mean(), 3), "净均R": round(R.mean(), 3),
        "PF": round(w / o, 2) if o > 0 else np.inf,
        "R夏普": round(R.mean() / R.std(ddof=1), 3),
        "余量bp": round(net.mean() / tn.mean() * 1e4, 2),
        "中位持仓": int(np.median(g.b.values)),
        "t值": round(gR.mean() / (gR.std(ddof=1) / np.sqrt(len(g))), 2),
    }


def report(d: pd.DataFrame) -> None:
    for tf, x in d.groupby("tf"):
        span = (pd.to_datetime(x.date).max()
                - pd.to_datetime(x.date).min()).total_seconds() / 86400
        print("\n" + "#" * 92)
        print(f"##########  {tf}  ·  {x.sym.nunique()} 币 · 跨 {span:.0f} 天"
              f"  ##########")
        rows = []
        for (kind, dirn), g in x.groupby(["kind", "direction"]):
            if len(g) < 40:
                continue
            nm = {"fastB1": ("一买", "一卖"), "fastB3": ("三买", "三卖")}[kind]
            rows.append({"信号": f"{kind} {nm[0 if dirn == 1 else 1]}",
                         "每币每天": round(len(g) / span / x.sym.nunique(), 3),
                         **stats(g)})
        if rows:
            print(pd.DataFrame(rows).to_string(index=False))
        print("\n对照 · step55 引擎原生（同周期同出场）：")
        print("  5m   B1 PF 0.24 胜率 18.6% t −17.3 ｜ S1 PF 0.20 胜率 14.8% "
              "t −22.3")
        print("  15m  B1 PF 0.21 t −18.9 ｜ S1 PF 0.23 t −15.9")

        f1 = x[x.kind == "fastB1"]
        if len(f1) >= 60:
            print("\n背驰强度分档（div = 离开段面积/进入段面积，越小背驰越强）：")
            q = pd.qcut(f1["div"], 4, labels=["Q1最强", "Q2", "Q3", "Q4最弱"],
                        duplicates="drop")
            print(pd.DataFrame([{"档": k, **stats(v)}
                                for k, v in f1.groupby(q, observed=True)
                                if len(v) >= 20]).to_string(index=False))
            print("\n滞后分档（lag = 入场根 − 离开段极值根）：")
            b = pd.cut(f1["lag"], [-1, 1, 3, 6, 12, 1e9],
                       labels=["≤1根", "2-3根", "4-6根", "7-12根", ">12根"])
            print(pd.DataFrame([{"档": k, **stats(v)}
                                for k, v in f1.groupby(b, observed=True)
                                if len(v) >= 20]).to_string(index=False))

    dg = d[d.kind == "fastB1"].diag.dropna()
    if len(dg):
        print("\n" + "=" * 92)
        print("fast B1 漏斗（首个中枢样本）：", dg.iloc[0])


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE")
    ap.add_argument("--tfs", default="5m,15m")
    ap.add_argument("--rows", type=int, default=200_000)
    ap.add_argument("--workers", type=int, default=3)
    ap.add_argument("--reuse", action="store_true")
    args = ap.parse_args()

    if args.reuse and OUT.exists():
        d = pd.read_feather(OUT)
    else:
        syms = [s.strip() for s in args.symbols.split(",")]
        tfs = [t.strip() for t in args.tfs.split(",")]
        print(f"[实时版一类] {len(syms)} 币 × {tfs} × {args.rows} 根\n",
              flush=True)
        parts = []
        with ProcessPoolExecutor(max_workers=args.workers) as ex:
            fut = {ex.submit(collect, s, args.rows, t): (s, t)
                   for s in syms for t in tfs}
            for i, f in enumerate(as_completed(fut), 1):
                r = f.result()
                s, t = fut[f]
                print(f"  [{i}/{len(fut)}] {s} {t} "
                      f"{0 if r is None else len(r)}", flush=True)
                if r is not None:
                    parts.append(r)
        if not parts:
            print("无结果")
            return
        d = pd.concat(parts, ignore_index=True)
        OUT.parent.mkdir(exist_ok=True)
        d.to_feather(OUT)
    report(d)


if __name__ == "__main__":
    main()