币池可配,实测十币的排队代价;标出漂移的 tick 地板

加 --syms(start.sh 传 SYMS),默认仍是三币。TRX 不进十币池:实盘口径 208 天
只有 5 笔,ATR 门控几乎全刷掉。

十币实测(对比三币):

  queue_ms        2 → 169ms(P90 519ms,每刻最大排队中位 533ms)
  inner_ms      144 → 196ms(CPU 争抢也拖慢了纯计算)
  lag_signal_ms 621 → 958ms;同一收盘时刻最后算完的币中位 1376ms

排队从可忽略变成主项,与「10 币 × 196ms ÷ 2 核 ≈ 640ms 突发」吻合。这也是
之前「加核没用」那个结论唯一会翻转的场景:CPU 占用率只有约 3%,问题纯粹是
所有币同一秒收盘的突发,加核压的是并行度而非单币耗时。

但 800ms 哨兵不受影响——它喂的是 t_data − kline_ts(数据腿),十币下逐币
154~532ms 全在线内。加币不碰那道闸。

另外发现一个会被静默误读的东西:ADA/AVAX/DOGE/LINK/LTC 的漂移中位精确等于
半个 tick 且在四个延迟点上完全相同。那不是漂移,是中价的最小变动量——ADA 半
tick 就有 2.34bp。拿这个数去比预算会误判某币不可做。shadow_report 加了
tick_floor() 标注;方向上安全(真实漂移只会更小,这些是上界)。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jack
2026-08-28 04:50:09 +08:00
co-authored by Cursor
parent 7c13781f90
commit ef2a85dd8c
3 changed files with 44 additions and 1 deletions
+31
View File
@@ -149,6 +149,36 @@ def book_quality(sig: pd.DataFrame, drf: pd.DataFrame) -> None:
f"不改变曲线形状,但要确认没有异常长尾")
def tick_floor(drf: pd.DataFrame) -> None:
"""标出哪些币的漂移只是 tick 量化的地板,不是真实漂移。
中价的最小变动是半个 tick,所以 tick 相对价格粗的币(ADA 半 tick 就有
2.34bp、LTC 1.00bp),漂移中位会**精确等于半 tick** 且在所有延迟点上
完全相同。这很容易被读成「ADA 漂移 2.34bp」,实际是「测不到更细」。
方向上是安全的——真实漂移只会更小,所以这些数是上界。但必须标出来,
否则会拿一个测量地板去和预算做比较,然后误判某个币不可做。
"""
if drf.empty or "mid" not in drf.columns:
return
print("\n ── tick 地板检查")
hit = False
for sym, g in drf.groupby("sym"):
mid = g["mid"].median()
# 同一币在各延迟点的漂移若几乎不变,就是被量化了
by = g.groupby("delay_label")["drift_bp_long"].apply(
lambda x: x.abs().median())
if len(by) < 3 or not np.isfinite(mid) or mid <= 0:
continue
spread = by.max() - by.min()
if spread < 0.02 and by.median() > 0.2:
hit = True
print(f" {sym:<6}漂移在各延迟点恒为 {by.median():.2f}bp"
f" → 半 tick 地板,真实漂移低于此值")
if not hit:
print(" 没有币落在 tick 地板上,漂移数值可直接读")
def drift_split(sig: pd.DataFrame, drf: pd.DataFrame) -> None:
"""条件漂移 vs 无条件漂移。两者的差就是「系统性追价」的大小。"""
print("\n\n########## 三、条件漂移 vs 无条件漂移 ##########")
@@ -173,6 +203,7 @@ def drift_split(sig: pd.DataFrame, drf: pd.DataFrame) -> None:
f"{cm - um:>9.2f}")
if len(cond) and len(cond[cond["delay_label"] == "1.0s"]) < MIN_N:
print(f"\n 条件侧样本不足 {MIN_N},差值还读不出方向")
tick_floor(drf)
def slippage_curve(sig: pd.DataFrame) -> None: