Crypto tick data in Python: record a reference-price stream

Build a small dataset of the BTCUSD and ETHUSD reference-price updates delivered to your Python client. This guide records a bounded WebSocket session to Parquet, keeps both source and receive timestamps, and reads the result back without pretending the recording is a complete market tape. You can get a free API key before starting; the example uses one connection and two symbols.

The output is useful for comparing price movements, examining observed spreads and testing how your application handles updates. It is a forward recording from the time you start the script, not a download of historical exchange trades.

What kind of crypto tick data will you record?#

SiftingIO's Crypto Data API provides aggregated reference prices for supported USD-quoted pairs. Those values are not individual exchange trades or executable liquidity. The data methodology explains what the reference price represents.

That distinction matters when naming your dataset. A minute OHLCV bar summarizes a period; it does not preserve the sequence of updates your client saw. Recording the stream gives you that observed sequence, but not venue-specific trade IDs, order flow or a guarantee that every published update arrived. Use it for reference-price research and application testing, not to infer fills at a particular exchange.

Here is an illustrative frame, not a measured market observation. The protocol and field definitions are in the WebSocket documentation.

{ "f": "tick", "class": "cex", "s": "BTCUSD", "p": 64210.5, "P": 0.42, "b": 64209.0, "B": 1.8, "a": 64212.0, "A": 2.1, "t": 1790344800000 }

s identifies the symbol, t is the API timestamp in Unix milliseconds, and p/P, b/B and a/A carry price/size fields. The recorder also adds recv_ms, taken from the recording machine. Subtracting these clocks does not isolate network latency.

Before you start#

Use Python 3.12. This example was checked locally with websockets 13.1, pyarrow 17.0.0 and pandas 2.2.3, using simulated WebSocket frames and real Parquet files. That verifies the example's behaviour, not live-feed availability or performance.

The Free plan currently permits one connection and five symbol subscriptions, so BTCUSD and ETHUSD fit within it. Close another client using the same key before running this example. Your authentication acknowledgement reports the limits attached to your key.

Three protocol details shape the recorder:

• Subscribe with product cex, not the REST asset name crypto.
• The first tick can be a cached snapshot, with the same shape as a live update. The code keeps it as first_observed, rather than claiming it knows which kind it is.
• Send the application-level ping regularly. A pong confirms transport activity; it does not prove a symbol is still updating.

For connection basics, see REST snapshots and WebSocket streams for FX and crypto. The focus here is saving and inspecting a recording.

Record one bounded session#

Save the following program as record_ticks.py. It runs for one hour by default. Each run gets a separate directory under ticks/, containing numbered Parquet parts and a session.jsonl log.

In Bash, install the dependencies and enter your key without placing its value in shell history:

python -m pip install "websockets==13.1" "pyarrow==17.0.0" "pandas==2.2.3"
read -rs SIFTING_KEY && export SIFTING_KEY
DURATION_S=3600 python record_ticks.py

There is no automatic reconnect. A disconnect ends this recording, attempts a final flush and exits with a nonzero status. Start another run deliberately if you want another session.

#!/usr/bin/env python3
"""Record SiftingIO crypto reference-price ticks to Parquet for a fixed window.

Python 3.12, websockets 13, pyarrow 17.
Environment: SIFTING_KEY (required), DURATION_S (default 3600), TICK_DIR (default ticks).
Exit status: 0 when the window ran to its end, 1 on any other outcome, 130 on Ctrl-C.
"""
import asyncio
import json
import math
import os
import re
import sys
import time

import pyarrow as pa
import pyarrow.parquet as pq
from websockets.asyncio.client import connect

KEY = os.environ["SIFTING_KEY"]
URL = f"wss://stream.sifting.io/ws/v1?key={KEY}"
SYMBOLS = ["BTCUSD", "ETHUSD"]
OUT = os.environ.get("TICK_DIR", "ticks")
DURATION = int(os.environ.get("DURATION_S", "3600"))  # capture window in seconds
FLUSH_EVERY = 60    # seconds between part files; rows younger than this exist only in memory
PING_EVERY = 30     # docs: a client frame at least every 60 s, idle close at 90 s
RECV_TIMEOUT = 120  # no frame of any kind for this long: treat the socket as dead
FATAL = {"bad_op", "auth_required", "auth_timeout", "auth_failed",
         "unknown_product", "max_connections", "max_subscriptions"}
KEY_RE = re.compile(r"sft_[A-Za-z0-9_-]+")

SCHEMA = pa.schema([
    ("s", pa.string()),
    ("t", pa.int64()),               # timestamp the API attached to the value, Unix epoch ms
    ("recv_ms", pa.int64()),         # this machine's clock when the frame arrived
    ("first_observed", pa.bool_()),  # True for the first tick of a symbol on this connection
    ("p", pa.float64()), ("P", pa.float64()),
    ("b", pa.float64()), ("B", pa.float64()),
    ("a", pa.float64()), ("A", pa.float64()),
])
PRICE_FIELDS = ["p", "P", "b", "B", "a", "A"]


