Chan/research/step65_room.py

"""改识别规则之前，先量机会本身装不装得下确认成本。

用户问：改 B1/B2 的识别规则行不行？

「改识别规则」有两种含义，第一种已经被测过上限：
  (a) 换规则**挑**出更好的 B1 —— step60 用未来函数只留极值落在真线段底 ±2 根内
      的信号，这是任何识别规则的理论最好情况，PF 仅 0.70~0.83。这一类改法封死。

但 0.83 这个上限本身是**结果**，背后是一个从没直接测过的量：

    一段线段从真底走到真顶，一共有几个 ATR？而等笔确认要花掉 2.9 个。

这决定了 (b) 类改法（改**入场时点/构造**，而非改选样）有没有空间：

  线段幅度 4 ATR  -> 进场吃掉 2.9，剩 1.1 去扛 2 ATR 止损，**任何规则都救不活**
  线段幅度 12 ATR -> 2.9 只是小费，值得继续改规则

注意这测的是**市场**，不是我们的检测器：标准答案直接取 `seg_list` 的真实端点，
完全不经过 `find_all_bsp`。所以结论对「换任何一套识别规则」都成立。

而且这个比值大概率**随级别变化**（ATR 与线段幅度未必同比例缩放），所以跑
5m/15m/1h/4h。真正的产出不是「改不改规则」，而是**「在哪个级别上改才有意义」**。

预算恒等式（每根线段一行）：
    可捕获 = 线段幅度 − 确认成本      需要 > 止损 才有正期望的可能
"""
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" / "step65_room.feather"
SL_ATR = 2.0


def collect(sym: str, tf: str, rows: int) -> pd.DataFrame | None:
    from chanlun import TF_DF
    from chanlun.core.ChanEnum import Chan_SEG_DIR
    from lib.data import fetch_ohlcv

    try:
        df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows)
        if df is None or len(df) < 3_000:
            return None
        chan = TF_DF(df, 1, tf, lean=False)
        cdf = chan.dataframe
        segs = [s for s in (getattr(chan, "seg_list", []) or [])
                if s.end_time is not None and s.start_time is not None]
        if len(segs) < 20:
            return None

        dser = pd.to_datetime(cdf["date"])
        if dser.dt.tz is not None:
            dser = dser.dt.tz_localize(None)
        didx = pd.DatetimeIndex(dser)
        n = len(cdf)
        atr = cdf["atr"].to_numpy(float)
        hi, lo = cdf["high"].to_numpy(float), cdf["low"].to_numpy(float)

        def to_i(ts) -> int:
            t = pd.Timestamp(ts)
            return int(didx.searchsorted(t.tz_localize(None) if t.tz else t))

        rec = []
        for sg in segs:
            i0, i1 = to_i(sg.start_time), to_i(sg.end_time)
            if not (0 <= i0 < i1 < n) or i1 - i0 < 2:
                continue
            a = atr[i0]
            if not np.isfinite(a) or a <= 0:
                continue
            up = sg.dir != Chan_SEG_DIR.DOWN
            # 从线段起点（上一段的真实反转点）到终点的幅度
            span = ((hi[i0:i1 + 1].max() - lo[i0]) if up
                    else (hi[i0] - lo[i0:i1 + 1].min()))
            rec.append({
                "sym": sym, "tf": tf, "dir": 1 if up else -1,
                "bars": i1 - i0,
                "span_atr": span / a,
                "atr_pct": a / float(cdf["close"].to_numpy(float)[i0]),
            })
        return pd.DataFrame(rec) if rec else None
    except Exception as e:                                    # noqa: BLE001
        print(f"  {sym} {tf} 失败: {type(e).__name__}: {e}", flush=True)
        return None


TFO = {"5m": 0, "15m": 1, "1h": 2, "4h": 3}


def report(d: pd.DataFrame, cost: float) -> None:
    print("\n" + "=" * 96)
    print(f"【一】线段幅度 vs 确认成本（成本按一类实测的 {cost} ATR 计）")
    print("=" * 96)
    rows = []
    for tf, x in sorted(d.groupby("tf"), key=lambda kv: TFO.get(kv[0], 9)):
        q = x.span_atr.quantile([.25, .5, .75]).values
        room = x.span_atr - cost
        rows.append({
            "级别": tf, "线段数": len(x),
            "幅度Q1": round(q[0], 1), "幅度中位": round(q[1], 1),
            "幅度Q3": round(q[2], 1),
            "确认后剩余(中位)": round(q[1] - cost, 1),
            f"剩余>止损{SL_ATR}": f"{(room > SL_ATR).mean()*100:.0f}%",
            "剩余/止损": round((q[1] - cost) / SL_ATR, 2),
            "中位时长(根)": int(x.bars.median()),
        })
    print(pd.DataFrame(rows).to_string(index=False))
    print(f"""
  「剩余/止损」是这张表的结论行：真底进场、扣掉确认成本后，还剩几倍止损的空间。
    < 1  机会装不下确认成本，**换任何识别规则都没用**
    1~2  勉强打平，要求选样精度极高（step60 实测最好 30%）
    > 2  有空间，值得改规则/改入场构造""")

    print("\n【二】把确认成本当变量：多低才够用")
    rows = []
    for tf, x in sorted(d.groupby("tf"), key=lambda kv: TFO.get(kv[0], 9)):
        r = {"级别": tf}
        for c in (0.0, 1.0, 2.0, 2.9):
            r[f"成本{c}"] = round((x.span_atr.median() - c) / SL_ATR, 2)
        rows.append(r)
    print(pd.DataFrame(rows).to_string(index=False))
    print("\n  成本 0 = 完美实时（在真底那根就进）。若连成本 0 那列都 < 2，"
          "\n  说明**不是滞后的问题，是这个级别的线段本身就太小**。")

    print("\n【三】逐币（中位幅度 ATR），看结论是否普适")
    print(d.pivot_table(index="sym", columns="tf", values="span_atr",
                        aggfunc="median").round(1)
          .reindex(columns=[t for t in TFO if t in set(d.tf)])
          .to_string())


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

    if args.reuse and OUT.exists():
        report(pd.read_feather(OUT), args.cost)
        return
    syms = [s.strip() for s in args.symbols.split(",")]
    tfs = [t.strip() for t in args.tfs.split(",")]
    parts = []
    with ProcessPoolExecutor(max_workers=args.workers) as ex:
        fut = {ex.submit(collect, s, t, args.rows): (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, args.cost)


if __name__ == "__main__":
    main()