從零寫一個實時行情錄製器：斷線續錄、缺口檢測與可回放數據集

{
  "action": "subscribe",
  "market": "crypto",
  "symbols": ["BTCUSDT", "ETHUSDT"]
}

#!/usr/bin/env python3
"""實時行情錄製器：把 WebSocket tick 流落成按 UTC 日期分片的 JSONL。
依賴：pip install websockets
運行：TIQEX_API_KEY=你的密鑰 python3 recorder.py
"""
import asyncio
import json
import os
import time
from datetime import datetime
from pathlib import Path

import websockets

WS_URL = "wss://api.tiqex.io/api/v1/market/stream"
API_KEY = os.environ["TIQEX_API_KEY"]     # 從環境變量讀，不要寫死在代碼裡
MARKET = "crypto"
SYMBOLS = ["BTCUSDT", "ETHUSDT"]
OUT_DIR = Path("ticks")
FLUSH_EVERY = 200                         # 攢夠 200 條落一次盤


class Sink:
    """帶緩衝的追加寫入器，同一天複用同一個文件句柄。"""

    def __init__(self) -> None:
        OUT_DIR.mkdir(exist_ok=True)
        self._fh = None
        self._day = None
        self._buf: list[str] = []
        self.total = 0

    def add(self, record: dict) -> None:
        ts = datetime.fromisoformat(record["at"].replace("Z", "+00:00"))
        day = ts.strftime("%Y-%m-%d")
        if day != self._day:              # 跨天：換分片文件
            self.flush()
            if self._fh:
                self._fh.close()
            self._fh = open(OUT_DIR / f"{MARKET}-{day}.jsonl", "a", encoding="utf-8")
            self._day = day
        self._buf.append(json.dumps(record, ensure_ascii=False))
        self.total += 1
        if len(self._buf) >= FLUSH_EVERY:
            self.flush()

    def flush(self) -> None:
        if self._buf and self._fh:
            self._fh.write("\n".join(self._buf) + "\n")
            self._fh.flush()              # 先過 Python 緩衝
            os.fsync(self._fh.fileno())   # 再落盤，進程被殺也不丟
            self._buf.clear()


def normalize(frame: dict) -> dict | None:
    data = frame.get("data") or {}
    if data.get("type") != "tick":
        return None
    return {
        "symbol": data.get("symbol"),
        "at": data.get("at"),                      # RFC3339 行情時間
        "last": data.get("last"),
        "bid": data.get("bid"),
        "ask": data.get("ask"),
        "volume": data.get("volume"),
        "recvMs": int(time.time() * 1000),         # 本地接收時間，用於算延遲
    }


async def record() -> None:
    sink = Sink()
    backoff = 1

    while True:
        try:
            async with websockets.connect(
                WS_URL,
                additional_headers={"X-API-Key": API_KEY},  # websockets<14 用 extra_headers
                ping_interval=20,
                ping_timeout=20,
                max_queue=4096,                             # 隊列滿則丟棄，不阻塞
            ) as ws:
                await ws.send(json.dumps({
                    "action": "subscribe",
                    "market": MARKET,
                    "symbols": SYMBOLS,
                }))
                backoff = 1                                 # 連上即重置退避
                async for raw in ws:
                    frame = json.loads(raw)
                    if not frame.get("succ"):
                        print(f"[error] {frame.get('code')} {frame.get('msg')}")
                        continue
                    data = frame.get("data") or {}
                    if data.get("type") == "subscribed":
                        print(f"[ack] {data.get('market')} {data.get('symbols')}")
                        continue
                    rec = normalize(frame)
                    if rec is not None:
                        sink.add(rec)
        except (websockets.ConnectionClosed, OSError) as exc:
            print(f"[reconnect] {type(exc).__name__} -> retry in {backoff}s")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 30)                  # 指數退避，封頂 30 秒
        except asyncio.CancelledError:
            sink.flush()
            print(f"[done] 共寫入 {sink.total} 條 tick")
            raise


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

#!/usr/bin/env python3
"""掃描錄製的 JSONL，輸出缺口 / 重複 / 覆蓋率報告。"""
import json
from datetime import datetime
from pathlib import Path

GAP_SEC = 60          # 交易時段內超過 60 秒沒有 tick，記為缺口


def parse(ts: str) -> datetime:
    return datetime.fromisoformat(ts.replace("Z", "+00:00"))


def scan(path: Path) -> dict:
    rows = [json.loads(l) for l in path.read_text(encoding="utf-8").splitlines() if l]
    rows.sort(key=lambda r: r["at"])           # 亂序到達是常態，先排序

    by_symbol: dict[str, list[dict]] = {}
    for r in rows:
        by_symbol.setdefault(r["symbol"], []).append(r)

    report = {"file": path.name, "rows": len(rows), "duplicates": 0, "symbols": {}}
    for symbol, series in by_symbol.items():
        gaps, dups = [], 0
        prev = None
        for r in series:
            ts = parse(r["at"])
            if prev is not None:
                delta = (ts - prev).total_seconds()
                if delta == 0:
                    dups += 1                  # 同一 (symbol, at) 重複到達
                elif delta > GAP_SEC:
                    gaps.append({"from": prev.isoformat(),
                                 "to": ts.isoformat(),
                                 "seconds": round(delta, 3)})
            prev = ts
        report["symbols"][symbol] = {"ticks": len(series),
                                     "gaps": gaps, "duplicates": dups}
        report["duplicates"] += dups
    return report


if __name__ == "__main__":
    for f in sorted(Path("ticks").glob("*.jsonl")):
        print(json.dumps(scan(f), ensure_ascii=False, indent=2))

#!/usr/bin/env python3
"""按錄製時的真實節奏回放 tick，speed=10 表示 10 倍速。"""
import asyncio
import json
from datetime import datetime
from pathlib import Path


def parse(ts: str) -> datetime:
    return datetime.fromisoformat(ts.replace("Z", "+00:00"))


async def replay(path: Path, speed: float = 1.0):
    rows = [json.loads(l) for l in path.read_text(encoding="utf-8").splitlines() if l]
    rows.sort(key=lambda r: r["at"])
    t0_wall = parse(rows[0]["at"])
    t0_mono = asyncio.get_running_loop().time()
    for r in rows:
        target = (parse(r["at"]) - t0_wall).total_seconds() / speed
        wait = target - (asyncio.get_running_loop().time() - t0_mono)
        if wait > 0:
            await asyncio.sleep(wait)
        yield r                      # 消費方：策略、告警、看板


async def main():
    async for tick in replay(Path("ticks/crypto-2026-09-19.jsonl"), speed=10.0):
        print(tick["symbol"], tick["last"])


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