Chan/research/step66_ceiling.py

"""改识别规则到底有没有用：把天花板一次性测到顶。

用户问「改 B1/B2 的识别规则呢」。step65 让这个问题变得可判定，但也制造了一个
必须解释的矛盾：

    线段中位 11.4 ATR，进场花掉 2.9，剩 8.5 —— 是 2 ATR 止损的 4.26 倍。
    可 step60 用未来函数选出真底的那批 B1，PF 仍只有 0.70~0.83。

空间明明在，就是拿不到。两种解释对用户的问题给出**相反**的答案：

  选样问题：只有 30% 的 B1 落在真线段底，那 8.5 ATR 只存在于这 30% 里，
            其余 70% 是趋势中途，后面根本没有行情   -> **改识别规则有用**
  路径问题：就算落在真底，11.4 ATR 是净幅度，路上的回撤照样打掉止损
                                                 -> **改识别规则没用**

区分只需要一个从没跑过的组合：**未来函数选样 + 结构止损 + 绝对目标**。
三样单独都失败过（step60 的 0.83 / step58 的 0.36 / 本轮 abs 的 0.36），
组合起来没试过。它测的不是某条规则，是**所有识别规则的上界**：

    上界 < 1.0  -> 路径问题，一类线彻底关闭，别再改规则
    上界 > 1.5  -> 选样问题，改识别规则有用，且这里就是要够到的目标

同时量 MFE 做交叉验证：真底那批到底跑没跑出 8.5 ATR。
  跑出来了却还是亏 -> 铁证是止损/路径
  根本没跑出来     -> 说明 11.4 ATR 那个测量传导不到这批信号上，step65 要重解释
"""
from __future__ import annotations

import argparse
import sys
import warnings
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))

SRC = HERE / "out" / "step64_b2.feather"
MAXB = 48


def mfe_mae(cdf: pd.DataFrame, idx: np.ndarray, dirs: np.ndarray,
            atr: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
    """入场后 MAXB 根内最远的顺向/逆向行程（ATR）。"""
    hi, lo = cdf["high"].to_numpy(float), cdf["low"].to_numpy(float)
    op = cdf["open"].to_numpy(float)
    n = len(cdf)
    fe = np.full(len(idx), np.nan)
    ae = np.full(len(idx), np.nan)
    for k, (i, d, a) in enumerate(zip(idx, dirs, atr)):
        e = int(i) + 1
        if e >= n or not np.isfinite(a) or a <= 0:
            continue
        j = min(e + MAXB, n)
        px = op[e]
        if d == 1:
            fe[k] = (hi[e:j].max() - px) / a
            ae[k] = (px - lo[e:j].min()) / a
        else:
            fe[k] = (px - lo[e:j].min()) / a
            ae[k] = (hi[e:j].max() - px) / a
    return fe, ae


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--tf", default="5m")
    ap.add_argument("--rows", type=int, default=200_000)
    args = ap.parse_args()

    from chanlun import TF_DF
    from lib.data import fetch_ohlcv
    from step58_struct_stop import simulate, stats

    d = pd.read_feather(SRC)
    d = d[(d.tf == args.tf) & d.type.isin(["B1", "S1"])].copy()
    d["to_ext"] = (d.entry_px - d.own_ext) * d.dir / d.atr_at_entry
    print(f"[识别规则天花板] {args.tf} · 一类 {len(d)} 笔 · "
          f"其中落在真线段端点 {d.hit.sum()} 笔 ({d.hit.mean()*100:.1f}%)\n")

    cache = {}
    for sym in sorted(d.sym.unique()):
        df = fetch_ohlcv(f"{sym}/USDT:USDT", args.tf, args.rows)
        cache[sym] = TF_DF(df, 1, args.tf).dataframe

    print("=" * 96)
    print("【一】MFE 交叉验证：真底那批到底跑没跑出 step65 说的 8.5 ATR")
    print("=" * 96)
    rows = []
    for k, g in d.groupby(d.hit):
        fe, ae = [], []
        for sym, x in g.groupby("sym"):
            f_, a_ = mfe_mae(cache[sym], x.i_sure.values, x.dir.values,
                             x.atr_at_entry.values)
            fe.append(f_)
            ae.append(a_)
        fe, ae = np.concatenate(fe), np.concatenate(ae)
        ok = np.isfinite(fe)
        rows.append({
            "落在真线段端点": "是" if k else "否", "笔数": int(ok.sum()),
            "MFE中位": round(float(np.median(fe[ok])), 2),
            "MFE≥3ATR": f"{(fe[ok] >= 3).mean()*100:.0f}%",
            "MFE≥8ATR": f"{(fe[ok] >= 8).mean()*100:.0f}%",
            "MAE中位": round(float(np.median(ae[ok])), 2),
            "MAE≥2ATR(会被现止损打掉)": f"{(ae[ok] >= 2).mean()*100:.0f}%",
        })
    print(pd.DataFrame(rows).to_string(index=False))
    print("""
  判读：若「是」那行 MFE 中位远低于 8.5，说明 step65 的线段幅度传导不到
  信号上（进场时行情已经走掉了大半）；若 MFE 够大而 MAE 也大，则是路径问题。""")

    print("\n" + "=" * 96)
    print("【二】天花板：未来函数选样 + 结构止损 + 绝对目标（三者首次组合）")
    print("=" * 96)
    rows = []
    for k, g in d.groupby(d.hit):
        for label, margin in [("固定2ATR", None), ("结构+0.5", 0.5),
                              ("结构+1.0", 1.0), ("结构+1.5", 1.5)]:
            gs, rs, cs, ap_, sl_ = [], [], [], [], []
            for sym, x in g.groupby("sym"):
                x = x.reset_index(drop=True)
                t = pd.DataFrame({"entry_idx": x.i_sure.values,
                                  "direction": x.dir.values,
                                  "atr_at_entry": x.atr_at_entry.values})
                sl = (np.full(len(x), 2.0) if margin is None
                      else (x.to_ext.values + margin))
                res = simulate(cache[sym], t, sl, r_scale=False)
                ok = res.g.notna().values
                gs.append(res.g[ok])
                rs.append(res.r[ok])
                cs.append(res.c[ok])
                ap_.append((x.atr_at_entry.values / x.entry_px.values)[ok])
                sl_.append(sl[ok])
            rows.append({
                "落在真线段端点": "是" if k else "否", "止损": label,
                **stats(pd.concat(gs, ignore_index=True),
                        pd.concat(rs, ignore_index=True),
                        pd.concat(cs, ignore_index=True),
                        np.concatenate(ap_), np.concatenate(sl_))})
    print(pd.DataFrame(rows).to_string(index=False))
    print("""
  「是」那几行就是**任何识别规则的上界**（因为选样已经用了未来函数）：
    < 1.0  路径问题，一类线关闭，改规则无解
    > 1.5  选样问题，改识别规则有用，且这就是要够到的目标""")


if __name__ == "__main__":
    main()