class FatalError(Exception):
    """A server error that a retry would only repeat."""


def now_ms():
    return int(time.time() * 1000)


def redact(text):
    """Mask the configured key and anything shaped like a key before any text is logged or printed."""
    return KEY_RE.sub("sft_***", str(text).replace(KEY, "sft_***"))


def note(session_dir, kind, **fields):
    with open(os.path.join(session_dir, "session.jsonl"), "a") as f:
        f.write(json.dumps({"kind": kind, "at_ms": now_ms(), **fields}) + "\n")


def to_float(v):
    """None stays None; ints, floats and numeric strings become a finite float; anything else is rejected."""
    if v is None:
        return None
    if isinstance(v, bool):
        raise ValueError("bool is not a price")
    x = float(v)
    if not math.isfinite(x):
        raise ValueError("non-finite value")
    return x


def make_row(frame, recv_ms, first_observed):
    """Build one complete, validated row from a tick frame, or return None if any field is unusable."""
    try:
        s = frame["s"]
        t = frame["t"]
        if not isinstance(s, str) or not s or isinstance(t, bool):
            return None
        t = int(t)
        if t <= 0:
            return None
        row = {"s": s, "t": t, "recv_ms": recv_ms, "first_observed": first_observed}
        for k in PRICE_FIELDS:
            row[k] = to_float(frame.get(k))
        return row
    except (KeyError, TypeError, ValueError, OverflowError):
        return None


class Recorder:
    def __init__(self, session_dir):
        self.dir = session_dir
        self.rows = []
        self.written = 0
        self.rejected = 0
        self.parts = 0

    def add(self, row):
        self.rows.append(row)

    def flush(self):
        """Write buffered rows to a new part file. The buffer is cleared only after the rename succeeds."""
        if not self.rows:
            return
        n = self.parts + 1
        final = os.path.join(self.dir, f"part-{n:05d}.parquet")
        tmp = final + ".tmp"
        pq.write_table(pa.Table.from_pylist(self.rows, schema=SCHEMA), tmp, compression="zstd")
        os.replace(tmp, final)
        self.parts = n
        self.written += len(self.rows)
        self.rows = []


async def capture(rec):
    """One bounded session on one connection. Returns the reason it ended; raises FatalError for config faults."""
    deadline = time.monotonic() + DURATION
    async with connect(URL, ping_interval=None) as ws:
        await ws.send(json.dumps({"op": "subscribe", "product": "cex", "symbols": SYMBOLS}))
        seen = set()
        started = time.monotonic()
        next_ping = started + PING_EVERY
        next_flush = started + FLUSH_EVERY
        last_frame = started
        while True:
            now = time.monotonic()
            if now >= deadline:
                return "window elapsed"
            if now - last_frame >= RECV_TIMEOUT:
                return f"no frame of any kind for {RECV_TIMEOUT} s"
            if now >= next_ping:
                await ws.send(json.dumps({"op": "ping"}))
                next_ping = now + PING_EVERY
            if now >= next_flush:
                rec.flush()  # a disk error raises here and ends the session with the buffer intact
                next_flush = now + FLUSH_EVERY
            wait = min(deadline, next_ping, next_flush, last_frame + RECV_TIMEOUT) - now
            try:
                raw = await asyncio.wait_for(ws.recv(), timeout=max(wait, 0.05))
            except TimeoutError:
                continue
            recv_ms = now_ms()
            last_frame = time.monotonic()
            try:
                frame = json.loads(raw)
            except ValueError:
                rec.rejected += 1
                continue
            if not isinstance(frame, dict):
                rec.rejected += 1
                continue
            kind = frame.get("f")
            if kind == "tick":
                sym = frame.get("s")
                row = make_row(frame, recv_ms, first_observed=sym not in seen)
                if row is None:
                    rec.rejected += 1
                    continue
                seen.add(sym)
                rec.add(row)
            elif kind == "ack":
                limits = {k: frame.get(k) for k in ("tier", "max_conn", "max_subs")}
                print("ack:", redact(json.dumps(limits)), flush=True)
            elif kind == "error":
                code = redact(frame.get("code"))[:40]
                if code in FATAL:
                    raise FatalError(f"server error {code}")
                return f"server error {code}"
            # pong and anything unrecognised are ignored


def new_session_dir():
    stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
    path = os.path.join(OUT, f"{stamp}-{os.getpid()}")
    os.makedirs(path, exist_ok=False)
    return path


