Chan/research/step43_fill_aware_budget.py

"""Step 43：把「限价单全额成交」的假设换成按成交量结算，重算滑点预算。

step42 的预算（BTC 8.58 / ETH 20.64 / SOL 16.83bp）建立在一个假设上：挂在
3ATR 与 8ATR 的止盈限价单全额成交在目标价。影子交易的成交流数据推翻了它——
真实仓位下全额成交率只有 30%/16%/1.5%（32 万仓位）。

预算因此不再是一个常数，而是**仓位规模的函数**。规模越大，止盈越难成交，
越多仓位被拖到止损或超时（taker，且吃滑点），预算越低。这条曲线与「冲击
反推的容量」是两个不同的约束，而后者宽松得多（100~500 万 vs 数万）。

数据用 Bitget 210 天 1m，与影子测量同源同交易所。全量 366 万根峰值 24.5GB，
本机 15GB 跑不动；210 天 30 万根峰值约 2GB。

    python research/step43_fill_aware_budget.py --syms BTC,ETH,SOL
"""
from __future__ import annotations

import argparse
import os
import sys
import warnings
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", 400)

LTF, HTF = "1m", "5m"
SL, SCALE_AT, RUNNER, RUNNER_STOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48
NOTIONALS = [5e3, 1e4, 2e4, 5e4, 1e5, 2e5, 3.2e5, 5.3e5, 1e6]


def signals_for(sym: str, cache: Path) -> tuple[pd.DataFrame, pd.DataFrame]:
    """跑缠论链路，返回 (cdf, 过完三滤网的信号)。口径抄 step42.run_one。"""
    from chanlun import TF_DF
    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
    from lib.shadow_budget import ATR_GATE_BP

    def load(tf: str) -> pd.DataFrame:
        c = sorted(cache.glob(f"bitget_{sym}_{tf}_*.feather"),
                   key=lambda p: p.stat().st_size, reverse=True)
        if not c:
            raise FileNotFoundError(f"没有 {sym} {tf} 缓存")
        return pd.read_feather(c[0])

    chan_l = TF_DF(load(LTF), 1, LTF)
    cdf = chan_l.dataframe
    zones = build_htf_zones(cdf, LTF, chan=chan_l).reset_index(drop=True)
    if zones.empty:
        raise RuntimeError("无中枢")
    z = zones.copy()
    pg, pdn = z["zg"].shift(), z["zd"].shift()
    z["z_above"], z["z_below"] = z["zd"] > pg, z["zg"] < pdn
    z["zone_i"] = np.arange(len(z))

    chan_h = TF_DF(load(HTF), 1, HTF)
    tl = htf_fx_timeline(
        signals_to_frame(extract_fx_signals(chan_h, chan_h.dataframe)),
        chan_h.dataframe)
    del chan_h

    sig = find_fast_bsp3(cdf, zones)
    if sig.empty:
        raise RuntimeError("无信号")
    sig = sig.merge(z[["zone_i", "z_above", "z_below"]], on="zone_i",
                    how="left")
    sig = attach_htf_context(sig, cdf, tl, "h1")
    push = np.where(sig["direction"] == 1, sig["z_above"], sig["z_below"])
    keep = ((sig["h1_agree"] == 1)
            & pd.Series(push, index=sig.index).fillna(False).astype(bool))
    sig = sig[keep].copy()

    # ATR 门控：与 shadow_signal 同源，分母取次根开盘价
    idx = sig["entry_idx"].astype(int).to_numpy()
    atr = cdf["atr"].to_numpy(float)
    op = cdf["open"].to_numpy(float)
    ref = op[np.minimum(idx + 1, len(op) - 1)]
    with np.errstate(invalid="ignore", divide="ignore"):
        atr_bp = atr[idx] / ref * 1e4
    sig = sig[np.isfinite(atr_bp) & (atr_bp >= ATR_GATE_BP)]
    return cdf, sig


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--syms", default="BTC,ETH,SOL")
    ap.add_argument("--cache", default="research/live/cache")
    ap.add_argument("--save", default="research/out/step43_fill_budget.csv")
    a = ap.parse_args()

    from lib.exit_fill import (assert_converges, budget_bp, net_bp,
                               walk_filled)
    from lib.exit_model import cfg_name, slip_budget
    from lib.exit_model import walk_exits

    cache = Path(a.cache)
    rows = []
    for sym in a.syms.split(","):
        print(f"\n{'=' * 74}\n{sym}")
        try:
            cdf, sig = signals_for(sym, cache)
        except Exception as e:
            print(f"  跳过：{e!r}")
            continue
        print(f"  {len(cdf):,} 根 1m · 过三滤网 {len(sig)} 笔信号")
        if len(sig) < 25:
            print("  样本不足 25 笔，不出统计")
            continue

        # 老口径：假定限价全额成交
        old = walk_exits(cdf, sig, [SL], [SCALE_AT], [MAXB], SCALE_AT,
                         [RUNNER], [RUNNER_STOP])
        c = cfg_name(SL, RUNNER, MAXB, RUNNER_STOP)
        b_old = slip_budget(old[f"{c}_g"].to_numpy(),
                            old[f"{c}_r"].to_numpy(),
                            old[f"{c}_c"].to_numpy())
        # 先证明两套实现在「仓位趋近 0」这个极限上逐笔一致，
        # 否则后面看到的差异分不清是成交量效应还是实现 bug
        assert_converges(cdf, sig, SL, SCALE_AT, RUNNER, RUNNER_STOP, MAXB)
        print(f"\n  老口径（限价全额成交）预算 {b_old:.2f}bp"
              f"   [极限一致性断言通过]")
        print(f"\n  {'仓位':>10} {'预算bp':>9} {'maker占比':>10} "
              f"{'止损占比':>9} {'超时占比':>9} {'净收益bp':>10}")
        for nt in NOTIONALS:
            r = walk_filled(cdf, sig, nt, SL, SCALE_AT, RUNNER,
                            RUNNER_STOP, MAXB)
            if r.empty:
                continue
            b = budget_bp(r)
            print(f"  {nt:>10,.0f} {b:>9.2f} "
                  f"{r['maker_share'].mean() * 100:>9.1f}% "
                  f"{r['w_stop'].mean() * 100:>8.1f}% "
                  f"{r['w_time'].mean() * 100:>8.1f}% "
                  f"{net_bp(r):>10.2f}")
            rows.append({"sym": sym, "notional": nt, "budget_bp": b,
                         "budget_bp_old": b_old,
                         "maker_share": r["maker_share"].mean(),
                         "w_stop": r["w_stop"].mean(),
                         "w_time": r["w_time"].mean(),
                         "net_bp": net_bp(r), "n": len(r)})
        del cdf
    if rows:
        out = pd.DataFrame(rows)
        Path(a.save).parent.mkdir(parents=True, exist_ok=True)
        out.to_csv(a.save, index=False)
        print(f"\n已存 {a.save}")


if __name__ == "__main__":
    main()