Quant coding cheatsheet

import numpy as np
import pandas as pd

df = pd.read_parquet("prices.parquet")
df["date"] = pd.to_datetime(df["date"], utc=True)
df = df.sort_values(["asset", "date"])
df.info()  # date, asset, close, volume

One row per date / asset. Use adjusted close prices and keep missing observations explicit.

cols = ["date", "asset", "close"]
liquid = df.loc[
    (df["volume"] > 0) & df["close"].notna(), cols
].copy()
df.loc[df["close"] <= 0, "close"] = np.nan
missing = df.isna().sum()
latest = df.groupby("asset").tail(1)

Use .loc for assignment; parentheses matter when combining boolean conditions.

g = df.groupby("asset", sort=False)
summary = g.agg(
    last_close=("close", "last"),
    avg_volume=("volume", "mean"),
)
df["prev_close"] = g["close"].shift(1)
df["mean_close"] = g["close"].transform("mean")

agg reduces rows; transform returns a result aligned to the original rows.

px = df.pivot(
    index="date", columns="asset", values="close"
).sort_index()
ret = px.pct_change(fill_method=None)
log_ret = np.log(px).diff()
month_end = px.resample("ME").last()
momentum = px.pct_change(20, fill_method=None)

px and ret use dates as rows and assets as columns. Returns are fractions: 0.01 = 1%.

window = px.rolling(20, min_periods=20)
mean = window.mean()
std = window.std(ddof=1).replace(0, np.nan)
z = (px - mean) / std
vol = ret.rolling(63, min_periods=63).std()
lagged_z = z.shift(1)
rank = momentum.rank(axis=1, pct=True)

Windows below count observations. Shift features before using them to predict the next return.

matched = pd.merge_asof(
    trades.sort_values("time"),
    quotes.sort_values("time"),
    on="time", by="asset",
    direction="backward",
    tolerance=pd.Timedelta("5min"),
)

For trades and quotes with time / asset columns, sort globally by time. Use publication timestamps for fundamentals.

valid = np.isfinite(X).all(axis=1) & np.isfinite(y)
A = np.column_stack([np.ones(valid.sum()), X[valid]])
beta, *_ = np.linalg.lstsq(A, y[valid], rcond=None)
fitted = A @ beta
residual = y[valid] - fitted
# @ = matrix product; * = elementwise product.
# Prefer vectorized operations to Python row loops.

X is an n × k feature array; y is length n. Remove nonfinite rows and fit on training data only.

raw = (momentum > 0).astype(float)
w = raw.div(raw.sum(axis=1).replace(0, np.nan), axis=0)
held = w.shift(1).fillna(0)  # previous signal
observed = ret.notna() | held.eq(0)
pnl = (held * ret).sum(axis=1)
pnl = pnl.where(observed.all(axis=1)).iloc[1:]
equity = (1 + pnl).cumprod(skipna=False)

Illustrative daily close-to-close calculation with prior-close execution assumed. Model execution timing, costs, and weight drift separately.

ann_vol = pnl.std(ddof=1) * np.sqrt(252)
# Include initial capital when computing the peak.
peak = equity.cummax().clip(lower=1)
drawdown = equity / peak - 1
max_drawdown = drawdown.min()
corr = ret.corr(min_periods=60)
cov = ret.cov(min_periods=60)

252 assumes daily trading observations; choose the factor for your market. Do not silently drop missing portfolio returns.

assert not df.duplicated(["date", "asset"]).any()
assert px.index.is_monotonic_increasing
assert px.index.is_unique
assert not np.isinf(ret.to_numpy()).any()
assert pnl.notna().all(), "Investigate missing returns"
rng = np.random.default_rng(42)
df.to_parquet("clean_prices.parquet", index=False)

Split train / validation / test chronologically. Fit scalers on training data and include delisted assets in historical universes.