From 3b14bba247c4bb1bd4a69e4ec5875f2955f86742 Mon Sep 17 00:00:00 2001 From: jack Date: Fri, 28 Aug 2026 03:52:23 +0800 Subject: [PATCH] =?UTF-8?q?=E7=A6=BB=E7=BA=BF=E9=AA=8C=E8=AF=81=E4=BF=A1?= =?UTF-8?q?=E5=8F=B7=E6=97=B6=E5=88=BB=E6=B5=81=E5=8A=A8=E6=80=A7=E4=B8=8D?= =?UTF-8?q?=E6=9B=B4=E5=B7=AE=EF=BC=8C=E5=85=8D=E6=8E=89=E6=94=92=2030=20?= =?UTF-8?q?=E7=AC=94=E4=BF=A1=E5=8F=B7=E7=9A=84=2017=20=E5=A4=A9=E7=AD=89?= =?UTF-8?q?=E5=BE=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三币过完三滤网只有 1.72 笔/天，攒 30 笔要 17 天。但滑点本来每根都在记，信号 时刻的测量只多回答一个问题：信号那一刻的流动性是否比普通根差。这个问题可以 用 210 天历史离线回答。 关键是必须做匹配对照。信号按 ATR ≥ 8bp 门控，信号根天然比平均根波动大，直接 和全体根比一定会「发现」一个我们自己施加的差异。所以对照组按「同时段 × 同 ATR 十分位」抽取，并排除距信号 48 根内的根（持仓期不独立）。自检 ATR 比值 0.976~1.004，匹配成立。 结果三币一致，方向与担心的相反：信号根成交额是对照的 2.1~2.6 倍、Amihud 非 流动性只有 0.47~0.60 倍、Roll 有效价差三个币都不显著。信号跟在突破后面， 突破自带成交量。所以全体根测出的冲击与价差偏保守而非偏乐观。 唯一真实差异是根内波幅宽 25~28%，但那是波动而非流动性，对应漂移而非冲击， 且可直接当缩放系数：1 秒延迟点漂移上调后仍只占预算 1.4%/3.0%/4.5%。 判读按流动性与波动分两组。初版把 range_bp 当成流动性红旗，会得出「信号时刻 更差、必须等样本」的相反结论——它是波动度量，且 ATR 已匹配。 检验用置换而非 t：成交额跨几个数量级，右尾太重。scipy 未安装，也不宜在采集 运行中动环境，所以用 numpy 自己实现。 Co-authored-by: Cursor --- research/live/signal_liquidity.py | 308 ++++++++++++++++++++++++++++++ research/out/signal_liquidity.csv | 13 ++ 2 files changed, 321 insertions(+) create mode 100644 research/live/signal_liquidity.py create mode 100644 research/out/signal_liquidity.csv diff --git a/research/live/signal_liquidity.py b/research/live/signal_liquidity.py new file mode 100644 index 0000000..8fd805e --- /dev/null +++ b/research/live/signal_liquidity.py @@ -0,0 +1,308 @@ +"""信号时刻的流动性，是否系统性地差于普通根。 + +## 结论先行（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() diff --git a/research/out/signal_liquidity.csv b/research/out/signal_liquidity.csv new file mode 100644 index 0000000..0cd37ce --- /dev/null +++ b/research/out/signal_liquidity.csv @@ -0,0 +1,13 @@ +sym,proxy,sig_med,ctrl_med,ratio,p,n_sig,n_ctrl +BTC,vol_usd,5550103.60788,2624836.584895,2.114456816023851,9.999000099990002e-05,91,1820 +BTC,range_bp,14.211131842272723,11.379813132903628,1.248801863115187,0.0007999200079992001,91,1820 +BTC,amihud,0.9320529751601416,1.995826823303005,0.4670009262715666,0.0007999200079992001,91,1820 +BTC,roll_bp,4.779754159181145,5.745083186251997,0.8319730113950504,0.08029197080291971,44,996 +ETH,vol_usd,3126641.7427,1207249.22975,2.589889200714149,9.999000099990002e-05,126,2520 +ETH,range_bp,15.77081223459847,12.35743868645172,1.2762201484267979,0.00029997000299970003,126,2520 +ETH,amihud,2.410291522791945,4.420234301516299,0.5452859188855959,0.0018998100189981002,126,2520 +ETH,roll_bp,6.347567691125725,5.949780489978004,1.0668574583243478,0.42275772422757724,56,1362 +SOL,vol_usd,345692.792,162027.9883,2.1335375179746032,9.999000099990002e-05,145,2900 +SOL,range_bp,15.652030645029573,12.452668963002594,1.25692176444523,0.00019998000199980003,145,2900 +SOL,amihud,20.14951190677575,33.887234963568275,0.594604780485579,0.001999800019998,145,2900 +SOL,roll_bp,5.7402773658108215,6.019208571018624,0.9536598205699658,0.5864413558644136,73,1632