jackandCursor
181bca303f
吃单查询换成 Hummingbot 的 OrderBook.get_vwap_for_volume：手写的 walk_book 返回的是按计价币吃单的加权均价，但框架的 get_price_for_quote_volume 返回 边际价、get_vwap_for_volume 收基础币量，两者语义不同。改为按基础币下单 （真实委托与 PositionExecutor.amount 均是基础币计价），深度不足由 query_volume/result_volume 判定，框架此时返回 nan 而非一个看似正常的 部分成交均价。 落盘完整盘口（双边 50 档）。此前只记三个固定名义额的成交价，这批数据的 寿命就等于那几个档位的寿命；存完整深度后任意资金量级的冲击都能离线重算。 仓位档同时从 1k/5k/20k 提到十万量级，此前低估真实仓位约两个数量级。 订阅成交流，按根按价位聚合。买卖分开存——多头在目标位挂卖出靠主动买盘 成交，混在一起会把成交率高估约一倍。BTC 每根总成交额中位与 210 天历史 的 volume×close 差 0.3%，可确认采集完整。 新增两项测算： - 冲击不是绑定约束。32 万仓位单边冲击 0.19~2.39bp，对 8.58~20.64bp 的 预算只占 1.6~14.2%，冲击反推的资金上限 100~500 万。 - maker 成交率才是。止盈位被首次触及时，限价在该根价格区间中的位置 中位 k=0.28（63.9 万次触及，三币一致）；合并每根成交额后，32 万仓位 的全额成交率仅 30.1%/15.6%/1.5%。要 80% 全额成交，仓位须 ≤ 4.7 万 /1.4 万/0.26 万——比冲击反推的上限低 40~370 倍。 回测把这些止盈按「全额成交在目标价」计，故预算所依据的收益流本身需重估。 Co-authored-by: Cursor <cursoragent@cursor.com>
284 lines
12 KiB
Python
284 lines
12 KiB
Python
"""影子交易器的报表:延迟门槛 + 滑点对延迟曲线。
|
||
|
||
### 判据常数一律从研究侧 import，不在这里写死
|
||
|
||
`BUDGET_BP` 等常数留在 `research/lib/shadow_budget.py`。理由是这些数会变——
|
||
2026-08-27 一天之内预算就动了四次（3.91 → 11.06 → 14.25 → 15.19bp），
|
||
费率也改了一次。本文件曾经写死过 BTC −0.13 / ETH 4.02 / SOL 2.92，那三个数
|
||
由六处差异叠加而来（只有同向没有阶梯、费率按 6bp 双边 taker、TP=3.0、
|
||
余量没除 taker 名义额、无 ATR 门控，且 BTC/ETH/SOL 恰是 ATR 最低的三个币）。
|
||
正确值是 8.58 / 20.64 / 16.83——**ETH 差了五倍**。
|
||
|
||
这件事要紧是因为下面的判读是自动停机开关：用 4.02 当 ETH 的预算，真实滑点
|
||
只要到 2.4bp 就会报「需要压延迟或放弃」，会误杀一个可行的策略。
|
||
|
||
### 什么时候能判什么
|
||
|
||
| | 一天的样本量 | 够不够 |
|
||
|---|---|---|
|
||
| 延迟 | 1440 根/币 | 够，统计上很厚 |
|
||
| 滑点 | 门控后 4~6 笔/天 | **不够**，判据要 30 笔以上，即一周起步 |
|
||
|
||
所以首日只能判延迟和管道通不通。滑点那一节在样本不足时会明说。
|
||
|
||
### 延迟门槛（提前止损用）
|
||
|
||
若**总延迟已令预期漂移超过预算**，说明方案在这台机器上就不成立，不必等
|
||
两周样本再停。漂移按随机游走折算 σ_1m · √(t/60)。这是下限——入场条件是
|
||
「收盘突破转强」，那一刻价格正朝我们方向跑，延迟造成的是系统性追价，
|
||
不会正负抵消。所以实测滑点理应比折算值更差，两者对照本身就是个校验。
|
||
|
||
### 滑点对延迟曲线
|
||
|
||
把延迟当自变量:0.5s / 1s / 2s / 5s 各一个滑点值,外加「actual」= 本机实际
|
||
算完的时刻。同信号内的受控对比，能直接读出「若延迟压到 X 秒，滑点是多少」,
|
||
决策不被当前实现拖累。
|
||
|
||
.venv/bin/python research/live/shadow_report.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
OUT = HERE.parent / "out"
|
||
sys.path.insert(0, str(HERE.parent))
|
||
|
||
from lib.shadow_budget import ( # noqa: E402
|
||
ATR_GATE_BP, BUDGET_PORTFOLIO_2026, LAG_ALARM_MS, budget_of, lag_healthy,
|
||
verdict,
|
||
)
|
||
|
||
SYMS = ("BTC", "ETH", "SOL")
|
||
ORDER = ["0.5s", "1.0s", "2.0s", "5.0s", "actual"]
|
||
MIN_N = 30 # 滑点判据的最低笔数，低于此只报数不下结论
|
||
|
||
|
||
def vol_bp() -> dict[str, float]:
|
||
v = {}
|
||
for s in SYMS:
|
||
f = HERE / "cache" / f"bitget_{s}_1m_30d.feather"
|
||
if not f.exists():
|
||
f = HERE / "cache" / f"bitget_{s}_1m_210d.feather"
|
||
if not f.exists():
|
||
continue
|
||
c = np.log(pd.read_feather(f)["close"].to_numpy(float))
|
||
v[s] = float(np.nanstd(np.diff(c)) * 1e4)
|
||
return v
|
||
|
||
|
||
def lag_health(lat: pd.DataFrame) -> None:
|
||
"""运行时 lag 探针的回看。补丁后实测 506~642ms，理论下限约 500ms。"""
|
||
print("\n\n########## 二、lag 探针（>%.0fms 该停开仓）##########"
|
||
% LAG_ALARM_MS)
|
||
print(f"{'币':<5}{'根数':>6}{'中位ms':>9}{'P90ms':>8}{'最差30根中位':>14}"
|
||
f"{'超阈根数':>10}{'判定':>8}")
|
||
for s in SYMS:
|
||
g = lat[lat["sym"] == s].sort_values("kline_ts")
|
||
if g.empty:
|
||
continue
|
||
x = g["lag_data_ms"].to_numpy(float)
|
||
roll = pd.Series(x).rolling(30).median()
|
||
worst = float(np.nanmax(roll)) if roll.notna().any() else float("nan")
|
||
ok = lag_healthy(x)
|
||
print(f"{s:<5}{len(g):>6}{np.median(x):>9.0f}"
|
||
f"{np.percentile(x, 90):>8.0f}{worst:>14.0f}"
|
||
f"{int((x > LAG_ALARM_MS).sum()):>10}"
|
||
f"{'健康' if ok else '退化':>8}")
|
||
if "lag_ok" in lat.columns:
|
||
bad = int((lat["lag_ok"] == 0).sum())
|
||
if bad:
|
||
print(f"\n 采集期间有 {bad} 根被判不健康，那些根上的信号"
|
||
f"（lag_ok=0）在真实运行下不会开仓，统计时应排除")
|
||
|
||
|
||
def latency_gate(lat: pd.DataFrame, vols: dict) -> None:
|
||
print("########## 一、延迟门槛 ##########")
|
||
span_h = (lat["t_close_ms"].max() - lat["t_close_ms"].min()) / 3.6e6
|
||
print(f"样本跨度 {span_h:.1f} 小时 · 共 {len(lat)} 根\n")
|
||
print(f"{'币':<5}{'根数':>6}{'数据ms':>9}{'计算ms':>9}{'总延迟ms':>10}"
|
||
f"{'P90ms':>8}{'折算漂移bp':>12}{'预算bp':>9}{'占预算':>9}")
|
||
rows = {}
|
||
for s in SYMS:
|
||
g = lat[lat["sym"] == s]
|
||
if g.empty:
|
||
continue
|
||
d = float(np.median(g["lag_data_ms"]))
|
||
c = float(np.median(g["compute_ms"]))
|
||
t = float(np.median(g["lag_signal_ms"]))
|
||
p90 = float(np.percentile(g["lag_signal_ms"], 90))
|
||
vol = vols.get(s)
|
||
drift = vol * np.sqrt(t / 60_000) if vol else float("nan")
|
||
b = budget_of(s)
|
||
share = drift / b if np.isfinite(b) and b > 0 else float("nan")
|
||
rows[s] = (drift, b)
|
||
txt = f"{share * 100:.0f}%" if np.isfinite(share) else "—"
|
||
print(f"{s:<5}{len(g):>6}{d:>9.0f}{c:>9.0f}{t:>10.0f}{p90:>8.0f}"
|
||
f"{drift:>12.2f}{b:>9.2f}{txt:>9}")
|
||
|
||
print("\n判读（预算来自 lib/shadow_budget，2026 年口径）：")
|
||
for s, (drift, b) in rows.items():
|
||
if not np.isfinite(b):
|
||
print(f" {s}: 当前环境预算不足，不作交易标的，仅作延迟参照")
|
||
elif not np.isfinite(drift):
|
||
# 不特判的话 nan > b 是 False，会一路落到「尚有空间」说反话
|
||
print(f" {s}: 缺 1m 波动率缓存，折算不出漂移，无法判读")
|
||
elif drift > b:
|
||
print(f" {s}: 折算漂移 {drift:.2f}bp 已超预算 {b:.2f}bp —— 停下改方案")
|
||
elif drift > b * 0.6:
|
||
print(f" {s}: 折算漂移 {drift:.2f}bp 吃掉预算 {b:.2f}bp 的六成以上，"
|
||
f"需要压延迟或放弃")
|
||
else:
|
||
print(f" {s}: 折算漂移 {drift:.2f}bp 对预算 {b:.2f}bp 尚有空间，继续收集")
|
||
|
||
|
||
def book_quality(sig: pd.DataFrame, drf: pd.DataFrame) -> None:
|
||
"""回查到的盘口比目标时刻晚多少。晚太多的已在采集侧丢弃，这里做复核。"""
|
||
frames = [d for d in (sig, drf) if not d.empty and "book_lag_ms" in d]
|
||
if not frames:
|
||
return
|
||
x = pd.concat([d["book_lag_ms"] for d in frames]).astype(float)
|
||
print("\n\n########## 二·五、盘口回查质量 ##########")
|
||
print(f" 回查 {len(x)} 次 · 中位 {x.median():.0f}ms · "
|
||
f"P90 {np.percentile(x, 90):.0f}ms · 最大 {x.max():.0f}ms")
|
||
print(f" 10Hz 采样下这个值应在 0~100ms。它是所有延迟点的同向偏置，"
|
||
f"不改变曲线形状，但要确认没有异常长尾")
|
||
|
||
|
||
def drift_split(sig: pd.DataFrame, drf: pd.DataFrame) -> None:
|
||
"""条件漂移 vs 无条件漂移。两者的差就是「系统性追价」的大小。"""
|
||
print("\n\n########## 三、条件漂移 vs 无条件漂移 ##########")
|
||
if drf.empty:
|
||
print(" 尚无逐根漂移数据（shadow_drift.csv 由本轮起才开始记）")
|
||
return
|
||
print(" 无条件 = 每根 K 线，方向未知故取 |漂移|；"
|
||
"条件 = 信号根按下单方向定号")
|
||
print(f"\n{'延迟':<8}{'无条件n':>9}{'无条件|漂移|':>14}"
|
||
f"{'条件n':>7}{'条件漂移':>10}{'追价差':>9}")
|
||
cond = sig[(sig["notional"] == sig["notional"].min())] if not sig.empty \
|
||
else sig
|
||
for lb in ORDER:
|
||
u = drf[drf["delay_label"] == lb]["drift_bp_long"].abs()
|
||
c = cond[cond["delay_label"] == lb]["drift_bp"] if not cond.empty \
|
||
else pd.Series(dtype=float)
|
||
if u.empty and c.empty:
|
||
continue
|
||
um = u.mean() if not u.empty else float("nan")
|
||
cm = c.mean() if not c.empty else float("nan")
|
||
print(f"{lb:<8}{len(u):>9}{um:>14.2f}{len(c):>7}{cm:>10.2f}"
|
||
f"{cm - um:>9.2f}")
|
||
if len(cond) and len(cond[cond["delay_label"] == "1.0s"]) < MIN_N:
|
||
print(f"\n 条件侧样本不足 {MIN_N}，差值还读不出方向")
|
||
|
||
|
||
def slippage_curve(sig: pd.DataFrame) -> None:
|
||
print("\n\n########## 四、滑点对延迟曲线 ##########")
|
||
if sig.empty:
|
||
print(" 尚无信号样本")
|
||
return
|
||
|
||
def n_of(df):
|
||
return df.groupby(["sym", "kline_ts", "direction"]).ngroups
|
||
|
||
has_flags = "pass_all" in sig.columns
|
||
if not has_flags:
|
||
print(" ⚠ 数据来自旧版采集（只有 h1_agree，无阶梯与 ATR 门控）。"
|
||
"这批不是我们要交易的那批信号，只能作管道验证，不能对预算判读。\n")
|
||
main_scope, main_name = sig[sig["h1_agree"] == 1], "仅 h1_agree=1（旧口径）"
|
||
else:
|
||
# depth_ok=0 是 25 档吃不满该仓位，均价按部分成交算会**低估**冲击
|
||
ok = (sig["pass_all"] == 1) & (sig["lag_ok"] == 1)
|
||
if "depth_ok" in sig.columns:
|
||
thin = int((sig["depth_ok"] == 0).sum())
|
||
ok &= sig["depth_ok"] == 1
|
||
if thin:
|
||
print(f" （{thin} 行深度吃不满，已排除；这些行会低估冲击）")
|
||
print(f"信号总数 {n_of(sig)} · 同向 {n_of(sig[sig['h1_agree'] == 1])}"
|
||
f" · 同向+阶梯 "
|
||
f"{n_of(sig[(sig['h1_agree'] == 1) & (sig['ladder_ok'] == 1)])}"
|
||
f" · 三项全过 {n_of(sig[sig['pass_all'] == 1])}"
|
||
f" · 再要求 lag 健康 {n_of(sig[ok])}")
|
||
print(f"（门控阈值 ATR ≥ {ATR_GATE_BP:.0f}bp，是费率的函数不是市场常数）\n")
|
||
main_scope = sig[ok]
|
||
main_name = "三项滤网全过 + lag 健康 + 深度吃满（主口径）"
|
||
|
||
scopes = [("全部信号（含不会下单的，仅作提前读数）", sig),
|
||
(main_name, main_scope)]
|
||
for scope, sub in scopes:
|
||
if sub.empty:
|
||
print(f"--- {scope} ---\n 尚无样本\n")
|
||
continue
|
||
print(f"--- {scope} ---")
|
||
print(f"{'延迟':<8}{'仓位':>9}{'n':>5}{'滑点均值bp':>12}"
|
||
f"{'中位bp':>9}{'漂移bp':>9}{'价差bp':>9}{'冲击bp':>9}")
|
||
for lb in ORDER:
|
||
g0 = sub[sub["delay_label"] == lb]
|
||
for nt in sorted(sub["notional"].unique()):
|
||
g = g0[g0["notional"] == nt]
|
||
if g.empty:
|
||
continue
|
||
print(f"{lb:<8}{int(nt):>9}{len(g):>5}"
|
||
f"{g['slip_bp'].mean():>12.2f}"
|
||
f"{g['slip_bp'].median():>9.2f}"
|
||
f"{g['drift_bp'].mean():>9.2f}"
|
||
f"{g['spread_bp'].mean():>9.2f}"
|
||
f"{g['impact_bp'].mean():>9.2f}")
|
||
print()
|
||
|
||
print("--- 分币种判读（主口径，仓位 5000）---")
|
||
m = main_scope[main_scope["notional"] == 5000.0] if not main_scope.empty \
|
||
else main_scope
|
||
if m.empty:
|
||
print(" 尚无样本")
|
||
return
|
||
print(f"{'币':<5}{'延迟':<8}{'n':>5}{'滑点中位bp':>12}{'预算bp':>9} 判读")
|
||
for s in SYMS:
|
||
for lb in ORDER:
|
||
g = m[(m["sym"] == s) & (m["delay_label"] == lb)]
|
||
if g.empty:
|
||
continue
|
||
med = float(g["slip_bp"].median())
|
||
b = budget_of(s)
|
||
note = verdict(med, s) if len(g) >= MIN_N \
|
||
else f"n={len(g)} < {MIN_N}，不下结论"
|
||
print(f"{s:<5}{lb:<8}{len(g):>5}{med:>12.2f}{b:>9.2f} {note}")
|
||
|
||
n_main = len(m[m["delay_label"] == "actual"])
|
||
if n_main < MIN_N:
|
||
print(f"\n ⚠ 主口径仅 {n_main} 笔。门控后约 4~6 笔/天/全部币种，"
|
||
f"滑点判据要 {MIN_N} 笔以上——**一周起步**。首日只能判延迟和管道。")
|
||
print(f" 单币样本薄时可先看组合口径：2026 预算 {BUDGET_PORTFOLIO_2026}bp")
|
||
|
||
|
||
def _load(name: str) -> pd.DataFrame:
|
||
f = OUT / name
|
||
if not f.exists() or f.stat().st_size == 0:
|
||
return pd.DataFrame()
|
||
return pd.read_csv(f)
|
||
|
||
|
||
def main() -> None:
|
||
vols = vol_bp()
|
||
print("1m 收益标准差（bp/分钟，Bitget 实测）："
|
||
+ " ".join(f"{s} {v:.2f}" for s, v in vols.items()) + "\n")
|
||
|
||
lat = _load("shadow_latency.csv")
|
||
if lat.empty:
|
||
print("尚无延迟数据")
|
||
else:
|
||
latency_gate(lat, vols)
|
||
lag_health(lat)
|
||
|
||
sig, drf = _load("shadow_signals.csv"), _load("shadow_drift.csv")
|
||
book_quality(sig, drf)
|
||
drift_split(sig, drf)
|
||
slippage_curve(sig)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|