Causal signal replay on real Polymarket books

import datetime as dt
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

import h5i_db
from h5i_db import backtest
import cookbook_utils as cu

CACHE = Path("data/cache/kaggle-polymarket")
sample = cu.load_kaggle_sample(CACHE, depth_levels=10)
print(sample.question)
print(pd.Series(sample.audit, name="value").to_frame())

features = sample.features.to_pandas()
depth = features["depth_imbalance"]
features["history_mean"] = depth.rolling(120, min_periods=60).mean().shift(1)
features["history_std"] = depth.rolling(120, min_periods=60).std().shift(1)
features["depth_z"] = (
    (depth - features["history_mean"]) / features["history_std"].replace(0, np.nan)
)

book_times = sample.book_deltas.column("ts_init")
book_start = pd.Timestamp(book_times[0].as_py())
book_end = pd.Timestamp(book_times[-1].as_py())
eligible = features[
    features["ts_init"].between(
        book_start, book_end - pd.Timedelta(hours=2), inclusive="both"
    )
    & features["depth_z"].abs().ge(2.0)
]
assert not eligible.empty
entry = eligible.iloc[0]
exit_time = entry["ts_init"] + pd.Timedelta(hours=2)
direction = "buy" if entry["depth_z"] > 0 else "sell"
close_direction = "sell" if direction == "buy" else "buy"
print(
    f"{direction=} at {entry['ts_init']} z={entry['depth_z']:.2f}; "
    f"close at {exit_time}"
)

quantity = 20.0
signals = backtest.signal_table(
    [
        {
            "ts": entry["ts_init"].to_pydatetime(),
            "instrument_id": sample.market_id,
            "side": direction,
            "quantity": quantity,
            "tag": "depth-z-entry",
        },
        {
            "ts": exit_time.to_pydatetime(),
            "instrument_id": sample.market_id,
            "side": close_direction,
            "quantity": quantity,
            "tag": "time-exit",
        },
    ]
)
signals.to_pandas()

db = h5i_db.Database(cu.fresh_db("05_kaggle_polymarket_replay"), create=True)
for name, table in {
    "instruments": sample.instruments,
    "book_deltas": sample.book_deltas,
    "trades": sample.trades,
    "features_1m": sample.features,
}.items():
    db.create_table(name, table.schema, time_column="ts_init")
    db.append(name, table, note="bounded CC BY-NC Kaggle Polymarket sample")
db.snapshot(
    "approved-kaggle-cut",
    tables=["instruments", "book_deltas", "trades", "features_1m"],
    note="Real snapshot replay inputs; resolution label excluded",
)
backtest.create_signal_table(db)
db.append("signals", signals, note="causal depth-imbalance experiment")

scenarios = {
    "frictionless": {
        "fee_rate": 0.0,
        "latency_nanos": 0,
        "slippage_ticks": 0,
    },
    "base": {
        "fee_rate": 0.02,
        "latency_nanos": 50_000_000,
        "slippage_ticks": 0,
    },
    "stressed": {
        "fee_rate": 0.035,
        "latency_nanos": 250_000_000,
        "slippage_ticks": 2,
    },
}
reports = {}
inspections = {}
for name, assumptions in scenarios.items():
    config = backtest.BacktestConfig(
        run_id=f"kaggle-{name}",
        portfolio=backtest.PortfolioConfig(starting_cash=10_000.0),
        data=backtest.DataConfig(
            signals="signals",
            snapshot="approved-kaggle-cut",
            minimum_coverage=0.75,
        ),
        execution=backtest.ExecutionConfig(
            fee_kind="prediction_market",
            fee_rate=assumptions["fee_rate"],
            latency_nanos=assumptions["latency_nanos"],
            slippage_ticks=assumptions["slippage_ticks"],
        ),
        risk=backtest.RiskConfig(
            max_order_quantity=quantity,
            max_abs_position=quantity,
            max_open_orders=1,
        ),
        output=backtest.OutputConfig(
            equity_interval_nanos=60_000_000_000,
        ),
        metadata={
            "dataset": cu.KAGGLE_POLYMARKET_DATASET,
            "scenario": name,
            "signal": "causal depth imbalance",
        },
    )
    inspections[name] = backtest.inspect(db, config)
    inspections[name].raise_for_errors()
    reports[name] = backtest.execute(db, config)

summary = pd.DataFrame(reports).T[
    ["fills", "commissions", "realized_pnl", "final_cash", "digest", "fork"]
]
summary

{
    name: {
        "fidelity": inspection.fidelity.value,
        "warnings": [issue.message for issue in inspection.warnings],
    }
    for name, inspection in inspections.items()
}

assert summary["fills"].ge(2).all()
assert summary["digest"].nunique() == len(summary)

fill_frames = []
for scenario, report in reports.items():
    run_db = db.fork(report["fork"])
    scenario_fills = run_db.read("bt_fills").to_pandas()
    filled_by_tag = scenario_fills.groupby("tag")["quantity"].sum()
    assert set(filled_by_tag.index) == {"depth-z-entry", "time-exit"}
    assert filled_by_tag.eq(quantity).all()
    scenario_fills["scenario"] = scenario
    fill_frames.append(scenario_fills)
    run_db.close()

# verify() reruns the config into a scratch fork and tears it down afterwards,
# so read the run's own fork before calling it.
assert reports["base"].verify()["verified"]

fills = pd.concat(fill_frames, ignore_index=True)
fills[["scenario", "ts", "side", "price", "quantity", "commission", "tag"]]

stressed_db = db.fork(reports["stressed"]["fork"])
equity = stressed_db.read("bt_equity").to_pandas()
fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(equity["ts"], equity["equity"], linewidth=1.6)
ax.set(
    title="Stressed execution scenario",
    xlabel="simulated time",
    ylabel="portfolio value",
)
fig.tight_layout()