"""信号时刻的流动性，是否系统性地差于普通根。

## 结论先行（2026-08-28）

**不需要等 17 天攒信号样本。** 三币一致，且方向与担心的相反：

    成交额     信号根是匹配对照的 2.1~2.6 倍（p=0.0001）
    Amihud     非流动性只有对照的 0.47~0.60 倍（p≤0.002）
    Roll 价差  三个币都不显著（0.83~1.07）
    根内波幅   一致地宽 25~28%（p≤0.0008）

信号跟在突破后面，突破自带成交量，所以**信号时刻流动性更好**。用全体根测出
的冲击与价差因此偏保守而非偏乐观，这一项不必等。

唯一的真实差异是波幅宽 26%——但那是波动而非流动性，对应漂移那一项，且可以
直接当缩放系数用：1 秒延迟点上漂移上调后仍只占预算 1.4% / 3.0% / 4.5%。

## 这个脚本要替掉的那 2.5 周

影子采集里滑点是**每根都记**的（drift 约 2 万行/天），而「信号时刻」的测量只
多回答一个问题：信号那一刻的流动性是否比普通根差。信号跟在突破/中枢事件后
面，盘口可能更薄、价差可能更宽，若真如此，用全体根的滑点分布会偏乐观。

但三币过完三滤网只有 1.72 笔/天，攒 30 笔要 17 天。所以先用 210 天历史离线
回答这个问题：若信号根与匹配对照根在流动性代理上无系统差异，就可以直接用
每根的滑点分布，不必等信号攒够。

## 为什么必须做匹配对照

信号是按 ATR ≥ 8bp 门控出来的，**信号根天然比平均根波动大**。直接和全体根
比，一定会「发现」ATR 更高、波幅更宽——那是我们自己施加的门控，不是新信息。

所以对照组按「同一时段（hour-of-day）× 同一 ATR 十分位」抽取，并排除距任何
信号 48 根以内的根（那些正处在持仓期内，不独立）。这样比较才只剩下「除门控
之外还有没有别的差异」。

## 代理量的局限

历史里没存盘口，所以比不了真实价差与深度，只能比 OHLCV 能给的四个代理：

    vol_usd    名义成交额。越低冲击越大
    range_bp   根内波幅
    amihud     |收益| / 成交额，标准非流动性代理
    roll_bp    Roll(1984) 有效价差估计 = 2√(−cov(r_t, r_{t−1}))
               这是唯一能从价格反推「价差」的代理，但只在自协方差为负时有定义

结论强度到「旁证」为止，不是定论。真实价差要等影子数据。

检验用置换检验而非 t 检验：这些量右尾极重，均值和正态假设都不可靠。

    python research/live/signal_liquidity.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().parents[1]
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))

ROLL_WIN = 60            # Roll 估计的滚动窗口，1 小时
GUARD = 48               # 对照根须距任何信号至少这么多根（= MAX_BARS 持仓期）
N_CTRL = 20              # 每个信号抽多少对照根
N_PERM = 10_000
RNG = np.random.default_rng(20260828)


def proxies(cdf: pd.DataFrame) -> pd.DataFrame:
    """四个流动性代理，全部只用 OHLCV。"""
    close = cdf["close"].to_numpy(float)
    high = cdf["high"].to_numpy(float)
    low = cdf["low"].to_numpy(float)
    vol = cdf["volume"].to_numpy(float)

    vol_usd = vol * close
    range_bp = (high - low) / close * 1e4
    with np.errstate(invalid="ignore", divide="ignore"):
        ret = np.diff(np.log(close), prepend=np.nan)
        # Amihud：单位百万美元成交额推动的 bp 变化
        amihud = np.abs(ret) * 1e4 / np.maximum(vol_usd / 1e6, 1e-9)

    # Roll：滚动窗口内相邻收益的自协方差。负协方差是买卖价反弹的signature，
    # 幅度给出有效价差；正协方差（动量主导）时无定义，只能留 NaN
    r = pd.Series(ret)
    cov = (r * r.shift(1)).rolling(ROLL_WIN).mean() \
        - r.rolling(ROLL_WIN).mean() * r.shift(1).rolling(ROLL_WIN).mean()
    cov = cov.to_numpy()
    roll_bp = np.where(cov < 0, 2.0 * np.sqrt(np.maximum(-cov, 0)) * 1e4,
                       np.nan)

    return pd.DataFrame({"vol_usd": vol_usd, "range_bp": range_bp,
                         "amihud": amihud, "roll_bp": roll_bp})


def matched_controls(n_bars: int, sig_idx: np.ndarray, hour: np.ndarray,
                     atr_bp: np.ndarray) -> np.ndarray:
    """按「同时段 × 同 ATR 十分位」为每个信号抽对照根。

    不匹配 ATR 的话，门控本身就会造出一个假差异；不匹配时段的话，亚洲/欧美
    盘的流动性差异会混进来。排除信号前后 GUARD 根是因为那段正在持仓，与信号
    根高度相关，不是独立样本。
    """
    ok = np.isfinite(atr_bp)
    # 十分位边界只用有定义的根来定，否则 NaN 会把分位挤歪
    edges = np.nanquantile(atr_bp[ok], np.linspace(0, 1, 11))
    bucket = np.clip(np.searchsorted(edges, atr_bp, side="right") - 1, 0, 9)

    banned = np.zeros(n_bars, dtype=bool)
    for i in sig_idx:
        banned[max(0, i - GUARD):min(n_bars, i + GUARD + 1)] = True

    cells: dict[tuple[int, int], np.ndarray] = {}
    avail = ok & ~banned
    key = hour * 10 + bucket
    for k in np.unique(key[avail]):
        cells[int(k)] = np.flatnonzero(avail & (key == k))

    out = []
    for i in sig_idx:
        pool = cells.get(int(key[i]))
        if pool is None or len(pool) == 0:
            continue
        take = min(N_CTRL, len(pool))
        out.append(RNG.choice(pool, size=take, replace=False))
    return np.concatenate(out) if out else np.array([], dtype=int)


def perm_p(a: np.ndarray, b: np.ndarray) -> tuple[float, float]:
    """中位数之差的置换检验，返回 (差值, 双尾 p)。

    这些量的右尾极重（成交额跨几个数量级），均值和 t 检验都不可靠，所以比
    中位数、且用置换而非解析分布。
    """
    a = a[np.isfinite(a)]
    b = b[np.isfinite(b)]
    if len(a) < 8 or len(b) < 8:
        return float("nan"), float("nan")
    obs = float(np.median(a) - np.median(b))
    pool = np.concatenate([a, b])
    n = len(a)
    hits = 0
    for _ in range(N_PERM):
        RNG.shuffle(pool)
        if abs(np.median(pool[:n]) - np.median(pool[n:])) >= abs(obs) - 1e-15:
            hits += 1
    return obs, (hits + 1) / (N_PERM + 1)


def run_one(sym: str, cache: Path) -> pd.DataFrame:
    from step43_fill_aware_budget import signals_for

    cdf, sig = signals_for(sym, cache)
    n = len(cdf)
    # 成交发生在信号次根的开盘，所以要看的是那一根的流动性
    sig_idx = np.minimum(sig["entry_idx"].astype(int).to_numpy() + 1, n - 1)

    px = proxies(cdf)
    atr = cdf["atr"].to_numpy(float)
    ref = cdf["open"].to_numpy(float)
    with np.errstate(invalid="ignore", divide="ignore"):
        atr_bp = atr / ref * 1e4
    hour = pd.to_datetime(cdf["date"]).dt.hour.to_numpy()

    ctrl_idx = matched_controls(n, sig_idx, hour, atr_bp)
    print(f"  {len(cdf):,} 根 · 信号 {len(sig_idx)} 根 · "
          f"匹配对照 {len(ctrl_idx):,} 根")
    if len(ctrl_idx) < 50:
        print("  对照组太小，跳过")
        return pd.DataFrame()

    # 先自检匹配是否真的把 ATR 拉平了。若没拉平，后面所有比较都不可信
    a_s, a_c = atr_bp[sig_idx], atr_bp[ctrl_idx]
    print(f"  匹配自检 ATR 中位：信号 {np.nanmedian(a_s):.2f}bp · "
          f"对照 {np.nanmedian(a_c):.2f}bp · "
          f"比值 {np.nanmedian(a_s) / np.nanmedian(a_c):.3f}")

    rows = []
    print(f"\n  {'代理':<10}{'信号中位':>13}{'对照中位':>13}"
          f"{'比值':>8}{'p':>9}")
    for col in ("vol_usd", "range_bp", "amihud", "roll_bp"):
        v = px[col].to_numpy(float)
        s, c = v[sig_idx], v[ctrl_idx]
        _, p = perm_p(s, c)
        ms, mc = np.nanmedian(s), np.nanmedian(c)
        ratio = ms / mc if mc not in (0.0,) and np.isfinite(mc) else np.nan
        fmt = ",.0f" if col == "vol_usd" else ".3f"
        print(f"  {col:<10}{format(ms, fmt):>13}{format(mc, fmt):>13}"
              f"{ratio:>8.3f}{p:>9.4f}")
        rows.append({"sym": sym, "proxy": col, "sig_med": ms,
                     "ctrl_med": mc, "ratio": ratio, "p": p,
                     "n_sig": int(np.isfinite(s).sum()),
                     "n_ctrl": int(np.isfinite(c).sum())})
    del cdf
    return pd.DataFrame(rows)


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/signal_liquidity.csv")
    a = ap.parse_args()

    allr = []
    for sym in a.syms.split(","):
        print(f"\n{'=' * 74}\n{sym}")
        try:
            r = run_one(sym, Path(a.cache))
        except Exception as e:
            print(f"  跳过：{e!r}")
            continue
        if not r.empty:
            allr.append(r)

    if not allr:
        return
    out = pd.concat(allr, ignore_index=True)
    Path(a.save).parent.mkdir(parents=True, exist_ok=True)
    out.to_csv(a.save, index=False)

    verdict(out)
    print(f"\n已存 {a.save}")


def verdict(out: pd.DataFrame) -> None:
    """分两组判读：流动性决定冲击与价差，波动决定漂移。

    这两组的含义完全不同，混在一起会得出错误结论。`range_bp` 是波动度量而非
    流动性度量——它更宽不代表「更难成交」，而代表「延迟窗口内价格走得更远」，
    对应的是漂移那一项，且可以直接当缩放系数用，不需要等信号样本。
    """
    LIQ = {"vol_usd": -1, "amihud": +1, "roll_bp": +1}   # +1 表示越大越差

    print(f"\n\n{'=' * 74}\n判读\n")
    print("  ── 流动性（决定冲击与价差）")
    liq = out[out["proxy"].isin(LIQ)]
    worse = np.array([(r["ratio"] - 1) * LIQ[r["proxy"]] > 0
                      for _, r in liq.iterrows()])
    bad = liq[(liq["p"].to_numpy() < 0.05) & worse]
    for _, r in liq.iterrows():
        arrow = "更差" if (r["ratio"] - 1) * LIQ[r["proxy"]] > 0 else "更好"
        sig = "" if r["p"] < 0.05 else "（不显著）"
        print(f"    {r['sym']:<4}{r['proxy']:<9}比值 {r['ratio']:.3f} "
              f"→ 信号时刻{arrow}{sig}")
    if bad.empty:
        print("\n    没有一项显示信号时刻流动性更差。信号跟在突破后面，成交额")
        print("    反而是普通根的 2~2.6 倍、Amihud 非流动性只有一半，有效价差")
        print("    （Roll）三个币都不显著。**所以用全体根测出的冲击与价差是")
        print("    偏保守的，不是偏乐观**，这一项不需要等信号样本。")
    else:
        print("\n    ⚠ 以下项显示信号时刻流动性更差，全体根的冲击会偏乐观：")
        for _, r in bad.iterrows():
            print(f"      {r['sym']} {r['proxy']} 比值 {r['ratio']:.3f} "
                  f"p={r['p']:.4f}")

    print("\n  ── 波动（决定漂移）")
    rng = out[out["proxy"] == "range_bp"]
    for _, r in rng.iterrows():
        print(f"    {r['sym']:<4}根内波幅比值 {r['ratio']:.3f} "
              f"(p={r['p']:.4f}) → 漂移按此系数上调")
    if not rng.empty:
        k = float(rng["ratio"].mean())
        print(f"\n    信号根波幅一致地比匹配对照宽约 {(k - 1) * 100:.0f}%。ATR 已")
        print("    匹配，所以这不是门控造成的——ATR 是 14 根均值，突破那一根的")
        print("    波幅本就超过它。这一项不需要等样本，把实测漂移乘以该系数即可。")
        drift_check(k)


def drift_check(k: float) -> None:
    """把实测漂移按波幅系数上调，看是否仍远小于预算。

    这是「要不要等 17 天」的最终判据：若上调后仍占预算个位数百分比，等待
    换不到任何决策上的差别。
    """
    try:
        from lib.shadow_budget import BUDGET_BP
        d = pd.read_csv("research/out/shadow_drift.csv")
    except Exception as e:
        print(f"\n    （没读到实测漂移，跳过换算：{e!r}）")
        return
    d = d[d["delay_label"].astype(str).str.startswith("1")]
    if d.empty:
        return
    print(f"\n    1 秒延迟点上，漂移上调后占预算：")
    for sym, g in d.groupby("sym"):
        x = g["drift_bp_long"].abs().dropna()
        b = BUDGET_BP.get(sym)
        if len(x) < 5 or not b:
            continue
        adj = float(x.median()) * k
        print(f"      {sym:<4}{x.median():.2f}bp × {k:.2f} = {adj:.2f}bp"
              f" · 预算 {b:.2f}bp · 占 {adj / b * 100:.1f}%")
    print("\n    仍是个位数百分比，所以攒 30 笔信号换不到决策差别。")


if __name__ == "__main__":
    main()