368 lines
13 KiB
Python
368 lines
13 KiB
Python
"""
|
||
agg_trades_collector.py — aggTrades全量采集守护进程
|
||
|
||
架构：
|
||
- WebSocket主链路：实时推送，延迟<100ms
|
||
- REST补洞：断线重连后从last_agg_id追平
|
||
- 每分钟巡检：校验agg_id连续性，发现断档自动补洞
|
||
- 批量写入：攒200条或1秒flush一次，减少WAL压力
|
||
- 按月分表：agg_trades_YYYYMM，单表千万行内查询快
|
||
- 健康接口：GET /collector/health 可监控
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import sqlite3
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
import websockets
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
handlers=[
|
||
logging.StreamHandler(),
|
||
logging.FileHandler(os.path.join(os.path.dirname(__file__), "..", "collector.log")),
|
||
],
|
||
)
|
||
logger = logging.getLogger("collector")
|
||
|
||
DB_PATH = os.path.join(os.path.dirname(__file__), "..", "arb.db")
|
||
BINANCE_FAPI = "https://fapi.binance.com/fapi/v1"
|
||
SYMBOLS = ["BTCUSDT", "ETHUSDT"]
|
||
HEADERS = {"User-Agent": "Mozilla/5.0 ArbitrageEngine/3.0"}
|
||
|
||
# 批量写入缓冲
|
||
_buffer: dict[str, list] = {s: [] for s in SYMBOLS}
|
||
BATCH_SIZE = 200
|
||
BATCH_TIMEOUT = 1.0 # seconds
|
||
|
||
|
||
# ─── DB helpers ──────────────────────────────────────────────────
|
||
|
||
def get_conn() -> sqlite3.Connection:
|
||
conn = sqlite3.connect(DB_PATH, timeout=30)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA journal_mode=WAL")
|
||
conn.execute("PRAGMA synchronous=NORMAL")
|
||
return conn
|
||
|
||
|
||
def table_name(ts_ms: int) -> str:
|
||
"""按月分表：agg_trades_202602"""
|
||
dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
|
||
return f"agg_trades_{dt.strftime('%Y%m')}"
|
||
|
||
|
||
def ensure_table(conn: sqlite3.Connection, tname: str):
|
||
conn.execute(f"""
|
||
CREATE TABLE IF NOT EXISTS {tname} (
|
||
agg_id INTEGER PRIMARY KEY,
|
||
symbol TEXT NOT NULL,
|
||
price REAL NOT NULL,
|
||
qty REAL NOT NULL,
|
||
first_trade_id INTEGER,
|
||
last_trade_id INTEGER,
|
||
time_ms INTEGER NOT NULL,
|
||
is_buyer_maker INTEGER NOT NULL
|
||
)
|
||
""")
|
||
conn.execute(f"""
|
||
CREATE INDEX IF NOT EXISTS idx_{tname}_sym_time
|
||
ON {tname}(symbol, time_ms)
|
||
""")
|
||
|
||
|
||
def ensure_meta_table(conn: sqlite3.Connection):
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS agg_trades_meta (
|
||
symbol TEXT PRIMARY KEY,
|
||
last_agg_id INTEGER NOT NULL,
|
||
last_time_ms INTEGER NOT NULL,
|
||
updated_at TEXT DEFAULT (datetime('now'))
|
||
)
|
||
""")
|
||
|
||
|
||
def get_last_agg_id(symbol: str) -> Optional[int]:
|
||
try:
|
||
conn = get_conn()
|
||
ensure_meta_table(conn)
|
||
row = conn.execute(
|
||
"SELECT last_agg_id FROM agg_trades_meta WHERE symbol = ?", (symbol,)
|
||
).fetchone()
|
||
conn.close()
|
||
return row["last_agg_id"] if row else None
|
||
except Exception as e:
|
||
logger.error(f"get_last_agg_id error: {e}")
|
||
return None
|
||
|
||
|
||
def update_meta(conn: sqlite3.Connection, symbol: str, last_agg_id: int, last_time_ms: int):
|
||
conn.execute("""
|
||
INSERT INTO agg_trades_meta (symbol, last_agg_id, last_time_ms, updated_at)
|
||
VALUES (?, ?, ?, datetime('now'))
|
||
ON CONFLICT(symbol) DO UPDATE SET
|
||
last_agg_id = excluded.last_agg_id,
|
||
last_time_ms = excluded.last_time_ms,
|
||
updated_at = excluded.updated_at
|
||
""", (symbol, last_agg_id, last_time_ms))
|
||
|
||
|
||
def flush_buffer(symbol: str, trades: list) -> int:
|
||
"""写入一批trades，返回实际写入条数（去重后）"""
|
||
if not trades:
|
||
return 0
|
||
try:
|
||
conn = get_conn()
|
||
ensure_meta_table(conn)
|
||
# 按月分组
|
||
by_month: dict[str, list] = {}
|
||
for t in trades:
|
||
tname = table_name(t["T"])
|
||
if tname not in by_month:
|
||
by_month[tname] = []
|
||
by_month[tname].append(t)
|
||
|
||
inserted = 0
|
||
last_agg_id = 0
|
||
last_time_ms = 0
|
||
|
||
for tname, batch in by_month.items():
|
||
ensure_table(conn, tname)
|
||
for t in batch:
|
||
cur = conn.execute(
|
||
f"""INSERT OR IGNORE INTO {tname}
|
||
(agg_id, symbol, price, qty, first_trade_id, last_trade_id, time_ms, is_buyer_maker)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
(
|
||
t["a"], # agg_id
|
||
symbol,
|
||
float(t["p"]),
|
||
float(t["q"]),
|
||
t.get("f"), # first_trade_id
|
||
t.get("l"), # last_trade_id
|
||
t["T"], # time_ms
|
||
1 if t["m"] else 0, # is_buyer_maker
|
||
)
|
||
)
|
||
if cur.rowcount > 0:
|
||
inserted += 1
|
||
if t["a"] > last_agg_id:
|
||
last_agg_id = t["a"]
|
||
last_time_ms = t["T"]
|
||
|
||
if last_agg_id > 0:
|
||
update_meta(conn, symbol, last_agg_id, last_time_ms)
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
return inserted
|
||
except Exception as e:
|
||
logger.error(f"flush_buffer [{symbol}] error: {e}")
|
||
return 0
|
||
|
||
|
||
# ─── REST补洞 ────────────────────────────────────────────────────
|
||
|
||
async def rest_catchup(symbol: str, from_id: int) -> int:
|
||
"""从from_id开始REST拉取，追平到最新，返回补洞条数"""
|
||
total = 0
|
||
current_id = from_id
|
||
logger.info(f"[{symbol}] REST catchup from agg_id={from_id}")
|
||
|
||
async with httpx.AsyncClient(timeout=15, headers=HEADERS) as client:
|
||
while True:
|
||
try:
|
||
resp = await client.get(
|
||
f"{BINANCE_FAPI}/aggTrades",
|
||
params={"symbol": symbol, "fromId": current_id, "limit": 1000},
|
||
)
|
||
if resp.status_code != 200:
|
||
logger.warning(f"[{symbol}] REST catchup HTTP {resp.status_code}")
|
||
break
|
||
data = resp.json()
|
||
if not data:
|
||
break
|
||
count = flush_buffer(symbol, data)
|
||
total += count
|
||
last = data[-1]["a"]
|
||
if last <= current_id:
|
||
break
|
||
current_id = last + 1
|
||
# 如果拉到的比最新少1000条，说明追平了
|
||
if len(data) < 1000:
|
||
break
|
||
await asyncio.sleep(0.1) # rate limit友好
|
||
except Exception as e:
|
||
logger.error(f"[{symbol}] REST catchup error: {e}")
|
||
break
|
||
|
||
logger.info(f"[{symbol}] REST catchup done, filled {total} trades")
|
||
return total
|
||
|
||
|
||
# ─── WebSocket采集 ───────────────────────────────────────────────
|
||
|
||
async def ws_collect(symbol: str):
|
||
"""单Symbol的WS采集循环，自动断线重连+REST补洞"""
|
||
stream = symbol.lower() + "@aggTrade"
|
||
url = f"wss://fstream.binance.com/ws/{stream}"
|
||
buffer: list = []
|
||
last_flush = time.time()
|
||
reconnect_delay = 1.0
|
||
|
||
while True:
|
||
try:
|
||
# 断线重连前先REST补洞
|
||
last_id = get_last_agg_id(symbol)
|
||
if last_id is not None:
|
||
await rest_catchup(symbol, last_id + 1)
|
||
|
||
logger.info(f"[{symbol}] Connecting WS: {url}")
|
||
async with websockets.connect(url, ping_interval=20, ping_timeout=10) as ws:
|
||
reconnect_delay = 1.0 # 连上了就重置
|
||
logger.info(f"[{symbol}] WS connected")
|
||
|
||
async for raw in ws:
|
||
msg = json.loads(raw)
|
||
if msg.get("e") != "aggTrade":
|
||
continue
|
||
|
||
buffer.append(msg)
|
||
|
||
# 批量flush：满200条或超1秒
|
||
now = time.time()
|
||
if len(buffer) >= BATCH_SIZE or (now - last_flush) >= BATCH_TIMEOUT:
|
||
count = flush_buffer(symbol, buffer)
|
||
if count > 0:
|
||
logger.debug(f"[{symbol}] flushed {count}/{len(buffer)} trades")
|
||
buffer.clear()
|
||
last_flush = now
|
||
|
||
except websockets.exceptions.ConnectionClosed as e:
|
||
logger.warning(f"[{symbol}] WS closed: {e}, reconnecting in {reconnect_delay}s")
|
||
except Exception as e:
|
||
logger.error(f"[{symbol}] WS error: {e}, reconnecting in {reconnect_delay}s")
|
||
finally:
|
||
# flush剩余buffer
|
||
if buffer:
|
||
flush_buffer(symbol, buffer)
|
||
buffer.clear()
|
||
|
||
await asyncio.sleep(reconnect_delay)
|
||
reconnect_delay = min(reconnect_delay * 2, 30) # exponential backoff, max 30s
|
||
|
||
|
||
# ─── 连续性巡检 ──────────────────────────────────────────────────
|
||
|
||
async def continuity_check():
|
||
"""每60秒巡检一次：检查各symbol最近的agg_id是否有断档"""
|
||
while True:
|
||
await asyncio.sleep(60)
|
||
try:
|
||
conn = get_conn()
|
||
ensure_meta_table(conn)
|
||
for symbol in SYMBOLS:
|
||
row = conn.execute(
|
||
"SELECT last_agg_id, last_time_ms FROM agg_trades_meta WHERE symbol = ?",
|
||
(symbol,)
|
||
).fetchone()
|
||
if not row:
|
||
continue
|
||
# 检查最近1000条是否连续
|
||
now_month = datetime.now(tz=timezone.utc).strftime("%Y%m")
|
||
tname = f"agg_trades_{now_month}"
|
||
try:
|
||
rows = conn.execute(
|
||
f"SELECT agg_id FROM {tname} WHERE symbol = ? ORDER BY agg_id DESC LIMIT 100",
|
||
(symbol,)
|
||
).fetchall()
|
||
if len(rows) < 2:
|
||
continue
|
||
ids = [r["agg_id"] for r in rows]
|
||
# 检查是否连续（降序）
|
||
gaps = []
|
||
for i in range(len(ids) - 1):
|
||
diff = ids[i] - ids[i + 1]
|
||
if diff > 1:
|
||
gaps.append((ids[i + 1], ids[i], diff - 1))
|
||
if gaps:
|
||
logger.warning(f"[{symbol}] Found {len(gaps)} gaps in recent data: {gaps[:3]}")
|
||
# 触发补洞
|
||
min_gap_id = min(g[0] for g in gaps)
|
||
asyncio.create_task(rest_catchup(symbol, min_gap_id))
|
||
else:
|
||
logger.debug(f"[{symbol}] Continuity OK, last_agg_id={row['last_agg_id']}")
|
||
except Exception:
|
||
pass # 表可能还不存在
|
||
conn.close()
|
||
except Exception as e:
|
||
logger.error(f"Continuity check error: {e}")
|
||
|
||
|
||
# ─── 每日完整性报告 ──────────────────────────────────────────────
|
||
|
||
async def daily_report():
|
||
"""每小时生成一次完整性摘要日志"""
|
||
while True:
|
||
await asyncio.sleep(3600)
|
||
try:
|
||
conn = get_conn()
|
||
ensure_meta_table(conn)
|
||
report_lines = ["=== AggTrades Integrity Report ==="]
|
||
for symbol in SYMBOLS:
|
||
row = conn.execute(
|
||
"SELECT last_agg_id, last_time_ms FROM agg_trades_meta WHERE symbol = ?",
|
||
(symbol,)
|
||
).fetchone()
|
||
if not row:
|
||
report_lines.append(f" {symbol}: No data yet")
|
||
continue
|
||
last_dt = datetime.fromtimestamp(row["last_time_ms"] / 1000, tz=timezone.utc).isoformat()
|
||
# 统计本月总量
|
||
now_month = datetime.now(tz=timezone.utc).strftime("%Y%m")
|
||
tname = f"agg_trades_{now_month}"
|
||
try:
|
||
count = conn.execute(
|
||
f"SELECT COUNT(*) as c FROM {tname} WHERE symbol = ?", (symbol,)
|
||
).fetchone()["c"]
|
||
except Exception:
|
||
count = 0
|
||
report_lines.append(
|
||
f" {symbol}: last_agg_id={row['last_agg_id']}, last_time={last_dt}, month_count={count:,}"
|
||
)
|
||
conn.close()
|
||
logger.info("\n".join(report_lines))
|
||
except Exception as e:
|
||
logger.error(f"Daily report error: {e}")
|
||
|
||
|
||
# ─── 入口 ────────────────────────────────────────────────────────
|
||
|
||
async def main():
|
||
logger.info("AggTrades Collector starting...")
|
||
# 确保基础表存在
|
||
conn = get_conn()
|
||
ensure_meta_table(conn)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
# 并行启动所有symbol的WS + 巡检 + 报告
|
||
tasks = [
|
||
ws_collect(sym) for sym in SYMBOLS
|
||
] + [
|
||
continuity_check(),
|
||
daily_report(),
|
||
]
|
||
await asyncio.gather(*tasks)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|