jackandCursor
75fbf4167b
两件事，都是「静默出错」那一类。 一、BookLog/TapeLog 追加到同一个 .gz，进程被 SIGKILL 时当前成员停在 deflate 块中间，下一轮追加的新成员接在垃圾字节之后。顺序解压在损坏点抛 invalid block type，该点之后全部读不出来——包括后续每轮写进去的。而读侧的异常处理 把这个当成「正常的尾部截断」静默跳过，于是只读出 21 行还不报错。 原 docstring 里写的「只丢最后一个缓冲块，不会毁掉整个文件」是错的，已证伪。 写侧改成每轮运行一个文件；读侧按 gzip 成员边界扫描、坏成员单独跳过并出声 报告，同时把同前缀的多轮文件一并读入。旧损坏文件因此多恢复出 31/21 条 （tape）与 132/95 条（books）。 二、tape_shape 只统计主动买、只自区间顶部累积，这条曲线只适用于多头止盈。 exit_fill 两侧共用它，等于把空头的可成交量按多头分布高估。实测二者不对称： 主动买在顶部 20% 内已占 40%，主动卖在底部 20% 内只有 18%。分成 SHAPE_F 与 SHAPE_F_SHORT，avail_at 按方向查各自曲线。 两条曲线只有 45 根成交流样本，所以补了 --sensitivity：把空头可成交量砍一半， 10 万仓位下 BTC/ETH 预算完全不动、SOL 动 0.07bp。结论不依赖这 45 根样本。 顺带撤掉 step43 docstring 里已作废的「成交率 30%/16%/1.5%」。 Co-authored-by: Cursor <cursoragent@cursor.com>
400 lines
18 KiB
Python
400 lines
18 KiB
Python
"""从落盘的完整盘口与成交流，算资金容量、排队量与单根成交率。
|
||
|
||
## 结论先行（2026-08-28）
|
||
|
||
主口径仓位 **10 万 USDT**。该规模下两条约束都不绑定：
|
||
|
||
冲击 单边 0.01~1.87bp，占预算 0.1~11.1%，冲击反推上限 100~500 万
|
||
成交率 按逐根累积结算，预算降幅 BTC −0% / ETH −0% / SOL −1.1%
|
||
|
||
⚠ 本文件里的 `composite_fill` 曾给出「10 万仓位全额成交率仅 7~63%」这种数，
|
||
那个口径只算**首次触及那一根**的可成交量，系统性偏悲观，已不作为结论。
|
||
真实成交率见 lib/exit_fill.py（挂单常驻多根逐步成交）。
|
||
|
||
|
||
这两个数都不该等实盘暴露：
|
||
|
||
**容量**。预算 20bp 意味着存在一个资金上限，超过它策略就不工作。既然完整
|
||
深度已落盘，任意仓位的冲击都能重算——一次采集回答所有资金量级，换个规模
|
||
不必重测一周。绑定约束是**薄盘时段**而非中位盘口，所以按分位数报。
|
||
|
||
**maker 成交率**。回测假设 3ATR / 8ATR 的限价单全额成交。深度回答不了这个
|
||
问题：深度说的是「现在挂着多少」，成交率问的是「之后打过来多少」。只有
|
||
成交流能回答，而且买卖必须分开——多头在 3ATR 挂卖出，靠主动买盘成交。
|
||
|
||
读 gzip 时必须容忍末尾成员不完整：采集进程还在写，最后一个 gzip 成员没有
|
||
结尾标记，直接遍历会在文件尾抛 EOFError 而丢掉**全部**已读记录。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import zlib
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
|
||
def out_dir() -> Path:
|
||
p = Path("/out")
|
||
return p if p.is_dir() else Path(__file__).resolve().parents[1] / "out"
|
||
|
||
|
||
GZ_MAGIC = b"\x1f\x8b\x08"
|
||
|
||
|
||
def _members(blob: bytes):
|
||
"""把可能损坏的多成员 gzip 拆成「逐个成员解压」，坏成员跳过。
|
||
|
||
采集进程被 SIGKILL 时，当前 gzip 成员停在 deflate 块中间、没有结尾标记。
|
||
下一次运行以追加方式写入的新成员就接在这段垃圾字节后面。此时用
|
||
`gzip.open` 顺序读会在损坏点抛 `invalid block type`，**该点之后的所有
|
||
数据都读不出来**——包括后续每一轮运行写进去的。曾因此只读出 21 行而
|
||
误以为样本就那么少，且不报错。
|
||
|
||
所以按成员边界扫描：某个成员解压失败，只丢它，然后前进到下一个 magic
|
||
继续。返回 (解压出的字节, 跳过的成员数)。
|
||
"""
|
||
out, skipped, i, n = [], 0, blob.find(GZ_MAGIC), len(blob)
|
||
while 0 <= i < n:
|
||
d = zlib.decompressobj(16 + zlib.MAX_WBITS)
|
||
try:
|
||
chunk = d.decompress(blob[i:])
|
||
except zlib.error:
|
||
chunk = b""
|
||
if chunk:
|
||
out.append(chunk)
|
||
# 成员完整时 unused_data 指向下一成员；否则只能往前找 magic
|
||
if d.eof and d.unused_data:
|
||
nxt = n - len(d.unused_data)
|
||
else:
|
||
nxt = blob.find(GZ_MAGIC, i + 3)
|
||
if chunk == b"":
|
||
skipped += 1
|
||
i = nxt if nxt > i else -1
|
||
return b"".join(out), skipped
|
||
|
||
|
||
def read_jsonl_gz(path: Path, quiet: bool = False):
|
||
"""读 gzip JSONL，跨运行文件汇总，坏成员跳过而非静默截断。
|
||
|
||
`path` 既可以是单个文件，也当作前缀用：同目录下 `<stem>.*.jsonl.gz`
|
||
（每轮运行一个）会一并读入，这样重启不再把历史数据连坐。
|
||
"""
|
||
base = path.name.replace(".jsonl.gz", "")
|
||
files = sorted({*path.parent.glob(f"{base}.*.jsonl.gz"),
|
||
*([path] if path.exists() else [])})
|
||
if not files:
|
||
return
|
||
for f in files:
|
||
blob = f.read_bytes()
|
||
data, skipped = _members(blob)
|
||
n_ok = n_bad = 0
|
||
for line in data.split(b"\n"):
|
||
if not line:
|
||
continue
|
||
try:
|
||
rec = json.loads(line)
|
||
except (json.JSONDecodeError, UnicodeDecodeError):
|
||
n_bad += 1 # 损坏边界上的半行
|
||
continue
|
||
n_ok += 1
|
||
yield rec
|
||
if not quiet and (skipped or n_bad):
|
||
print(f" ⚠ {f.name}: 读出 {n_ok:,} 条，"
|
||
f"跳过 {skipped} 个损坏成员 / {n_bad} 个半行")
|
||
|
||
|
||
def impact_bp(levels: list, notional: float, mid: float) -> float | None:
|
||
"""吃掉 notional 计价币后的加权均价相对中间价，bp。深度不足返回 None。"""
|
||
need = notional
|
||
cost = 0.0
|
||
qty = 0.0
|
||
for px, amt in levels:
|
||
avail = px * amt
|
||
take = min(avail, need)
|
||
q = take / px
|
||
cost += q * px
|
||
qty += q
|
||
need -= take
|
||
if need <= 1e-9:
|
||
break
|
||
if need > 1e-9 or qty <= 0:
|
||
return None
|
||
return (cost / qty / mid - 1.0) * 1e4
|
||
|
||
|
||
def queue_ahead(books_path: Path, notional: float = 1e5) -> None:
|
||
"""限价单排在多少量之后。
|
||
|
||
这是唯一还没建模的成本项。exit_fill 假定我们能吃到该价位的全部对手方
|
||
成交量，实际上我们排在该价位既有挂单之后。
|
||
|
||
这里量不了 3ATR 处的排队——3ATR 约 30bp，而 50 档盘口只覆盖 1.4~4.5bp，
|
||
那个价位远在可见深度之外。但价格走到我们的限价时，我们的单就成了盘口
|
||
最优档附近的一员，所以「最优档通常趴着多少量」是个有用的上界参照。
|
||
"""
|
||
per: dict[str, dict[str, list[float]]] = {}
|
||
for r in read_jsonl_gz(books_path):
|
||
asks, bids = r["asks"], r["bids"]
|
||
if not asks or not bids:
|
||
continue
|
||
mid = (asks[0][0] + bids[0][0]) / 2.0
|
||
d = per.setdefault(r["sym"], {"top": [], "b1": [], "b5": []})
|
||
d["top"].append(asks[0][0] * asks[0][1])
|
||
for tag, bp in (("b1", 1.0), ("b5", 5.0)):
|
||
lim = mid * (1 + bp / 1e4)
|
||
d[tag].append(sum(p * a for p, a in asks if p <= lim))
|
||
if not per:
|
||
return
|
||
print(f"\n\n########## 限价单前方的排队量 ##########")
|
||
print(f" 主口径仓位 {notional:,.0f} USDT\n")
|
||
for sym, d in sorted(per.items()):
|
||
top = np.array(d["top"]); b1 = np.array(d["b1"])
|
||
b5 = np.array(d["b5"])
|
||
print(f" {sym} 最优档 {np.median(top):>10,.0f} · "
|
||
f"1bp 内累计 {np.median(b1):>10,.0f} · "
|
||
f"5bp 内累计 {np.median(b5):>10,.0f}")
|
||
print(f" 我们的 {notional:,.0f} 相当于最优档的 "
|
||
f"{notional / max(np.median(top), 1):.1f} 倍、"
|
||
f"1bp 内总量的 {notional / max(np.median(b1), 1):.2f} 倍")
|
||
print("\n 倍数远小于 1 则排队可忽略；接近或超过 1 则我们本身就是那一档的")
|
||
print(" 主要挂单，exit_fill 的成交量假设需要打折。")
|
||
print(" SOL 的最优档倍数畸高是 tick 更细所致（Bitget 的 SOL tick 比同类")
|
||
print(" 细约 10 倍，同样的量摊到 10 倍多的价位上），所以对 SOL 该看 5bp")
|
||
print(" 档而非最优档；但即便如此它仍是三个币里排队压力最大的一个")
|
||
|
||
|
||
def capacity(books_path: Path, budgets: dict[str, float],
|
||
pctl: float = 10.0) -> None:
|
||
"""报各币的深度曲线与「冲击吃掉预算多少」的资金上限。"""
|
||
grid = [1e4, 2.5e4, 5e4, 1e5, 2e5, 5e5, 1e6, 2e6]
|
||
per: dict[str, dict[float, list[float]]] = {}
|
||
n = 0
|
||
for r in read_jsonl_gz(books_path):
|
||
asks, bids = r["asks"], r["bids"]
|
||
if not asks or not bids:
|
||
continue
|
||
mid = (asks[0][0] + bids[0][0]) / 2.0
|
||
d = per.setdefault(r["sym"], {g: [] for g in grid})
|
||
for g in grid:
|
||
v = impact_bp(asks, g, mid)
|
||
d[g].append(np.nan if v is None else v)
|
||
n += 1
|
||
|
||
if not n:
|
||
print("没有盘口快照，先跑采集")
|
||
return
|
||
|
||
print(f"\n########## 资金容量 ##########")
|
||
print(f" 基于 {n:,} 份完整盘口快照（单边买入方向）\n")
|
||
for sym, d in per.items():
|
||
b = budgets.get(sym)
|
||
print(f" {sym} 预算 {b:.2f}bp" if b else f" {sym}")
|
||
print(f" {'名义额':>12} {'冲击中位':>10} {'冲击P90':>10} "
|
||
f"{'吃满深度率':>10} {'占预算':>8}")
|
||
for g in grid:
|
||
a = np.array(d[g], dtype=float)
|
||
fill = float(np.isfinite(a).mean())
|
||
if fill == 0:
|
||
print(f" {g:>12,.0f} {'—— 50 档吃不下 ——':>30}")
|
||
continue
|
||
med = float(np.nanmedian(a))
|
||
p90 = float(np.nanpercentile(a, 90))
|
||
share = f"{med / b * 100:6.1f}%" if b else " na"
|
||
print(f" {g:>12,.0f} {med:>10.2f} {p90:>10.2f} "
|
||
f"{fill * 100:>9.1f}% {share:>8}")
|
||
if b:
|
||
# 上限：冲击的 P90（薄盘时段）吃掉预算三成为止。三成是留给
|
||
# 漂移与价差的余地——它们才是主项，冲击不该独占预算
|
||
cap = None
|
||
for g in grid:
|
||
a = np.array(d[g], dtype=float)
|
||
if not np.isfinite(a).any():
|
||
break
|
||
if float(np.nanpercentile(a, 90)) > b * 0.30:
|
||
break
|
||
cap = g
|
||
if cap is None:
|
||
print(f" → 连最小档 {grid[0]:,.0f} 的薄盘冲击都超预算三成")
|
||
else:
|
||
print(f" → 资金上限约 {cap:,.0f} USDT"
|
||
f"（薄盘 P90 冲击 ≤ 预算 30%）")
|
||
print()
|
||
|
||
|
||
def maker_fill(tape_path: Path, mults=(3.0, 8.0),
|
||
notionals=(5e4, 1e5, 2e5)) -> None:
|
||
"""限价单挂在离场目标位，本根内有多少主动量打到那里。
|
||
|
||
这里只回答「量够不够」。真实成交还要看排队位置——我们的单排在该价位
|
||
已有挂单之后，所以这是**上界**：量不够则必然不能全成交，量够也未必成交。
|
||
"""
|
||
rows = list(read_jsonl_gz(tape_path))
|
||
if not rows:
|
||
print("没有成交流数据，先跑采集")
|
||
return
|
||
print(f"\n########## maker 腿成交量上界 ##########")
|
||
print(f" 基于 {len(rows):,} 根的逐价位成交聚合")
|
||
print(f" 多头在目标位挂卖出，成交靠主动**买**盘，故只计买方向\n")
|
||
|
||
# 限价单只能被**价格 ≥ 限价**的主动买成交打到。而止盈位被触及的那一根，
|
||
# 限价往往就落在该根价格区间的顶部——最高价刚好碰到目标位是最典型的
|
||
# 情形。所以按「限价距最高价多近」分层：depth=0 表示限价正好在最高价
|
||
# （只有打在最高价那一档的量算数），depth=0.25 表示限价在区间顶部 25% 处
|
||
depths = (0.0, 0.10, 0.25, 1.0)
|
||
per: dict[str, dict[float, list[float]]] = {}
|
||
for r in rows:
|
||
buys = {float(p): v for p, v in r["buys"].items()}
|
||
d = per.setdefault(r["sym"], {k: [] for k in depths})
|
||
if not buys:
|
||
for k in depths:
|
||
d[k].append(0.0)
|
||
continue
|
||
hi, lo = max(buys), min(buys)
|
||
rng = hi - lo
|
||
for k in depths:
|
||
floor_px = hi - k * rng
|
||
d[k].append(sum(p * v for p, v in buys.items() if p >= floor_px))
|
||
|
||
for sym, d in per.items():
|
||
print(f" {sym} ≥ 限价的主动买成交额（USDT），按限价所处位置分层")
|
||
print(f" {'限价位置':>16} {'中位':>12} {'P25':>12} "
|
||
+ " ".join(f"{n:>9,.0f}全仓" for n in notionals))
|
||
for k in depths:
|
||
a = np.array(d[k], dtype=float)
|
||
where = ("正好在最高价" if k == 0 else
|
||
"整根全部成交" if k == 1.0 else
|
||
f"区间顶部 {k * 100:.0f}%")
|
||
cells = " ".join(f"{float((a >= n).mean()) * 100:8.1f}%"
|
||
for n in notionals)
|
||
print(f" {where:>16} {np.median(a):>12,.0f} "
|
||
f"{np.percentile(a, 25):>12,.0f} {cells}")
|
||
print()
|
||
print(" 「正好在最高价」那一行才是止盈被刚好触及时的真实处境；")
|
||
print(" 「整根全部成交」是最宽松的上界。两行差多少，就是回测那个")
|
||
print(" 「限价单全额成交」假设虚了多少。而且这仍未计排队——我们的单")
|
||
print(" 排在该价位既有挂单之后，所以真实成交率比表里更低")
|
||
|
||
|
||
def tape_shape(tape_path: Path, kgrid: np.ndarray) -> dict[str, np.ndarray]:
|
||
"""成交流给「形状」：一根的主动买成交额里，有多少比例落在区间顶部 k 之内。
|
||
|
||
形状与规模分开是为了绕开成交流样本小的限制——规模（每根成交多少钱）由
|
||
210 天历史成交量提供，成交流只需给出形状。
|
||
|
||
⚠ 只算主动**买**、只自顶部累积，所以这条曲线只适用于**多头**止盈（挂卖
|
||
出，靠主动买打上来）。空头止盈要用主动卖自底部累积的曲线，二者实测并不
|
||
对称：主动买在区间顶部 20% 内已占 40%，主动卖在底部 20% 内只有 18%，
|
||
BTC/ETH 平均偏差 0.21。lib/exit_fill 里两条曲线是分开的（SHAPE_F 与
|
||
SHAPE_F_SHORT）；用同一条会把空头的可成交量按多头分布高估。
|
||
|
||
「形状几十根就稳定」这个说法要打折：45 根样本足以看出上面那个方向性差异，
|
||
但不足以定数值。好在预算对形状不敏感（见 step43 --sensitivity）。
|
||
"""
|
||
acc: dict[str, list[np.ndarray]] = {}
|
||
for r in read_jsonl_gz(tape_path):
|
||
buys = {float(p): v for p, v in r["buys"].items()}
|
||
if len(buys) < 2:
|
||
continue
|
||
hi, lo = max(buys), min(buys)
|
||
rng = hi - lo
|
||
if rng <= 0:
|
||
continue
|
||
tot = sum(p * v for p, v in buys.items())
|
||
if tot <= 0:
|
||
continue
|
||
frac = np.array([sum(p * v for p, v in buys.items()
|
||
if p >= hi - k * rng) / tot for k in kgrid])
|
||
acc.setdefault(r["sym"], []).append(frac)
|
||
return {s: np.mean(np.vstack(v), axis=0) for s, v in acc.items() if v}
|
||
|
||
|
||
def composite_fill(tape_path: Path, pen_path: Path, cache: Path,
|
||
notionals=(5e4, 1e5, 2e5)) -> None:
|
||
"""把穿透深度分布与成交量曲线合并，出**单根**内的 maker 成交率。
|
||
|
||
⚠ 这个数只回答「限价单若仅有首次触及那一根可以成交，能否成交」。真实的
|
||
挂单是常驻的：它在那儿放最多 48 根，每根都在成交，且价格决定性穿过限价
|
||
时整根成交量都可用。所以本函数系统性**偏悲观**，不能当作成交率结论。
|
||
真实成交率见 lib/exit_fill.walk_filled 与 step43_fill_aware_budget.py，
|
||
那里按逐根累积结算，10 万仓位下预算降幅不足 1%。
|
||
"""
|
||
if not pen_path.exists():
|
||
print("\n没有 penetration.csv，先跑 penetration.py")
|
||
return
|
||
kgrid = np.linspace(0.0, 1.0, 51)
|
||
shape = tape_shape(tape_path, kgrid)
|
||
if not shape:
|
||
print("\n成交流样本不足，无法定形状")
|
||
return
|
||
pen = pd.read_csv(pen_path)
|
||
|
||
print(f"\n\n########## maker 腿真实成交率 ##########")
|
||
print(f" 穿透深度分布（历史 63 万次触及）× 每根成交额（210 天）")
|
||
print(f" × 区间内成交分布形状（影子成交流）\n")
|
||
|
||
for sym in sorted(shape):
|
||
cands = sorted(cache.glob(f"bitget_{sym}_1m_*.feather"),
|
||
key=lambda p: p.stat().st_size, reverse=True)
|
||
if not cands:
|
||
continue
|
||
bars = pd.read_feather(cands[0])
|
||
# 每根的主动买成交额。取总成交额的一半——买卖大致均衡，且这与
|
||
# 成交流实测的买卖比一致
|
||
bar_notional = (bars["volume"].to_numpy(float)
|
||
* bars["close"].to_numpy(float)) * 0.5
|
||
bar_notional = bar_notional[np.isfinite(bar_notional)
|
||
& (bar_notional > 0)]
|
||
f = shape[sym]
|
||
for tgt in sorted(pen["target_atr"].unique()):
|
||
k = pen[(pen["sym"] == sym)
|
||
& (pen["target_atr"] == tgt)]["k"].to_numpy(float)
|
||
if not k.size:
|
||
continue
|
||
# 独立配对：穿透位置与该根成交额各自抽样。真实触及根多为放量根，
|
||
# 故此处偏**保守**（低估可成交量）
|
||
rng = np.random.default_rng(0)
|
||
m = 200_000
|
||
ks = rng.choice(k, m)
|
||
ns = rng.choice(bar_notional, m)
|
||
avail = np.interp(ks, kgrid, f) * ns
|
||
print(f" {sym} · 目标 {tgt:g}ATR · 每根主动买额中位 "
|
||
f"{np.median(bar_notional):,.0f} USDT")
|
||
for nt in notionals:
|
||
full = float((avail >= nt).mean())
|
||
half = float((avail >= nt / 2).mean())
|
||
print(f" 仓位 {nt:>9,.0f}：全额成交 {full * 100:5.1f}%"
|
||
f" · 至少半额 {half * 100:5.1f}%"
|
||
f" · 可成交额中位 {np.median(avail):>10,.0f}")
|
||
# 成交率反推的资金上限。这才是绑定约束——它比冲击反推的上限
|
||
# 低一到两个数量级，而后者才是通常被当作「容量」的那个数
|
||
for want in (0.80, 0.90):
|
||
cap = float(np.quantile(avail, 1.0 - want))
|
||
print(f" → 要 {want * 100:.0f}% 的止盈全额成交，"
|
||
f"仓位须 ≤ {cap:,.0f} USDT")
|
||
print()
|
||
print(" 未计排队（我们的单排在该价位既有挂单之后），故仍是上界。")
|
||
print(" 回测把这些止盈按「全额成交在目标价」计，差多少就是收益虚多少")
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--books", default=None)
|
||
ap.add_argument("--tape", default=None)
|
||
a = ap.parse_args()
|
||
d = out_dir()
|
||
from lib.shadow_budget import BUDGET_BP
|
||
tape = Path(a.tape) if a.tape else d / "shadow_tape.jsonl.gz"
|
||
books = Path(a.books) if a.books else d / "shadow_books.jsonl.gz"
|
||
capacity(books, BUDGET_BP)
|
||
queue_ahead(books)
|
||
maker_fill(tape)
|
||
composite_fill(tape, d / "penetration.csv",
|
||
Path(__file__).resolve().parent / "cache")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|