From a vectorized equity backtest to an event-driven one

daily = cu.fetch_daily(cu.SP500_EXAMPLES[:10], start="2020-01-01", end="2026-07-01")
print(f"{daily.num_rows:,} rows x {daily.num_columns} columns")
daily.to_pandas().head()

tape = daily.to_pandas()
tape = tape[tape["ts"] >= pd.Timestamp("2022-01-01", tz="UTC")]
tape = pa.Table.from_pandas(tape, preserve_index=False)
market = cu.make_equity_market(tape, spread_bps=5.0, depth_fraction=0.002)
for name, table in market.items():
    print(f"{name}: {table.num_rows:,} rows x {table.num_columns} columns")
market["book_deltas"].to_pandas().head(4)

db = h5i_db.Database(cu.fresh_db("04_equity_bars_to_event_driven"), create=True)
prices = daily.sort_by([("ts", "ascending"), ("symbol", "ascending")])
db.create_table("prices", prices.schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("prices", prices, note="daily bars, 10 large caps")
for name in ("instruments", "book_deltas", "trades"):
    table = market[name]
    db.create_table(name, table.schema, time_column="ts_init")
    db.append(name, table, note="book synthesized from bars at 5bp")
db.snapshot(
    "tape-5bp",
    tables=["prices", "instruments", "book_deltas", "trades"],
    note="Prices plus the 5bp synthetic book that every run here reads",
)
db.tables()

monthly = (
    db.table("prices", snapshot="tape-5bp")
    .with_columns(month=time_bucket("1mo", col("ts")))
    .group_by("symbol", "month")
    .agg(close=col("close").last("ts"), formation_ts=col("ts").max())
    .with_columns(
        lag_1=sql_expr("lag(close, 1)").over(partition_by="symbol", order_by="month"),
        lag_12=sql_expr("lag(close, 12)").over(partition_by="symbol", order_by="month"),
    )
    .with_columns(momentum=col("lag_1") / col("lag_12") - 1)
    .filter(col("momentum").is_not_null())
    .sort(["month", "symbol"])
)
signal = monthly.to_pandas()
signal = signal[signal["formation_ts"] >= pd.Timestamp("2022-02-01", tz="UTC")]
print(f"{len(signal):,} symbol-months over {signal['month'].nunique()} rebalances")
signal.head()

CAPITAL = 300_000.0
TOP_N = 3

rows = []
for month, group in signal.groupby("month", sort=True):
    picks = group.nlargest(TOP_N, "momentum")
    formation = picks["formation_ts"].max()
    for pick in picks.itertuples():
        rows.append(
            {
                "formation_ts": formation,
                "symbol": pick.symbol,
                "shares": float(int(CAPITAL / TOP_N / pick.close)),
            }
        )
picked = pd.DataFrame(rows)
sessions = pd.DatetimeIndex(daily.to_pandas()["ts"].unique()).sort_values()
formation_index = sessions.searchsorted(picked["formation_ts"])
picked = picked[formation_index + 2 < len(sessions)]

grid = pd.MultiIndex.from_product(
    [sorted(picked["formation_ts"].unique()), sorted(picked["symbol"].unique())],
    names=["formation_ts", "symbol"],
)
positions = (
    picked.set_index(["formation_ts", "symbol"])
    .reindex(grid, fill_value=0.0)
    .reset_index()
    .sort_values(["formation_ts", "symbol"])
)
print(f"{len(positions):,} target rows over {positions['formation_ts'].nunique()} rebalances")
positions.head(6)

book = market["book_deltas"].to_pandas()
book["session"] = book["ts_init"].dt.floor("s").dt.tz_localize("UTC")
quote_at = {(row.session, row.instrument_id): row.ts_init for row in book.itertuples()}
touch_at = {
    (row.session, row.instrument_id, row.side): row.price for row in book.itertuples()
}
depth_at = {
    (row.session, row.instrument_id, row.side): row.size for row in book.itertuples()
}

positions["execution_ts"] = sessions[
    sessions.searchsorted(positions["formation_ts"], side="right")
]
stamps = [
    quote_at[(row.execution_ts, row.symbol)] + dt.timedelta(microseconds=1)
    for row in positions.itertuples()
]
signals = backtest.target_positions(
    stamps,
    positions["shares"].tolist(),
    instrument_id=positions["symbol"].tolist(),
    tag="momentum-12-1",
)
backtest.create_signal_table(db)
db.append("signals", signals, note="12-1 momentum, top 3, whole shares")

orders = signals.to_pandas()
orders["execution_session"] = pd.DatetimeIndex(orders["ts"]).tz_localize("UTC").floor("s")
orders["formation_session"] = sessions[
    sessions.searchsorted(orders["execution_session"]) - 1
]
print(f"{signals.num_rows:,} orders from {len(positions):,} targets")
orders.head()[["ts", "instrument_id", "side", "quantity", "formation_session"]]

FEE_RATE = 0.0005

report = backtest.run(
    db,
    "momentum-5bp",
    starting_cash=CAPITAL,
    signals="signals",
    snapshot="tape-5bp",
    fee_kind="proportional",
    fee_rate=FEE_RATE,
    equity_interval_nanos=86_400_000_000_000,
)
run_db = db.fork(report["fork"])
fills = run_db.read("bt_fills").to_pandas()
print(f"orders {report['orders']}  fills {report['fills']}  "
      f"commissions {report['commissions']:,.2f}")
print(f"final cash {report['final_cash']:,.2f}")
fills.head()

consumed = orders.apply(
    lambda order: order["quantity"]
    / depth_at[
        (
            order["execution_session"],
            order["instrument_id"],
            "sell" if order["side"] == "buy" else "buy",
        )
    ],
    axis=1,
)
assert len(fills) == report["orders"], "a partial fill would break the reconciliation"
assert consumed.max() < 1.0, "an order asked for more than the book showed"
print(f"orders filled in full     {len(fills)} of {report['orders']}")
print(f"largest order vs depth    {consumed.max():.1%}")
print(f"median order vs depth     {consumed.median():.1%}")

closes = daily.to_pandas().pivot(index="ts", columns="symbol", values="close").sort_index()
close_at = closes.stack().to_dict()
orders.head(3)[["ts", "instrument_id", "side", "quantity", "execution_session"]]

executions = fills.sort_values("order_id").reset_index(drop=True)
assert (executions["instrument_id"] == orders["instrument_id"]).all()
assert (executions["side"] == orders["side"]).all()
assert (executions["quantity"] == orders["quantity"]).all()

expected = [
    touch_at[(row.execution_session, row.instrument_id, "sell" if row.side == "buy" else "buy")]
    for row in orders.itertuples()
]
recorded = pd.DatetimeIndex(executions["ts"]).tz_localize("UTC")
assert (executions["price"] - expected).abs().max() < 1e-9, "a fill missed the touch"
assert (recorded > orders["execution_session"]).all(), "a fill was recorded too early"
print(f"all {len(orders)} fills took the touch of their execution session,")
print("and every one was recorded at the following session")

def ending_equity(price_of) -> float:
    """Ending equity from the order sequence, priced by `price_of`."""
    cash, held = CAPITAL, {}
    for order in orders.itertuples():
        signed = order.quantity if order.side == "buy" else -order.quantity
        cash -= signed * price_of(order)
        held[order.instrument_id] = held.get(order.instrument_id, 0.0) + signed
    return cash + sum(
        quantity * close_at[(sessions[-1], name)] for name, quantity in held.items()
    )


formation_close = ending_equity(
    lambda o: close_at[(o.formation_session, o.instrument_id)]
)
execution_close = ending_equity(
    lambda o: close_at[(o.execution_session, o.instrument_id)]
)
execution_touch = ending_equity(
    lambda o: touch_at[
        (o.execution_session, o.instrument_id, "sell" if o.side == "buy" else "buy")
    ]
)
for label, value in (
    ("formation close", formation_close),
    ("execution close", execution_close),
    ("execution touch", execution_touch),
):
    print(f"{label:<18} {value:>14,.2f}")

holdings = run_db.read("bt_positions").to_pandas()
event_driven = float(report["final_cash"]) + sum(
    row.quantity * close_at[(sessions[-1], row.instrument_id)]
    for row in holdings.itertuples()
)

waterfall = pd.DataFrame(
    [
        {"step": "vectorized (formation close)", "equity": formation_close, "delta": 0.0},
        {"step": "timing: trade the next close", "equity": execution_close,
         "delta": execution_close - formation_close},
        {"step": "spread: pay the touch", "equity": execution_touch,
         "delta": execution_touch - execution_close},
        {"step": "fees: proportional commission",
         "equity": execution_touch - report["commissions"],
         "delta": -report["commissions"]},
        {"step": "event-driven run", "equity": event_driven,
         "delta": event_driven - (execution_touch - report["commissions"])},
    ]
)
residual = waterfall["delta"].iloc[-1]
assert abs(residual) < 0.05, f"unexplained {residual:,.4f}"
waterfall.round(2)

fig, ax = plt.subplots(figsize=(9, 4))
steps = waterfall.iloc[1:-1]
ax.bar(steps["step"], steps["delta"], color=["#4c78a8", "#e45756", "#f58518"])
ax.axhline(0, color="black", linewidth=0.8)
ax.set_title("What the vectorized backtest was hiding")
ax.set_xlabel("Assumption dropped")
ax.set_ylabel("Effect on ending equity")
ax.tick_params(axis="x", labelrotation=10)
fig.tight_layout()

trials = []
for spread_bps in (1.0, 5.0, 20.0, 50.0):
    scenario = cu.make_equity_market(tape, spread_bps=spread_bps, depth_fraction=0.002)
    trial = h5i_db.Database(cu.fresh_db(f"04_equity_spread_{int(spread_bps)}"), create=True)
    for name in ("instruments", "book_deltas", "trades"):
        table = scenario[name]
        trial.create_table(name, table.schema, time_column="ts_init")
        trial.append(name, table)
    trial.snapshot("tape", tables=["instruments", "book_deltas", "trades"])
    backtest.create_signal_table(trial)
    trial.append("signals", signals)
    trial_report = backtest.run(
        trial,
        f"momentum-{int(spread_bps)}bp",
        starting_cash=CAPITAL,
        signals="signals",
        snapshot="tape",
        fee_kind="proportional",
        fee_rate=FEE_RATE,
        equity_interval_nanos=86_400_000_000_000,
    )
    trial_fork = trial.fork(trial_report["fork"])
    still_held = trial_fork.read("bt_positions").to_pandas()
    ending = float(trial_report["final_cash"]) + sum(
        row.quantity * close_at[(sessions[-1], row.instrument_id)]
        for row in still_held.itertuples()
    )
    trials.append(
        {
            "spread_bps": spread_bps,
            "ending_equity": ending,
            "return_pct": 100 * (ending / CAPITAL - 1),
            "commissions": trial_report["commissions"],
        }
    )
    trial_fork.close()
    trial.close()
sensitivity = pd.DataFrame(trials)
sensitivity.round(2)

fig, ax = plt.subplots(figsize=(9, 4))
ax.plot(sensitivity["spread_bps"], sensitivity["return_pct"], marker="o", linewidth=1.8)
ax.set_title("Ending return against the assumed spread")
ax.set_xlabel("Assumed spread (basis points)")
ax.set_ylabel("Total return (%)")
fig.tight_layout()