4A Tick Store on Open Formats

At 16:05 the day’s capture files close. By 16:40 they must be compacted, sorted and queryable, because the overnight research starts at 17:00, and a researcher who asks for the quote prevailing before each of this week’s trades must not notice which days come from the files written during the session and which from the compacted history. A week in the life of such a store is this chapter: five simulated days go through capture, an intraday store written as the session runs, a compaction at the close, a historical store of partitioned columnar files, and one query layer over both. On the fourth day the quotes gain a field, and the store has to carry on. Nothing here needs a proprietary database: the store is built from the open formats of chapter 3, an embedded query engine and a dataframe library, with a memory-mapped flat file for the hottest data, read the same way from Python, C++ and Rust (firm.tickstore).

4.1 Live capture: append-only files

During the session the store writes the way data arrives: in time order, as it comes, with nothing sorted, compressed or indexed. The events of chapter 2 — the receive time followed by the feed handler’s 48-byte event — go to a flat file of fixed-width records; the quotes and trades derived from them (the top of book after each change, from Book 13’s book builder, and each execution with its price) go to Arrow IPC streams, the in-memory columnar format of chapter 3 written as a sequence of record batches.

The store’s flat file (Figure 4.1) starts with a magic number, the version, the record width and the layout as JSON, padded to 64 bytes so that every record that follows is aligned the same way; version 1 records are 56 bytes, version 2 appends a venue, flags and padding for 64. Listing 4.1 is the writer and the reader.

class FlatWriter:
    """Append-only fixed-width records behind a header padded to 64 bytes."""

    def __init__(self, path, version: int = 2):
        self.dtype = SCHEMAS[version]
        schema = json.dumps(self.dtype.descr).encode()
        head = HEAD.pack(MAGIC, version, self.dtype.itemsize, len(schema)) + schema
        self.f = open(path, "wb")
        self.f.write(head + b"\0" * (-len(head) % ALIGN))
        self.n = 0

    def append(self, rec: np.ndarray) -> None:
        self.f.write(np.ascontiguousarray(rec, dtype=self.dtype).tobytes())
        self.n += len(rec)

    def close(self) -> None:
        self.f.close()


def flat_open(path):
    """(version, records) with the records memory-mapped read-only: nothing is read until
    it is touched."""
    with open(path, "rb") as f:
        magic, version, size, slen = HEAD.unpack(f.read(HEAD.size))
    if magic != MAGIC or SCHEMAS[version].itemsize != size:
        raise ValueError(f"{path}: not a tick-store flat file of a known version")
    start = HEAD.size + slen
    start += -start % ALIGN
    return version, np.memmap(path, dtype=SCHEMAS[version], mode="r", offset=start)

4.2 End-of-day compaction

The compaction of firm.tickstore (Listing 4.2) sorts each table by symbol and time and writes one Parquet file per partition, under <table>/date=YYYY-MM-DD/symbol=S/: the directory names carry the partition values, a convention the query engines read as columns. It is Method 3.8 of chapter 3 made concrete: partition by date and symbol, sort by time inside.

def compact(tables: dict, root, date: str, row_group_size: int = 65_536) -> dict:
    out = {}
    for name, t in tables.items():
        t = t.sort_by([("symbol", "ascending"), ("ts", "ascending")])
        files, size = 0, 0
        for sym in sorted(set(t["symbol"].to_pylist())):
            part = t.filter(pa.compute.equal(t["symbol"], sym)).drop_columns(["symbol"])
            d = pathlib.Path(root) / name / f"date={date}" / f"symbol={sym}"
            d.mkdir(parents=True, exist_ok=True)
            pq.write_table(part, d / "part.parquet", row_group_size=row_group_size,
                           compression="zstd")
            files += 1
            size += (d / "part.parquet").stat().st_size
        out[name] = {"rows": t.num_rows, "files": files, "bytes": size}
    return out

Figure 4.2 counts the bytes of the simulated week in each form. The busiest day, Monday, has 258 535 events; its intraday files take 17.5 MB (the flat events and the two IPC streams), its compacted partitions 6.15 MB, 2.85 times less, with the events at 20.0 bytes each in their partitions. Compaction of that day took 0.12 s on the laptop: about 2.2 million events a second on one core, sorting, partitioning, encoding and compressing included.

A reader that lists a directory while the compactor writes into it sees half a day; a reader of a manifest sees either the old day or the new one. The verification step is cheap because the intraday files are still there, and it is the last moment at which a lost partition can be rebuilt without going back to the raw capture; chapter 7 turns such checks into the firm’s data-quality rules.

4.3 The intraday and the historical store

The split is invisible to a reader only if one query layer covers both. Store opens an embedded DuckDB and creates one view per table: the historical Parquet partitions, read with their directory names as columns, and the current day’s IPC tables, registered in memory, joined by a union that matches columns by name (Listing 4.3). A query on quotes then covers Monday to Friday, whichever store holds each day.

class Store:
    """DuckDB over the historical Parquet partitions and the day's intraday tables."""

    def __init__(self, root, intraday: dict | None = None, date: str | None = None):
        self.root = pathlib.Path(root)
        self.con = duckdb.connect()
        for name in ("quotes", "trades"):
            hist = self.root / name
            parts = []
            if hist.exists() and any(hist.glob("date=*/symbol=*/*.parquet")):
                files = f"{hist}/date=*/symbol=*/*.parquet"
                parts.append(f"SELECT * FROM read_parquet('{files}', "
                             "hive_partitioning = true, union_by_name = true)")
            if intraday and name in intraday:
                t = intraday[name]
                self.con.register(f"{name}_intraday", t)
                parts.append(f"SELECT *, DATE '{date}' AS date FROM {name}_intraday")
            if parts:
                union = " UNION ALL BY NAME ".join(parts)
                self.con.execute(f"CREATE VIEW {name} AS {union}")

    def sql(self, q: str) -> pa.Table:
        return self.con.execute(q).to_arrow_table()

