Replaying an account's ledger

raw = cu.make_prediction_markets(n_markets=12, steps=24, seed=5)
offsets = {
    name: pd.Timedelta(milliseconds=index)
    for index, name in enumerate(sorted(set(raw["instruments"].column("instrument_id").to_pylist())))
}


def stagger(table):
    """Give every market its own instant, as a real feed would."""
    frame = table.to_pandas()
    if "instrument_id" not in frame.columns:
        return table
    frame["ts_init"] = frame["ts_init"] + frame["instrument_id"].map(offsets)
    if "ts_event" in frame.columns:
        frame["ts_event"] = frame["ts_init"]
    frame = frame.sort_values("ts_init", kind="stable")
    return pa.Table.from_pandas(frame, schema=table.schema, preserve_index=False)


panel = {name: stagger(table) for name, table in raw.items()}
db = h5i_db.Database(cu.fresh_db("04_replaying_an_account_ledger"), create=True)
for name, table in panel.items():
    db.create_table(name, table.schema, time_column="ts_init")
    db.append(name, table, note="panel fixture, one instant per market")
db.snapshot("books-v1", tables=list(panel), note="The book the ledger is checked against")
for name, table in panel.items():
    print(f"{name}: {table.num_rows:,} rows x {table.num_columns} columns")

specs = venues.polymarket_markets_from_json(cu.polymarket_market_payloads(raw))
print(f"\n{len(specs)} market specs, tokens resolved positionally to outcomes")
panel["book_deltas"].to_pandas().head(4)

book = panel["book_deltas"].to_pandas()
quotes = (
    book.pivot_table(
        index=["ts_init", "instrument_id", "outcome"],
        columns="side",
        values=["price", "size"],
        aggfunc="last",
    )
    .dropna()
    .reset_index()
)
quotes.columns = ["ts_init", "instrument_id", "outcome", "bid", "ask", "bid_size", "ask_size"]
quotes = quotes[quotes["outcome"] == 0].sort_values("ts_init")
print(f"{len(quotes):,} usable two-sided quotes")
quotes.head(3)

rows, kinds = [], []
for index, quote in enumerate(quotes.iloc[40:100:6].itertuples()):
    kind = ("at the touch", "better than the touch", "oversized")[index % 3]
    # Prices stay on the market's 0.001 grid; an off-grid order is refused
    # before it ever meets the book.
    price = (
        round(float(quote.bid) - TICK, 3)
        if kind == "better than the touch"
        else float(quote.ask)
    )
    quantity = float(quote.ask_size) * (3.0 if kind == "oversized" else 0.4)
    rows.append(
        venues.LedgerRow(
            ts_ns=int(pd.Timestamp(quote.ts_init).value),
            instrument_id=quote.instrument_id,
            outcome=0,
            side="buy",
            quantity=round(quantity, 2),
            price=price,
            trade_id=f"0x{index:04d}",
        )
    )
    kinds.append(kind)

ledger = venues.ledger_table(rows).to_pandas().assign(kind=kinds)
print(f"{len(rows)} ledger rows over {ledger['instrument_id'].nunique()} markets")
ledger.head(6)

for description, build in (
    ("price of 1.4", lambda: venues.LedgerRow(
        ts_ns=0, instrument_id="m", outcome=0, side="buy", quantity=10.0, price=1.4)),
    ("quantity of 0", lambda: venues.LedgerRow(
        ts_ns=0, instrument_id="m", outcome=0, side="buy", quantity=0.0, price=0.5)),
    ("side 'long'", lambda: venues.LedgerRow(
        ts_ns=0, instrument_id="m", outcome=0, side="long", quantity=10.0, price=0.5)),
):
    try:
        build()
    except ValueError as error:
        print(f"{description:16} refused: {str(error).split(';')[0][:70]}")

commands = venues.commands_from_ledger(rows, specs)
backtest.create_command_table(db, "commands")
db.append("commands", commands, note="one account's published trades, as intent")
frame = commands.to_pandas()
print(f"{commands.num_rows} commands")
frame[["ts", "action", "instrument_id", "side", "quantity", "kind", "limit_price",
       "time_in_force", "reduce_only"]].head(4)

result = backtest.execute(
    db,
    backtest.BacktestConfig(
        run_id="ledger-replay",
        portfolio=backtest.PortfolioConfig(starting_cash=100_000.0),
        data=backtest.DataConfig(commands="commands", snapshot="books-v1"),
        execution=backtest.ExecutionConfig(fee_kind="prediction_market", fee_rate=0.02),
        output=backtest.OutputConfig(equity_interval_nanos=900_000_000_000),
    ),
)
print(f"orders {result['orders']}  fills {result['fills']}")
result.orders.to_pandas()[["order_id", "instrument_id", "side", "quantity", "filled", "status"]].head(6)

comparison = venues.compare_to_ledger(result, rows)
print(f"ledger rows        {comparison['ledger_rows']}")
print(f"replay fills       {comparison['replay_fills']}")
print(f"ledger quantity    {comparison['ledger_quantity']:,.2f}")
print(f"replay quantity    {comparison['replay_quantity']:,.2f}")
print(f"fill ratio         {comparison['fill_ratio']:.1%}")
print(f"markets reproduced {comparison['markets_reproduced']} of {len(comparison['markets'])}")
pd.DataFrame(comparison["markets"]).round(3).head(8)

submitted = result.orders.to_pandas().sort_values("order_id").reset_index(drop=True)
assert (submitted["quantity"].round(2).to_numpy() == ledger["quantity"].round(2).to_numpy()).all(), (
    "orders and ledger rows are no longer in the same order"
)
ledger["replayed"] = submitted["filled"].to_numpy()
by_kind = ledger.groupby("kind").agg(
    rows=("quantity", "size"),
    ledger_quantity=("quantity", "sum"),
    replayed_quantity=("replayed", "sum"),
)
by_kind["ratio"] = by_kind["replayed_quantity"] / by_kind["ledger_quantity"]
by_kind.round(3)

assert by_kind.loc["at the touch", "ratio"] > 0.99
assert by_kind.loc["better than the touch", "ratio"] < 0.01
assert by_kind.loc["oversized", "ratio"] < 1.0
print("the reconciliation found exactly the rows that were built to fail")

explanation = result.explain()
print(f"orders without fills {explanation.get('orders_without_fills')}")
unfilled = result.orders.to_pandas()
unfilled = unfilled[unfilled["filled"] == 0]
print(f"{len(unfilled)} of {result['orders']} orders never filled")
unfilled[["instrument_id", "side", "quantity", "limit_price", "status"]].head(5)