10Dataframes

Two researchers compute five-minute bars from the same trades and get different volumes. One library names a bar by its left edge and puts a trade stamped exactly on a boundary into the bar that starts there; the other names it by its right edge and puts the same trade into the bar that ends there; one of them drops the minutes in which nothing traded, the other does not. Both are right by their own conventions, and a signal that compares the two sets of bars finds a pattern that is nothing but the convention. The dataframe is the research platform’s working language; its idioms — bars, as-of joins, rolling windows, cross-sectional ranks, pivots — are where its silent bugs live, and its engines differ enormously in what they cost. This chapter writes the idioms once with explicit conventions for three engines (firm.marketdf), checks that the engines agree, and runs one research pipeline over a month of quotes and trades eagerly, lazily and out of core.

10.1 The dataframe model

The three engines of the chapter implement the same model differently. pandas holds each column as a numpy array (or an Arrow array) with a row index attached, and executes each operation when it is called. Polars holds columns in Arrow’s memory layout, has no row index, and can either execute each operation or build a plan and execute it later. DuckDB is a SQL engine: a query is a plan, executed when its result is asked for (chapter 5). The columnar layout of chapter 3 underlies all three.

Market data are stored long (chapter 4’s partitions are long tables), and most transformations are clearer long: a rolling window is a window over the rows of one symbol, a cross-sectional rank a rank over the rows of one minute. The wide format serves the matrix operations of portfolio construction (One Quant Book 7, chapters 24–26), and the pivot between the two is where missing values appear: a minute in which one symbol did not trade is a missing cell in the wide table, never a row in the long one.

10.2 Eager and lazy evaluation

Eager evaluation is easy to reason about and to debug step by step; every intermediate table exists, and every intermediate table costs memory. Lazy evaluation gives the engine the whole pipeline, which is what makes optimisation possible.

10.3 Query plans and optimisation

A lazy query for one symbol’s prices on one day, written as a scan of every trade file followed by a filter and a selection, is optimised by Polars into a plan that reads one partition and four of six columns, and applies the filter while scanning (Figure 10.2).

The partition pruning (only date=5 is scanned), the projection pushdown (four columns: the two asked for and the two the filter needs) and the predicate pushdown of chapter 3 are all the optimiser’s work; an eager version of the same code would read the month and filter it in memory.

10.4 Out-of-core and streaming execution

Chapter 8 wrote out-of-core loops by hand; a lazy engine with a streaming executor does the same for a whole plan. It cannot do it for every operator: an as-of join needs both sides sorted, and a sort needs all its input, so a plan containing them buffers at those points. The pipeline below has both, which is why streaming saves memory here but does not make it constant.

10.5 Idioms for market data

Five idioms cover most of research on market data, and each has a convention that must be stated rather than inherited.

