Think in columns and processes — so a strategy search takes minutes instead of hours, and your data fits before you rent a cluster.
ProcessPoolExecutor and collect one metric per configuration.on_tick does no heavy compute.Weeks 4 and 6 gave you a backtester and fast signals. One run, one symbol set, a DataFrame that fits in RAM. That works — until it doesn't.
Two things break at the same time. Feature pipelines explode in size: rolling features across ten symbols × months of ticks × a dozen transforms quickly exceeds a single pandas frame held in memory. And a real search is thousands of runs: tuning a lookback or a threshold means re-running the backtest across a grid — sequentially that is hours, in parallel it is minutes.
So this week has exactly two skills: store and stream data efficiently, and spread compute across cores and machines. Both are prerequisites for Week 8's machine learning, where training is only feasible if your data is already cheap to read and your compute already fans out.
A DataFrame lives in RAM. pandas loads the whole frame plus overhead — often 2–10× the
file size — and once your working set approaches physical memory, everything falls off a cliff. The OS
starts swapping to disk, which is roughly 1000× slower. The symptoms are familiar: a
MemoryError, a machine grinding to a halt, a notebook that never returns. The dataset didn't
change. Your access pattern did.
Downcast float64→float32, object→category,
and write columnar Parquet instead of CSV text. Often this alone removes the problem.
If it won't fit, don't load it all. Stream bounded pieces and reduce as you go: constant memory, any file size.
Spread the work over processes, then over machines. Last resort, not first — and only after you have measured.
Measure first. df.info(memory_usage='deep') and df.memory_usage() tell you
what you actually hold. The widget below is that calculation, made draggable: pick a shape, pick dtypes, and
watch the bar cross the RAM line.
object column does not store 8 bytes per row — it stores an
8-byte pointer plus a Python string object of roughly 60 bytes. That single choice is usually the
whole 2–10× overhead. category replaces it with a 1-byte code and one shared
dictionary. Parquet sizes assume zstd at per-dtype ratios; a real file will differ, the ordering will not.How bytes are laid out on disk decides how many of them you must read, and how well they compress. CSV is row-oriented text: no types, no column pruning, huge on disk, and re-parsed from scratch every time. Parquet is columnar, typed and compressed, so you read only the columns and row groups you actually touch. Columns of similar values compress 5–20× under zstd or snappy.
The rule is simple: CSV to exchange with humans, Parquet for anything you re-read. You will feel the difference in Widget 4.
# each line runs now, single-core, in RAM df = pd.read_parquet('ticks.parquet') out = (df[df.symbol == 'AAPL'] .assign(ret=df.close.pct_change()) .groupby('symbol').ret.std())
# builds a plan, optimises it, runs multi-core out = (pl.scan_parquet('ticks.parquet') .filter(pl.col('symbol') == 'AAPL') .select(pl.col('close').pct_change()) .collect()) # the plan runs HERE
The course stack ships numpy + pandas; Polars is the big-data option, not a requirement. pandas is eager, single-core and in-RAM — mature, universal, and fine below roughly 1–5 GB. Polars builds a query plan, optimises it, runs multi-core and can stream larger-than-RAM. Pick pandas for interop and small-to-medium data; pick Polars for wide scans, many gigabytes, and speed-critical pipelines.
Three levers, in this order. Most people reach for the third first and wonder why a cluster didn't help.
Lever one: stop looping in Python. A Python-level for over a price series pays
interpreter overhead on every element — boxing, dispatch, bounds checks. numpy pushes the same arithmetic
into a typed C loop over contiguous memory and pays it once. df.apply() sits awkwardly in the
middle: it is a Python call per row wearing a vectorised costume.
forinterpreted, per elementdf.apply()a Python call per row# 1 · Python loop — interpreter per element out = [] for i in range(1, len(px)): out.append(px[i] / px[i-1] - 1) # 2 · df.apply — still a Python call per row df['ret'] = df.close.apply(f) # 3 · vectorised — one C loop, no Python ret = px[1:] / px[:-1] - 1
apply, and 0.9 ns plus a fixed 12 µs of call overhead for numpy. Those
constants are illustrative but the shape is real: below a few thousand elements numpy's overhead dominates
and the loop is fine; above a million, the loop is a coffee break and above ten million it is a lunch break.Lever two: bound your memory, not your file size. If it won't fit, don't load it all. Read a stream of chunks, reduce each one, and combine — constant memory for any file size. This is a local map–reduce, and it is the same shape you will later run across processes and machines.
# bounded pieces, streaming reduce → O(1) memory total, n = 0.0, 0 for chunk in pd.read_csv('huge.csv', chunksize=1_000_000): total += chunk['volume'].sum() # MAP: reduce per chunk n += len(chunk) print(total / n) # REDUCE: never held the full file in RAM
Lever three: use processes, not threads. The GIL lets only one thread execute Python bytecode at a
time, so for CPU-bound work threads buy you nothing — they take turns. Separate processes have separate
interpreters and separate GILs, which is real parallelism. ProcessPoolExecutor is the clean stdlib
API; joblib's Parallel/delayed is the same idea in one line. The
if __name__ == '__main__': guard is mandatory, not stylistic: without it each spawned child
re-imports your module and re-launches the pool.
Map–reduce is the mental model behind all of it. Map applies the same function to each independent piece — per symbol, per chunk, per parameter. Reduce combines the partials: a sum, a concat, a best-Sharpe. The shard key is the split, and you choose it so the pieces are genuinely independent. Backtests are the poster child: no run needs another, so speedup is near-linear. A stateful order book is the counter-example — it cannot be sharded blindly, because the dependencies force coordination.
# scripts/param_sweep.py — one backtest per process from concurrent.futures import ProcessPoolExecutor from tests.sim_session import SimSession def backtest(lookback): fn = make_momentum(lookback) # one signal_fn per config res = SimSession().run(n_ticks=2000, strategies=[('m', fn)], symbols=['AAPL', 'TSLA'], seed=7, verbose=False) return lookback, sharpe(res.traders[0].pnl_curve) if __name__ == '__main__': # mandatory for multiprocessing with ProcessPoolExecutor() as pool: for lb, sr in pool.map(backtest, [5, 10, 20, 40, 80]): print(f'lookback={lb:>3} Sharpe={sr:6.2f}')
And make it safe to re-run. Distributed jobs retry, reorder and run out of process, so correctness
demands determinism. A task is idempotent when running it twice yields the same result and no
double-writes — key each output by a hash of its inputs. Pin the seed (SimSession(seed=7)),
pin data versions, pin dtypes. Keep workers pure: inputs in, output out, no shared mutable state and no
hidden globals. Dask and Ray extend exactly these ideas across many machines when one box is no longer enough.
A single backtest answers one question. Research is a thousand questions — and the team that can ask a thousand cheaply finds the edge that the team asking ten never sees.
Concretely: a month of arena ticks across ten symbols with a dozen features is tens of gigabytes as CSV and a couple as Parquet. If every sweep re-parses the CSV, each configuration costs minutes of pure I/O before any strategy logic runs. Cache the features to Parquet once and each configuration reads in seconds. That is the difference between a grid of five lookbacks and a grid of five hundred.
The same logic runs in the live arena. AlgoArena is not one exchange — each Exchange team runs its own
WebSocket venue, and EXCHANGE_URLS=ws://host:8765,ws://host:8766 lets one bot span them all
(broker/config.py, trader/arb_trader.py), while the teacher dashboard aggregates flow
and P&L across venues into a single view. Meanwhile the headless sim is pure, in-process and network-free,
which is precisely why it fans out safely: one process per symbol set or per grid point, each with its seed
pinned.
WHERE clause saves it nothing
— it still parses every byte. Columnar storage reads the intersection of the columns you named and the
row groups that could possibly match, then decompresses only those.Scaling up before scaling out. A bigger box is simple — more RAM, more cores, no code change
— but there is a ceiling and the largest machines cost superlinearly. More boxes are near-unlimited, but
you must partition the work and coordinate the results. Start vertical: more RAM plus smaller dtypes plus
Parquet very often removes the problem outright. Reach for a cluster only when one machine genuinely cannot,
and in this course that means ProcessPoolExecutor across the cores you already own.
# build features offline, cache once, read cheaply forever PATH = 'cache/features.parquet' def features(): if os.path.exists(PATH): return pd.read_parquet(PATH) # seconds df = build_features() # minutes df.to_parquet(PATH) # pay it once return df
The fastest computation is the one you don't repeat. Materialise expensive stages once and every
sweep afterwards reads the cached frame; functools.lru_cache covers the pure, repeated calls
in-process. This cached frame is exactly the history that your
signal_fn(symbol, prices, history, book, portfolio) consumes — built offline, so
on_tick does no heavy compute in the loop.
Every one of these has been submitted by a previous cohort, usually with the words "but it worked on the sample".
You swap the loop for a ThreadPoolExecutor, run sixteen backtests, and the wall clock
barely moves — sometimes it gets worse. The GIL lets only one thread execute Python bytecode at a
time, so your sixteen threads are taking turns on one core while paying context-switch and lock overhead
for the privilege. Threads are for I/O-bound waiting, which is why the arena's WebSocket code is
async and single-threaded.
The fix. CPU-bound work goes in ProcessPoolExecutor —
separate interpreters, separate GILs, real cores. Toggle "threads, not processes" in Widget 3 to watch the
lanes serialise.
if __name__ == '__main__': guardWithout the guard, every spawned child re-imports your module top to bottom — which re-creates
the pool, which spawns more children. On macOS and Windows, where the default start method is
spawn rather than fork, this is an immediate fork bomb or a
RuntimeError about starting a process before the current one finished bootstrapping.
The fix. All pool creation and all top-level side effects live under the guard. Anything a worker needs must be importable — module-level functions, not closures or lambdas, because the arguments and the callable are pickled to cross the process boundary.
You pass the full 3 GB DataFrame as an argument to each of eight workers. Python serialises it eight times, copies it across eight pipes, and deserialises it eight times — and your "parallel" job is slower than the loop it replaced, while peak memory is now nine copies of the frame.
The fix. Send references, not payloads. Pass a symbol, a file path or a
parameter, and let the worker read its own shard from Parquet. That is why the shard key in Widget 3 is a
chunk boundary and why the sweep passes a lookback integer, not a dataset.
A worker appends its partial result to a shared CSV. The task times out, the framework retries it, and now those rows are in the file twice. Your Sharpe is computed over a dataset that silently contains duplicated trades, and nothing crashes — which is the worst kind of bug, because you will believe the number.
The fix. Key each output by a hash of its inputs and write to that path: re-running overwrites rather than appends. Pin the seed so a retry reproduces the same numbers, and make the reduce step tolerate seeing a partial twice.
The frame doesn't fit, so the instinct is a distributed engine. But the frame is
float64 everywhere and the symbol column is object — roughly 60 bytes per
row for a ticker that has ten possible values. Half the memory is Python string objects and the other
half is precision you do not have: your prices are quoted to the cent, not to sixteen significant digits.
The fix. Measure with df.info(memory_usage='deep'), downcast to
float32 and category, store Parquet, and re-measure. Widget 1 shows a typical
frame dropping back under the RAM line from those two changes alone — no cluster, no new code.
Instant feedback, nothing recorded. Pick the best answer.
ThreadPoolExecutor on a
16-core machine and is no faster than a plain loop. What is the explanation?pd.read_csv(path) with a loop over
pd.read_csv(path, chunksize=1_000_000) that sums one column. What have you actually bought?Deliverable: a data-architecture sign-off — your storage and parallelism plan —
plus a working parallel sweep. Graded 35 / 30 / 20 / 15: the sign-off covering columnar storage and caching, a
parallel backtest sweep that runs correctly and reproducibly with the seed pinned, a feature pipeline that
produces the history your on_tick consumes, and multi-venue connect via
EXCHANGE_URLS on a clean W7 branch.
make install.on_tick without blocking the loop.Reading before next class: tests/sim_session.py — the
run() API you fan out, and the seed= argument that makes every worker reproducible.
This week is the foundation for Week 8: training is only feasible if your data is already cheap to read and
your compute already spreads.