Chan/research/step8_bi_endpoint_fx.py

"""Step 8：把问题收敛到「能否提前认出笔端点分型」。

Step 7 表明：原始 KLC 分型太密（每 2.9 根一个），是纯噪声；
但笔端点分型在事后看是真正的转折。二者的差别决定了优化空间：

  - 若「笔端点分型 + 分型确认时刻入场」有显著 alpha，
    问题就变成一个实时判别任务：在分型刚确认的第 1~2 根，
    预测它会不会成为笔端点。滞后可从 9~10 根压到 1~2 根。
  - 若连笔端点分型都没有 alpha，那这条路直接否掉。
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

import numpy as np
import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parent))

from lib.bsp_eval import baseline_stats
from lib.data import fetch_ohlcv
from lib.fx_signal import add_forward_returns, extract_fx_signals, signals_to_frame

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from chanlun import TF_DF
from chanlun.core.ChanEnum import Chan_BI_DIR

pd.set_option("display.width", 240)
HORIZONS = (3, 5, 10, 20, 40)


def summarize(g: pd.DataFrame, df: pd.DataFrame, label: str, min_n: int = 15) -> list[dict]:
    base = baseline_stats(df, HORIZONS).set_index("horizon")
    out = []
    for h in HORIZONS:
        col = f"ret_{h}"
        r = g[col].dropna().to_numpy() if col in g else np.array([])
        if len(r) < min_n:
            continue
        dirs = g.loc[g[col].notna(), "direction"].to_numpy()
        sd = r.std(ddof=1)
        out.append({
            "分组": label, "持有": h, "n": len(r),
            "收益": r.mean(), "胜率": (r > 0).mean(),
            "超额": r.mean() - float(np.mean(dirs) * base.loc[h, "base_mean_long"]),
            "t值": r.mean() / (sd / np.sqrt(len(r))) if sd else np.nan,
        })
    return out


def show(rows: list[dict]) -> None:
    if not rows:
        print("  (样本不足)")
        return
    d = pd.DataFrame(rows)
    d["收益"] = d["收益"].map(lambda v: f"{v * 100:+.2f}%")
    d["超额"] = d["超额"].map(lambda v: f"{v * 100:+.2f}%")
    d["胜率"] = d["胜率"].map(lambda v: f"{v * 100:.0f}%")
    d["t值"] = d["t值"].map(lambda v: f"{v:+.2f}")
    print(d.to_string(index=False))


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbol", default="BTC/USDT:USDT")
    ap.add_argument("--tf", default="1h")
    args = ap.parse_args()

    df = fetch_ohlcv(args.symbol, args.tf, 10**9)
    chan = TF_DF(df, 1, args.tf)
    cdf = chan.dataframe
    idx_of = {t: i for i, t in enumerate(cdf["date"].dt.strftime("%Y-%m-%d %H:%M:%S"))}

    sig = signals_to_frame(extract_fx_signals(chan, cdf))
    sig = add_forward_returns(sig, cdf, HORIZONS)

    # 标注哪些分型最终成为了笔端点（事后信息，仅用于确认 alpha 是否存在）
    endpoint_idx: set[int] = set()
    bi_lag: dict[int, int] = {}
    for bi in chan.bi_list:
        for klc in (bi.start_klc, bi.end_klc):
            if klc is None:
                continue
            k = str(getattr(klc, "end_time", ""))
            if k in idx_of:
                endpoint_idx.add(idx_of[k])
        e = str(getattr(bi, "end_time", ""))
        s = str(getattr(bi, "sure_time", "") or "")
        if bi.is_sure and e in idx_of and s in idx_of:
            bi_lag[idx_of[e]] = idx_of[s] - idx_of[e]

    sig["is_endpoint"] = sig["fx_idx"].isin(endpoint_idx)
    sig["bi_confirm_lag"] = sig["fx_idx"].map(bi_lag)

    n_ep = int(sig["is_endpoint"].sum())
    print(f"[样本] 分型 {len(sig)}  其中笔端点 {n_ep} ({n_ep / len(sig) * 100:.1f}%)")
    print(f"[滞后] 分型确认 {sig['lag'].median():.0f} 根 / "
          f"笔确认 {sig['bi_confirm_lag'].median():.0f} 根 "
          f"→ 可压缩 {sig['bi_confirm_lag'].median() - sig['lag'].median():.0f} 根\n")

    print("########## 核心对照：笔端点分型 vs 普通分型（均以分型确认时刻入场）##########")
    rows = summarize(sig[sig.is_endpoint], cdf, "笔端点分型")
    rows += summarize(sig[~sig.is_endpoint], cdf, "非端点分型")
    show(rows)

    print("\n########## 笔端点分型 多空拆分 ##########")
    ep = sig[sig.is_endpoint]
    rows = summarize(ep[ep.direction == 1], cdf, "端点做多")
    rows += summarize(ep[ep.direction == -1], cdf, "端点做空")
    show(rows)

    print("\n########## 对照：同一批端点，改用笔确认时刻入场（滞后 9~10 根）##########")
    closes = cdf["close"].to_numpy(dtype=float)
    n = len(cdf)
    late = ep.dropna(subset=["bi_confirm_lag"]).copy()
    for h in HORIZONS:
        vals = []
        for _, r in late.iterrows():
            i = int(r["fx_idx"]) + int(r["bi_confirm_lag"])
            j = i + h
            vals.append(
                int(r["direction"]) * (closes[j] - closes[i]) / closes[i]
                if j < n and i < n else np.nan
            )
        late[f"ret_{h}"] = vals
    show(summarize(late, cdf, "端点@笔确认时刻"))

    print("\n########## 实时可得特征对端点的判别力 ##########")
    # 这些特征在分型确认当根就全部已知，可用于实时预测
    feats = {
        "背驰 is_divergence": sig["is_divergence"].astype(bool),
        "面积比 ratio<0.5": sig["ratio"] < 0.5,
        "面积比 ratio<0.8": sig["ratio"] < 0.8,
        "确认滞后 lag==1": sig["lag"] == 1,
        "确认滞后 lag>=2": sig["lag"] >= 2,
    }
    rows = []
    base_rate = sig["is_endpoint"].mean()
    for name, m in feats.items():
        sub = sig[m]
        if len(sub) < 30:
            continue
        rate = sub["is_endpoint"].mean()
        rows.append({
            "特征": name, "命中数": len(sub),
            "端点率": f"{rate * 100:.1f}%",
            "基准端点率": f"{base_rate * 100:.1f}%",
            "提升": f"{(rate / base_rate - 1) * 100:+.0f}%",
        })
    print(pd.DataFrame(rows).to_string(index=False))

    print("\n########## 分型间隔的判别力（距上一分型的K线数）##########")
    sig = sig.sort_values("fx_idx").reset_index(drop=True)
    sig["gap"] = sig["fx_idx"].diff().fillna(0).astype(int)
    rows = []
    for lo, hi in [(0, 2), (3, 4), (5, 8), (9, 15), (16, 10**6)]:
        sub = sig[(sig.gap >= lo) & (sig.gap <= hi)]
        if len(sub) < 30:
            continue
        label = f"间隔{lo}-{hi}" if hi < 10**5 else f"间隔>{lo - 1}"
        rows.append({
            "分组": label, "n": len(sub),
            "端点率": f"{sub['is_endpoint'].mean() * 100:.1f}%",
        })
    print(pd.DataFrame(rows).to_string(index=False))
    print("\n  提示：缠论要求笔的两端分型之间至少间隔一个KLC，间隔本身就是实时可得的强过滤。")

    rows = []
    for lo, hi in [(5, 8), (9, 15), (16, 10**6)]:
        sub = sig[(sig.gap >= lo) & (sig.gap <= hi)]
        label = f"间隔{lo}-{hi}" if hi < 10**5 else f"间隔>{lo - 1}"
        rows += summarize(sub, cdf, label)
    print("\n########## 按间隔分组的收益（全部分型，非仅端点）##########")
    show(rows)


if __name__ == "__main__":
    main()