Chan/research/step43_fill_aware_budget.py

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

step42 的预算建立在一个假设上：挂在 3ATR 与 8ATR 的止盈限价单全额成交在目标
价。本脚本把它换成按成交流实测的成交量结算，让预算变成**仓位规模的函数**。

## 结论（2026-08-28）

**主口径 10 万 USDT 下成交率不是绑定约束。** 预算相对全额成交假设的降幅：
BTC 0% / ETH 0% / SOL 1.1%；即便到 100 万也只有 2.6% / 0.9% / 8.0%。

⚠ 此前一版本文档写「真实仓位下全额成交率只有 30%/16%/1.5%」，那是
shadow_depth.composite_fill 的口径——只算首次触及那一根的可成交量，而真实
挂单在那儿常驻最多 48 根、每根都在成交。该数系统性偏悲观，已撤回。

另一个约束是冲击反推的容量上限（100~500 万），比成交率宽松得多。两者都不
绑定，所以 10 万仓位上限制来自别处，不是流动性。

结论对成交分布曲线**不敏感**：把空头侧可成交量砍一半，10 万仓位下 BTC/ETH
预算完全不动、SOL 动 0.07bp。见 `--sensitivity`。这一点重要，因为那两条曲线
目前只有 45 根成交流样本。

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

    python research/step43_fill_aware_budget.py --syms BTC,ETH,SOL
    python research/step43_fill_aware_budget.py --sensitivity
"""
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 sensitivity(syms: list[str], cache: Path, notional: float = 1e5) -> None:
    """预算对成交分布曲线的敏感性。

    SHAPE_F / SHAPE_F_SHORT 只有 45 根成交流样本，数值精度很低。所以必须先
    证明结论对它们不敏感，否则整条预算曲线都建立在 45 根样本上。

    三档：两侧共用买盘曲线（旧口径，空头偏乐观）／实测的方向各异曲线／把
    空头可成交量再砍一半的悲观上界。
    """
    from lib.exit_fill import SHAPE_F, SHAPE_F_SHORT, budget_bp, walk_filled

    half = np.clip(SHAPE_F_SHORT * 0.5, 0.0, 1.0)
    half[-1] = 1.0
    cases = [("对称(旧口径)", SHAPE_F, SHAPE_F),
             ("实测不对称", SHAPE_F, SHAPE_F_SHORT),
             ("悲观:空头量减半", SHAPE_F, half)]

    print(f"\n{'=' * 74}\n成交分布曲线敏感性 · 仓位 {notional:,.0f} USDT\n")
    print(f"  {'币':<5}" + "".join(f"{n:>18}" for n, _, _ in cases))
    for sym in syms:
        try:
            cdf, sig = signals_for(sym, cache)
        except Exception as e:
            print(f"  {sym}: 跳过 {e!r}")
            continue
        row = f"  {sym:<5}"
        for _, fl, fs in cases:
            r = walk_filled(cdf, sig, notional, SL, SCALE_AT, RUNNER,
                            RUNNER_STOP, MAXB, f=fl, f_short=fs)
            row += f"{budget_bp(r):>13.2f}bp  " if not r.empty \
                else f"{'—':>18}"
        print(row)
        del cdf
    print("\n  2026-08-28 实测三档差异 ≤ 0.07bp，结论对曲线不敏感。")


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")
    ap.add_argument("--sensitivity", action="store_true",
                    help="只跑成交分布曲线的敏感性检查")
    a = ap.parse_args()

    if a.sensitivity:
        sensitivity(a.syms.split(","), Path(a.cache))
        return

    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()