def bars(ticks: pd.DataFrame, width: int, label: str = "left", closed: str = "left",
         fill: bool = False, engine: str = "pandas") -> pd.DataFrame:
    shift = width if label == "right" else 0
    if engine == "pandas":
        t = ticks.sort_values(["symbol", "ts", "seq"], kind="stable")
        k = _bucket_pd(t["ts"], width, closed)
        g = t.assign(bar=k * width + shift).groupby(["symbol", "bar"], sort=True)
        b = g.agg(open=("price", "first"), high=("price", "max"), low=("price", "min"),
                  close=("price", "last"), volume=("qty", "sum"), trades=("qty", "size"))
        b = b.reset_index()
    elif engine == "polars":
        c = pl.col
        k = (c("ts") // width) if closed == "left" else ((c("ts") - 1) // width)
        b = (pl.from_pandas(ticks).lazy().sort(["symbol", "ts", "seq"], maintain_order=True)
             .with_columns((k * width + shift).alias("bar"))
             .group_by(["symbol", "bar"], maintain_order=True)
             .agg(c("price").first().alias("open"), c("price").max().alias("high"),
                  c("price").min().alias("low"), c("price").last().alias("close"),
                  c("qty").sum().alias("volume"), pl.len().alias("trades"))
             .sort(["symbol", "bar"]).collect().to_pandas())
    elif engine == "duckdb":
        x = "ts" if closed == "left" else "(ts - 1)"   # floor division, whatever `//` does
        k = f"(({x} - ((({x}) % {width}) + {width}) % {width}) // {width})"
        con = duckdb.connect()
        con.register("ticks", ticks)
        b = con.execute(f"""
            SELECT symbol, {k} * {width} + {shift} AS bar,
                   first(price ORDER BY ts, seq) AS open, max(price) AS high,
                   min(price) AS low, last(price ORDER BY ts, seq) AS close,
                   sum(qty) AS volume, count(*) AS trades
            FROM ticks GROUP BY ALL ORDER BY symbol, bar""").df()
    else:
        raise ValueError(engine)
    b = b[COLS].astype({c: "int64" for c in COLS[1:]}).reset_index(drop=True)
    return _fill(b, width) if fill else b

Listing 10.1 is the bars idiom in the three engines. Two details are the kind that breaks equivalence silently: the first and last price of a bar must be taken in time order with a tie-breaker (the feed’s sequence number), and the index of a right-closed bar is $\lceil t/w \rceil - 1 = \lfloor (t-1)/w \rfloor$ for integer times, which each engine must compute with floor division — DuckDB’s integer division truncates toward zero, and the SQL version computes the floor explicitly. The acceptance tests compare the three engines on trades stamped to the whole second, so that boundaries and ties occur.

10.6 Joins, windows and ranks: where ties hide

Bars are the most visible convention; three others hide in the joins and windows. An as-of join on the timestamp matches each trade with the last quote at or before its time: when a trade and the quote change it caused share a timestamp, the inclusive join returns the quote after the trade. Chapter 4 found this on 223 of 17 281 trades and made the join strict on the feed’s sequence number; firm_marketdf.asof takes strict as an argument in all three engines (pandas allow_exact_matches, Polars the same, DuckDB the inequality > instead of >=).

A rolling window has a frame in rows or in time, and a minimum number of observations below which it returns nothing or a partial value: the idiom’s rolling mean averages whatever rows exist, one at the start of each symbol, which must be known to the code that uses it. A cross-sectional rank has a tie rule, and engines differ in their default.

10.7 One pipeline, five ways

The chapter’s research pipeline runs over the chapter-3 month of quotes (now 3 million, three thousand per symbol per day) and a month of 1.2 million trades, stored as date partitions of Parquet: one-minute bars, their log returns, the cross-sectional rank of each return within its minute, each symbol’s average rank over the month; and, in parallel, every trade joined as of its prevailing quote and each symbol’s mean effective spread in basis points. Listing 10.2 is the Polars version, a lazy plan that the three Polars modes share.

def _polars_plan(t: pl.LazyFrame, q: pl.LazyFrame):
    c = pl.col
    b = (t.with_columns((c("ts") // MIN).alias("bar")).sort(["symbol", "ts", "seq"])
         .group_by(["symbol", "bar"]).agg(c("price").last().alias("close"))
         .sort(["symbol", "bar"])
         .with_columns(c("close").log().diff().over("symbol").alias("ret")).drop_nulls("ret")
         .with_columns(c("ret").rank("average").over("bar").alias("rank")))
    ranks = b.group_by("symbol").agg(c("rank").mean().alias("avg_rank"))
    quote = q.select(["symbol", "ts", "bid", "ask"]).sort("ts")
    j = (t.sort("ts").join_asof(quote, on="ts", by="symbol", check_sortedness=False)
         .with_columns(((c("bid") + c("ask")) / 2.0).alias("mid"))
         .with_columns((2e4 * (c("price") - c("mid")).abs() / c("mid")).alias("spread_bp")))
    spreads = j.group_by("symbol").agg(c("spread_bp").mean())
    return ranks, spreads

def pipeline_duckdb(root: pathlib.Path) -> dict:
    con = duckdb.connect()
    con.execute("SET threads = 1")
    for name in ("trades", "quotes"):
        con.execute(f"CREATE VIEW {name[0]} AS SELECT * FROM "
                    f"read_parquet('{root}/{name}/*/*.parquet', hive_partitioning = true)")
    ranks = con.execute(f"""
        WITH b AS (SELECT symbol, ts // {MIN} AS bar, last(price ORDER BY ts, seq) AS close
                   FROM t GROUP BY ALL),
             r AS (SELECT symbol, bar, ln(close) - lag(ln(close))
                          OVER (PARTITION BY symbol ORDER BY bar) AS ret FROM b),
             k AS (SELECT symbol, bar,
                          (2 * rank() OVER w + count(*) OVER (PARTITION BY bar, ret) - 1)
                          / 2.0 AS rank
                   FROM r WHERE ret IS NOT NULL WINDOW w AS (PARTITION BY bar ORDER BY ret))
        SELECT symbol, avg(rank) AS avg_rank FROM k GROUP BY symbol""").df()
    spreads = con.execute("""
        WITH j AS (SELECT t.symbol, t.price, (q.bid + q.ask) / 2 AS mid
                   FROM t ASOF LEFT JOIN q ON t.symbol = q.symbol AND t.ts >= q.ts)
        SELECT symbol, avg(2e4 * abs(price - mid) / mid) AS spread_bp
        FROM j GROUP BY symbol""").df()
    return _finish(ranks, spreads)

All five modes return the same fifty average ranks and fifty mean spreads to nine decimals. Figure 10.4 measures them, each in its own process with one thread, memory counted as the process’s high-water mark minus that of a process that only imports the libraries (155 MB). pandas, eager and with a row index, reads the whole month into memory and holds every intermediate table: it needs 895 MB more than the imports and 1.3 s. Polars eager holds the same tables in Arrow form: 769 MB. Lazily, it needs 654 MB, and with the streaming engine 614 MB; the as-of join and the sorts keep the saving modest. DuckDB, executing the SQL over the files with its own streaming operators, needs 317 MB, a third of pandas.

10.8 Tutorial: bars, conventions and five engines

10.9 Build: market-data idioms

10.10 Exercises

10.11 Problem: Two Sets of Bars

10.12 Interview questions