arbitrage-engine/backend/agg_trades_collector.py

"""
agg_trades_collector.py — aggTrades全量采集守护进程（PostgreSQL版）

架构：
  - WebSocket主链路：实时推送，延迟<100ms
  - REST补洞：断线重连后从last_agg_id追平
  - 每分钟巡检：校验agg_id连续性，发现断档自动补洞
  - 批量写入：攒200条或1秒flush一次
  - PG分区表：按月自动分区，MVCC并发无锁冲突
"""

import asyncio
import json
import logging
import os
import time
from datetime import datetime, timezone
from typing import Optional

import httpx
import psycopg2
import psycopg2.extras
import websockets

from db import get_sync_conn, get_sync_pool, ensure_partitions, PG_HOST, PG_PORT, PG_DB, PG_USER, PG_PASS

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")

BINANCE_FAPI = "https://fapi.binance.com/fapi/v1"
SYMBOLS = ["BTCUSDT", "ETHUSDT"]
HEADERS = {"User-Agent": "Mozilla/5.0 ArbitrageEngine/3.0"}

BATCH_SIZE = 200
BATCH_TIMEOUT = 1.0


# ─── DB helpers ──────────────────────────────────────────────────

def get_last_agg_id(symbol: str) -> Optional[int]:
    try:
        with get_sync_conn() as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT last_agg_id FROM agg_trades_meta WHERE symbol = %s", (symbol,))
                row = cur.fetchone()
                return row[0] if row else None
    except Exception as e:
        logger.error(f"get_last_agg_id error: {e}")
        return None


def update_meta(conn, symbol: str, last_agg_id: int, last_time_ms: int):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO agg_trades_meta (symbol, last_agg_id, last_time_ms, updated_at)
            VALUES (%s, %s, %s, NOW())
            ON CONFLICT(symbol) DO UPDATE SET
                last_agg_id = EXCLUDED.last_agg_id,
                last_time_ms = EXCLUDED.last_time_ms,
                updated_at = NOW()
        """, (symbol, last_agg_id, last_time_ms))


def flush_buffer(symbol: str, trades: list) -> int:
    """写入一批trades到PG，返回实际写入条数"""
    if not trades:
        return 0
    try:
        # 确保分区存在
        ensure_partitions()

        with get_sync_conn() as conn:
            with conn.cursor() as cur:
                # 批量插入（ON CONFLICT忽略重复）
                values = []
                last_agg_id = 0
                last_time_ms = 0

                for t in trades:
                    agg_id = t["a"]
                    time_ms = t["T"]
                    values.append((
                        agg_id, symbol,
                        float(t["p"]), float(t["q"]),
                        time_ms,
                        1 if t["m"] else 0,
                    ))
                    if agg_id > last_agg_id:
                        last_agg_id = agg_id
                        last_time_ms = time_ms

                # 批量INSERT
                psycopg2.extras.execute_values(
                    cur,
                    """INSERT INTO agg_trades (agg_id, symbol, price, qty, time_ms, is_buyer_maker)
                    VALUES %s
                    ON CONFLICT (time_ms, symbol, agg_id) DO NOTHING""",
                    values,
                    template="(%s, %s, %s, %s, %s, %s)",
                    page_size=1000,
                )
                inserted = cur.rowcount

                if last_agg_id > 0:
                    update_meta(conn, symbol, last_agg_id, last_time_ms)

                conn.commit()
                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:
    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
                if len(data) < 1000:
                    break
                await asyncio.sleep(0.1)
            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):
    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:
            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)

                    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:
            if buffer:
                flush_buffer(symbol, buffer)
                buffer.clear()

        await asyncio.sleep(reconnect_delay)
        reconnect_delay = min(reconnect_delay * 2, 30)


# ─── 连续性巡检 ──────────────────────────────────────────────────

async def continuity_check():
    while True:
        await asyncio.sleep(60)
        try:
            with get_sync_conn() as conn:
                with conn.cursor() as cur:
                    for symbol in SYMBOLS:
                        cur.execute(
                            "SELECT last_agg_id, last_time_ms FROM agg_trades_meta WHERE symbol = %s",
                            (symbol,)
                        )
                        row = cur.fetchone()
                        if not row:
                            continue
                        # 检查最近100条是否连续
                        cur.execute(
                            "SELECT agg_id FROM agg_trades WHERE symbol = %s ORDER BY agg_id DESC LIMIT 100",
                            (symbol,)
                        )
                        rows = cur.fetchall()
                        if len(rows) < 2:
                            continue
                        ids = [r[0] 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: {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[0]}")
        except Exception as e:
            logger.error(f"Continuity check error: {e}")


# ─── 每小时报告 ──────────────────────────────────────────────────

async def daily_report():
    while True:
        await asyncio.sleep(3600)
        try:
            with get_sync_conn() as conn:
                with conn.cursor() as cur:
                    report = ["=== AggTrades Integrity Report ==="]
                    for symbol in SYMBOLS:
                        cur.execute(
                            "SELECT last_agg_id, last_time_ms FROM agg_trades_meta WHERE symbol = %s",
                            (symbol,)
                        )
                        row = cur.fetchone()
                        if not row:
                            report.append(f"  {symbol}: No data yet")
                            continue
                        last_dt = datetime.fromtimestamp(row[1] / 1000, tz=timezone.utc).isoformat()
                        cur.execute(
                            "SELECT COUNT(*) FROM agg_trades WHERE symbol = %s", (symbol,)
                        )
                        count = cur.fetchone()[0]
                        report.append(f"  {symbol}: last_agg_id={row[0]}, last_time={last_dt}, total={count:,}")
                    logger.info("\n".join(report))
        except Exception as e:
            logger.error(f"Report error: {e}")


# ─── 入口 ────────────────────────────────────────────────────────

async def main():
    logger.info("AggTrades Collector (PG) starting...")
    # 确保分区存在
    ensure_partitions()

    tasks = [
        ws_collect(sym) for sym in SYMBOLS
    ] + [
        continuity_check(),
        daily_report(),
    ]
    await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(main())