jackandCursor
3b14bba247
三币过完三滤网只有 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 <cursoragent@cursor.com>
309 lines
13 KiB
Python
309 lines
13 KiB
Python
"""信号时刻的流动性，是否系统性地差于普通根。
|
||
|
||
## 结论先行（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()
|