4.4 As-of joins at scale

The as-of join of One Quant Book 7, chapter 3 — each row of one table with the last row of another known at its time — is the tick store’s most frequent query: each trade with the quote that prevailed when it happened, for effective spreads, trade signs and mark-outs. The store computes it three ways, which must agree row for row: DuckDB’s ASOF JOIN, Polars’ join_asof, and Book 7’s firm.pit.asof_join applied to each day and symbol.

ASOF_SQL = """
SELECT t.date, t.symbol, t.seq, t.ts, t.price, t.qty, q.bid, q.ask
FROM trades t ASOF LEFT JOIN quotes q
  ON t.date = q.date AND t.symbol = q.symbol AND t.{key} {op} q.{key}
ORDER BY t.date, t.symbol, t.seq"""
COLS = ["date", "symbol", "seq", "ts", "price", "qty", "bid", "ask"]


def asof_duckdb(store: Store, key: str = "seq") -> pa.Table:
    """The quote prevailing before each trade: the last quote of an earlier sequence number
    (key='seq'), or, to show the tie problem, the last quote at or before the trade's
    timestamp (key='ts')."""
    return store.sql(ASOF_SQL.format(key=key, op=">" if key == "seq" else ">="))

The first version of the chapter joined on the timestamp, and the three engines disagreed. The reason is in the data: an execution and the change of quote it causes carry the same exchange timestamp, so “the last quote at or before the trade’s time” can be the quote the trade itself produced, and which of two quotes with equal timestamps an engine returns is not specified. The quote that prevailed before the trade is the last one of an earlier sequence number, and the feed’s sequence number totally orders events.

On the week’s 17 281 trades, the three engines agree on every row with the sequence-number join. The timestamp join returns a different quote for 223 of them, 1.3%, each time the quote the execution itself had just produced: exactly the trades whose effective spread a researcher would measure as zero or negative. On this data DuckDB’s join took 16 ms, Polars’ 45 ms, and the Python reference 0.17 s (Figure 4.4).

4.5 Flat files and memory mapping

A fixed-width file and a memory map fit together: record $i$ is at a known address, nothing is parsed, and only the pages a query touches are read. Book 12’s memory-mapped datasets (chapter 23) use the same mechanism for training data. The store’s reader in Python is numpy.memmap; the C++20 reader (Listing 4.5) calls mmap itself, and the Rust twin declares mmap and munmap from the C library, since the series’ Rust uses no external crates. The three compute the same per-instrument summary — records, quantity executed, last sequence number — on the same fixtures of both versions. Scanning one column of Monday’s events took 0.8 ms from the mapped flat file and 2.1 ms from its Parquet partitions: the flat file is more than twice as fast to scan and 2.8 times larger to keep.

class FlatFile {
public:
    explicit FlatFile(const std::string& path) {
        fd_ = ::open(path.c_str(), O_RDONLY);
        if (fd_ < 0) throw std::runtime_error("cannot open " + path);
        struct stat st {};
        ::fstat(fd_, &st);
        size_ = static_cast<std::size_t>(st.st_size);
        void* m = ::mmap(nullptr, size_, PROT_READ, MAP_PRIVATE, fd_, 0);
        if (m == MAP_FAILED) throw std::runtime_error("mmap failed");
        base_ = static_cast<const std::uint8_t*>(m);
        if (size_ < 12 || std::memcmp(base_, "OQTS", 4) != 0)
            throw std::runtime_error("not a flat file");
        version_ = load<std::uint16_t>(base_ + 4);
        record_ = load<std::uint16_t>(base_ + 6);
        std::size_t start = 12 + load<std::uint32_t>(base_ + 8);
        start_ = (start + 63) / 64 * 64;
        bool known = (version_ == 1 && record_ == 56) || (version_ == 2 && record_ == 64);
        if (!known) throw std::runtime_error("unknown version");
        n_ = (size_ - start_) / record_;
    }
    FlatFile(const FlatFile&) = delete;
    FlatFile& operator=(const FlatFile&) = delete;
    ~FlatFile() { ::munmap(const_cast<std::uint8_t*>(base_), size_); ::close(fd_); }

    std::size_t size() const { return n_; }
    int version() const { return version_; }
    Event operator[](std::size_t i) const {   // pages are read from disk when first touched
        const std::uint8_t* p = base_ + start_ + i * record_;
        using U64 = std::uint64_t;
        return {load<U64>(p), load<U64>(p + 12), load<U64>(p + 20), load<U64>(p + 28),
                load<U64>(p + 36), load<U64>(p + 48), load<std::uint32_t>(p + 44),
                load<std::uint16_t>(p + 10), p[8], p[9]};
    }

4.6 Schema evolution

On Thursday the venue added a field to its quotes, and the store’s quotes carry it from that day on. Monday to Wednesday have no such column.

Under that rule a reader of the new version reads an old file by filling the new fields with their defaults, and a reader of the old version reads a new file by ignoring the fields it does not know; the flat file does both by its version number (upgrade), the query layer by matching columns by name (a missing column reads as null). The same week’s window query counts the venue on none of the first three days and on every quote of the last two. The registry’s check (check_evolution) refuses the changes that would break old readers: a field retyped from 32 to 64 bits, a field dropped, two fields swapped, a field inserted before existing ones.

4.7 Tutorial: a week in the store

4.8 Build: the tick store

4.9 Exercises

4.10 Problem: The 16:40 Deadline

4.11 Interview questions