Binance Futures 实时 Tick 数据采集、存储与分析管道。
通过 WebSocket 直连 Binance Futures，采集 10 个示例币种的 orderbook depth (50档) + trade 数据，每秒批量写入本地 Parquet 文件（VPS 部署）。
-orange.svg){kind=icon}
┌─────────────────┐ WebSocket ┌──────────────────────┐
│ Binance Futures │─────────────────────▶│ collect_tick.py │
│ (depth@100ms + │ │ │
│ trade streams) │ REST Snapshot │ LocalOrderBook │
│ │◀─────────────────────│ TradeAccumulator │
└─────────────────┘ │ TickBuffer │
│ │
│ 每秒 flush ────────▶│
└──────────────────────┘
│
▼
┌─────────────────┐
│ 本地 Parquet │
│ (VPS 存储) │
│ │
│ ticks/{sym}/ │
│ YYYY-MM-DD/ │
│ HH-MM.pq │
└─────────────────┘
数据管线: Binance WS → collector → 本地 Parquet 文件 (VPS)
- WebSocket 连接: 直连
wss://fstream.binance.com/ws，每连接 ≤512 对 stream (depth + trade) - 订单簿维护:
LocalOrderBook先通过 REST snapshot 初始化，再用 WebSocket diff 增量更新 - 成交累加:
TradeAccumulator按 taker side (buy/sell) 分别聚合成交统计 - 每秒 flush: 将当前 orderbook top-50 + 一秒内成交统计打包为一行
- 写入存储: 本地 Parquet 文件 (按分钟分文件，VPS 本地磁盘)
- 自动重同步: 每 30 分钟自动 REST snapshot 重同步，防止 orderbook 漂移
- 限流保护: 遇 418/429 自动降速，指数退避重试
- 数据验证: 每 60 批本地验证数据完整性
| 指标 | 数值 |
|---|---|
| 币种数量 | 10 个示例币种 |
| 采集频率 | 每秒 1 行 / symbol |
| 数据写入速率 | ~10 rows/s |
| 2 小时行数 | ~72,000 行 |
| 2 小时文件数 | ~1,200 个 Parquet 文件 |
| 2 小时磁盘占用 | ~72 MB (zstd compressed) |
| 币种 | 修复前数据点 | 修复后数据点 | 说明 |
|---|---|---|---|
| BTCUSDT | 7,200 | 7,200 | 高频币种，不受影响 |
| DOTUSDT | 0 | 7,200 | GAP-FIX 修复后正常采集 |
修复说明: GAP-FIX 解决了中低频币种 orderbook init → gap → resync 死循环导致的永久空数据问题。基于测试期间的 10 个示例币种。 可使用
--symbols参数配置任意 USDT 交易对，不受限于这 10 个示例。
- Python 3.11+
- pip
git clone https://github.com/hellozim22/tick-data-pipeline.git
cd tick-data-pipeline
pip install -r collector/requirements.txt# 采集所有 10 个示例币种，存储为本地 Parquet
python3 collector/collect_tick.py --local --data-dir ../data
# 只采集特定币种
python3 collector/collect_tick.py --local --symbols BTCUSDT,ETHUSDT,SOLUSDT
# 本地 Parquet 模式 (默认)
python3 collector/collect_tick.py --storage local --symbols BTCUSDT# 检查所有示例数据
python3 scripts/check_data_quality.py examples/
# 检查具体文件
python3 scripts/check_data_quality.py examples/btcusdt_sample.parquetpip install pytest pytest-asyncio
pytest tests/ -v每个 tick 包含 39 个字段，覆盖 orderbook depth 和 trade 统计。
详见 完整 Schema 文档。
| 类别 | 字段 | 说明 |
|---|---|---|
| 标识 | id, symbol_id, ts |
自增 ID, 币种 ID, 毫秒时间戳 |
| 订单簿 | bp[50], ap[50], bv[50], av[50] |
买卖价量, 各 50 档 |
| Long 成交 | ln, lv, lq, lmax, lmin, lfirst, llast |
买方主动成交统计 |
| Short 成交 | sn, sv, sq, smax, smin, sfirst, slast |
卖方主动成交统计 |
| BTC 订单簿 | btc_bp[50], btc_ap[50], btc_bv[50], btc_av[50] |
BTC/USDT 参考订单簿 |
| BTC 成交 | btc_ln~btc_slast (14 字段) |
BTC/USDT 成交统计 |
{data_dir}/ticks/{SYMBOL}/{YYYY-MM-DD}/{HH-MM}.parquet
例如: data/ticks/BTCUSDT/2026-06-23/06-45.parquet
- 增量压缩: 当前每分钟一个 Parquet 文件，可合并为每小时文件减少 inode 开销
- 水平扩展: 支持按币种分组多进程/多机器采集
- 实时告警: 接入 Prometheus metrics，异常时推送告警
- 回填工具: 自动补齐因重启/网络中断造成的数据缺失
- ClickHouse 后端: 为高频查询场景提供列存分析引擎
tick-data-pipeline/
├── README.md
├── LICENSE
├── .env.example
├── .gitignore
├── collector/ # 核心采集器
│ ├── __init__.py
│ ├── collect_tick.py # 主采集器 (WS + REST snapshot)
│ ├── local_storage.py # 本地 Parquet 存储后端
│ └── requirements.txt
├── tests/ # 测试
│ ├── __init__.py
│ ├── test_orderbook.py # 订单簿 gap 容忍逻辑测试
│ └── test_data_quality.py # 数据完整性测试
├── scripts/ # 运维脚本
│ ├── check_data_quality.py # 数据质量检查
│ └── sample_data.py # 数据抽样工具
├── docs/ # 文档
│ ├── DATA_SCHEMA.md # 完整 39 字段说明
└── examples/ # 示例
└── sample_analysis.py # 样本数据读取演示
本地开发运行即可，生产环境可通过 cron 调度:
# 每 10 小时重启采集器（防止内存累积）
0 */10 * * * /opt/tick-collector/run_10h.sh- 每秒采集 1 行 / symbol
- 10 symbols × 1 row/s × 3600s/h = ~36,000 rows/h
- 每行 ~1KB (zstd compressed) → ~36 MB/h
- 每日 ~864 MB
最新质量评估 (基于 10 个示例币种): 96/100 ✅
| 指标 | 结果 |
|---|---|
| 字段完整性 | 100% (39/39 字段，无 missing values) |
| 订单簿排序 | ✅ 全部正确 (bids 降序, asks 升序) |
| 时间戳单调性 | ✅ 单调递增 |
| 跨币种一致性 | ✅ BTC 参考列一致 |
个人非商业使用许可 — 详见 LICENSE
- ✅ 个人学习、研究、私下传播
- ❌ 任何形式的商业用途需授权