Recording live market data looks trivial: subscribe to a stream, append each tick to a file. Four problems surface at once — the window lost during a disconnect, a tick that arrives twice, sharding across midnight, and proving the dataset is complete. This tutorial builds a live market data recorder: subscribe over WebSocket at wss://api.tiqex.io/api/v1/market/stream, persist ticks as UTC-day-sharded JSONL, emit a gap report, and replay at original cadence.
REST snapshots suffice for a latest price; a recorder exists to capture what exists only in the moment: trade prints, bid/ask depth, millisecond timestamps.
Why record now instead of backfilling later
The instinct is to defer: "we will pull history when we need it." That holds at daily granularity, not at tick granularity.
- Retrievable later: daily and minute OHLCV bars — aggregations, re-fetchable anytime.
- Not retrievable later: trade prints, bid/ask quotes, volume increments, and the delta between local receive time and exchange timestamp.
The second category exists only when pushed; if it was not persisted then, nothing buys it back. That decides whether tick-level backtesting is viable: if your logic depends on order book state or trade detail, collection must start early and stay complete.
A second benefit is a latency baseline: writing both timestamps accumulates a latency series for free. See the tick data glossary entry for field composition.
The four hard requirements
Define correctness first. Four requirements must hold at once.
| Requirement | Typical failure mode | Mitigation |
|---|---|---|
| No loss | Process killed; the in-memory buffer evaporates | Batched writes plus flush() and fsync() |
| No duplicates | Reconnect writes the same tick twice | Use (symbol, at) as an idempotency key |
| Resumable | The disconnect window is missing, unnoticed | Track a high-water mark; reconcile via REST snapshot |
| Auditable | Hundreds of MB on disk, unknown completeness | Emit a gap report and a coverage metric |
Note the last two: gaps are not the danger — a dataset that merely looks complete is. Without a gap report, an anomalous backtest sends you debugging the strategy first.
Step 1: WebSocket subscription and frame handling
The endpoint is wss://api.tiqex.io/api/v1/market/stream, authenticated with X-API-Key (Authorization: Bearer also accepted). The first message is the subscribe payload:
{
"action": "subscribe",
"market": "crypto",
"symbols": ["BTCUSDT", "ETHUSDT"]
}Responses use a uniform succ / data / msg / code envelope; data.type determines handling.
| data.type | Meaning | Recorder action |
|---|---|---|
subscribed | Subscription ack with market and symbols | Log; reset the reconnect backoff |
tick | Live quote with symbol / last / at and bid / ask / volume | Normalize, then buffer |
error | Error frame with msg and code | Log; decide whether retrying helps |
One billing detail drives capture cost: the handshake counts toward daily quota and RPM, while pushed ticks are not re-counted per message — consumption tracks reconnect frequency, not volume. Plan caps (about 1 connection / 10 symbols on Free, 10 / 50 on Pro) mean partition by market across connections rather than over-subscribe, which returns WS_CONN_LIMIT. See the streaming documentation.
Step 2: Storage format and sharding
Write JSONL first and convert to Parquet offline; the write paths differ substantially.
| Format | Write complexity | Crash safety | Best for |
|---|---|---|---|
| JSONL | Very low, pure append | Good; at worst the last line is lost | Live capture |
| Parquet | High; needs a full row group | Fair; requires atomic rename | Offline analysis |
| SQLite | Medium; write locks and transactions | Good | Small-scale queries |
Capture cares about one property: the process can be killed at any moment without damaging existing data. JSONL gives that free; UTC-date sharding bounds file size and makes per-day replay or deletion a file operation.
#!/usr/bin/env python3
"""Live market data recorder: persist a WebSocket tick stream as
UTC-day-sharded JSONL.
Install: pip install websockets
Run: TIQEX_API_KEY=your_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"] # from the environment, never hardcoded
MARKET = "crypto"
SYMBOLS = ["BTCUSDT", "ETHUSDT"]
OUT_DIR = Path("ticks")
FLUSH_EVERY = 200 # flush every 200 records
class Sink:
"""Buffered append writer; reuses one handle per UTC day."""
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: # roll over at midnight UTC
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() # drain Python's buffer
os.fsync(self._fh.fileno()) # then force it to disk
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 exchange time
"last": data.get("last"),
"bid": data.get("bid"),
"ask": data.get("ask"),
"volume": data.get("volume"),
"recvMs": int(time.time() * 1000), # local receive time
}
async def record() -> None:
sink = Sink()
backoff = 1
while True:
try:
async with websockets.connect(
WS_URL,
additional_headers={"X-API-Key": API_KEY}, # extra_headers before v14
ping_interval=20,
ping_timeout=20,
max_queue=4096, # drop, never block
) as ws:
await ws.send(json.dumps({
"action": "subscribe",
"market": MARKET,
"symbols": SYMBOLS,
}))
backoff = 1 # reset on success
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) # capped exponential
except asyncio.CancelledError:
sink.flush()
print(f"[done] wrote {sink.total} ticks")
raise
if __name__ == "__main__":
asyncio.run(record())Key choices: flush() plus fsync() (the first hands data to the OS, the second persists it), backoff reset on successful connect, and bounded max_queue.
Step 3: Resuming after a disconnect, and gap detection
Reconnect logic must separate transient drops from authorization problems: the first warrants exponential backoff, the second never succeeds. The code field discriminates — NO_SUBSCRIPTION, MARKET_NOT_ENABLED, and API_KEY_SCOPE_DENIED are configuration errors — exit and alert; UPSTREAM_UNAVAILABLE and RATE_LIMIT_UNAVAILABLE are transient and worth retrying.
Quotes lost during a disconnect cannot be recovered, but they can be bounded: after reconnecting and receiving the first tick, fetch a REST snapshot and record the price at which the gap closed — GET https://api.tiqex.io/api/v1/market/quote?codes=BTCUSDT,ETHUSDT, codes comma-separated and capped at 50, with delayed and unavailable booleans per record.
The script below scans JSONL for gaps and duplicates. One trap: arrival order is not guaranteed to match exchange timestamp order, so sort by at first or out-of-order arrivals look like gaps.
#!/usr/bin/env python3
"""Scan recorded JSONL and report gaps, duplicates, and coverage."""
import json
from datetime import datetime
from pathlib import Path
GAP_SEC = 60 # more than 60s with no tick counts as a gap
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"]) # out-of-order arrival is normal
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 # same (symbol, at) twice
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))What matters is not rows but duplicates and gaps. Duplicates mean the deduplication key is not working; gaps mean strategy behavior in that window is not trustworthy.
Step 4: Replay and a completeness report
Recorded data must be replayable at its original cadence, or an offline alert test like "two triggers within 30 seconds" breaks on a compressed timeline. The replayer scales intervals from at and waits on the monotonic clock.
#!/usr/bin/env python3
"""Replay ticks at their original cadence; speed=10 means 10x."""
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 # consumers: strategy, alerts, dashboards
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())Archive four metrics with the dataset: coverage (flag below 99%); the gap list sorted by duration; the duplicate count keyed on (symbol, at), which should be zero; and the latency distribution of recvMs minus exchange timestamp at p50 / p95 / p99.
FAQ
Does a recorder burn a lot of API quota?
It depends on how you subscribe, not tick volume. The handshake counts toward daily quota and RPM, while pushed ticks are not re-counted per message — so a long-lived connection consumes quota in proportion to reconnect frequency, not message volume.
Can I backfill quotes missed during a disconnect?
Trade prints and order book quotes cannot; they exist only when pushed. Aggregations can — bars and daily candles are re-fetchable. Fill the price path with candles, but mark in the gap list that no tick detail exists for that window.
Why shard by UTC date instead of local date?
Because local dates cut one session into two files. A US equity session viewed from an Asia-based team spans local midnight, so local-date sharding splits one trading day in half and corrupts per-day statistics.
Can I run a recorder on the free plan?
Yes. Free starts at $0 with no credit card, includes 300 API calls per day and WebSocket streaming, and covers delayed crypto, US stocks, and metals. All-market real-time ticks require an upgraded plan; live crypto coverage is on the Crypto API market page.
Takeaways
A recorder's value is not its line count — it turns a one-off live stream into a reusable dataset. The four requirements map to buffered persistence, idempotency keys, backoff reconnects, and gap reports. Get single-symbol capture working, confirm at and recvMs are present, add gap detection until duplicates is zero, then wire in the replayer.
The code here performs capture and validation only. It is not investment advice and makes no performance claims about market data; latency and completeness depend on your network and upstream sources, so treat observed production metrics as authoritative. For the smallest starting example, read the WebSocket live quotes guide.