async def main():
    session_dir = new_session_dir()
    rec = Recorder(session_dir)
    note(session_dir, "start", symbols=SYMBOLS, planned_s=DURATION)
    reason, ok = "interrupted", False
    try:
        reason = await capture(rec)
        ok = reason == "window elapsed"
    except FatalError as e:
        reason = f"fatal: {redact(e)}"
    except Exception as e:
        reason = f"{type(e).__name__}: {redact(e)}"
    finally:
        unflushed = 0
        try:
            rec.flush()
        except Exception as e:
            unflushed = len(rec.rows)
            reason = f"{reason}; final flush failed: {type(e).__name__}: {redact(e)}"
            ok = False
        note(session_dir, "end", reason=reason, rows_written=rec.written,
             rows_unflushed=unflushed, frames_rejected=rec.rejected, parts=rec.parts)
        print(f"ended: {reason}; rows written {rec.written}; unflushed {unflushed}; "
              f"rejected {rec.rejected}; parts {rec.parts}", flush=True)
    return 0 if ok else 1


if __name__ == "__main__":
    try:
        sys.exit(asyncio.run(main()))
    except KeyboardInterrupt:
        sys.exit(130)

The receive loop also handles pings and timed writes. A write failure therefore reaches the main error handler instead of silently stopping a background task. The buffer is cleared only after the part file has been written and renamed; if that fails, the final flush gets one retry.

Rows with invalid numeric fields are rejected before they can partly enter the buffer. The log counts rejected frames, missing fields remain null, and both JSON numbers and numeric strings are accepted. No rows are deduplicated automatically.

Read the files without overstating coverage#

Save this as read_ticks.py, then pass the directory for one recorded session. Replace the example directory with the one created by your run.

#!/usr/bin/env python3
"""Report what one recorded session holds.

python read_ticks.py ticks/20260925T140000Z-12345
"""
import glob
import os
import sys

import pandas as pd

session = sys.argv[1]

log_path = os.path.join(session, "session.jsonl")
if os.path.exists(log_path):
    print(pd.read_json(log_path, lines=True).to_string())
else:
    print("no session.jsonl in", session)

parts = sorted(glob.glob(os.path.join(session, "part-*.parquet")))
if not parts:
    print("no part files in", session)
    sys.exit(0)

ticks = pd.concat([pd.read_parquet(p) for p in parts], ignore_index=True)
ticks = ticks.sort_values(["s", "recv_ms", "t"], kind="stable").reset_index(drop=True)
print("rows:", len(ticks), "parts:", len(parts))

by_sym = ticks.groupby("s")
report = by_sym.agg(rows=("t", "size"), first_t=("t", "min"), last_t=("t", "max"),
                    first_observed_rows=("first_observed", "sum"))
report["span_s"] = (report["last_t"] - report["first_t"]) / 1000
report["longest_receive_interval_s"] = by_sym["recv_ms"].diff().groupby(ticks["s"]).max() / 1000
report["longest_t_gap_s"] = by_sym["t"].diff().groupby(ticks["s"]).max() / 1000
print(report.to_string())

print(by_sym.head(1)[["s", "t", "recv_ms", "first_observed", "p", "b", "a"]].to_string())

shared = ticks.duplicated(["s", "t"], keep=False)
print(shared.groupby(ticks["s"]).sum().rename("rows_sharing_a_t").to_string())

An empty session prints a message rather than failing because there are no Parquet files. Otherwise, the report shows row counts, timestamp spans, first-observed rows and two distinct interval measurements.

A large interval could reflect an unchanged price, a source pause or a delivery problem. The frame has no sequence number or cause field that lets this recorder distinguish those cases. Likewise, a small interval does not prove nothing was missed.

Keep rows with matching timestamps unless your analysis has a justified deduplication rule. Different observations can share a timestamp; deleting them solely on (s, t) can discard data.

What can still be lost?#

This is a small recording example, not a durable ingestion service.

Rows stay in memory between flushes. A hard stop or power failure can lose unflushed rows and leave no end event in session.jsonl. Renaming a finished file helps avoid reading a partial part, but it is not a guarantee that the storage device has persisted its contents.

If a disk error prevents the final flush, the process reports an unsuccessful outcome and the number of rows still unflushed, provided the session log itself can still be written. That count explains the failed recording; it cannot recover the lost rows later.

Run on a machine that stays awake, watch process exit codes and available disk space, and treat each new session separately. The log records when the program attempted capture and how it finished when it could record an outcome. It does not certify uninterrupted data delivery between those times.

Use the result for a specific question#

Start with a short capture and inspect the first rows before collecting hours of data. Confirm the symbols, the two timestamps and the first-observed flag. Then ask a narrow question: how did the reference spread change, or which price updates did your application actually receive?

The live spread snapshot across crypto, FX and gold offers related context for interpreting reference quotes. Keep your own measured results separate from illustrative examples, and describe the data as reference-price updates received by one client during a recording session.

Ready to try it? Get an API key, then use the WebSocket docs to check the protocol and the Crypto Data API page for supported workflows.