Binary parity and the fee curve

panel = cu.make_prediction_markets(n_markets=120, steps=32, seed=11)
book = panel["book_deltas"]
print(f"{book.num_rows:,} rows x {book.num_columns} columns")
book.to_pandas().head()

for name, table in panel.items():
    db.create_table(name, table.schema, time_column="ts_init")
    db.append(name, table, note="panel load")
db.snapshot("panel-v1", tables=list(panel), note="parity study input")
print({name: db.sql(f"SELECT count(*) AS n FROM {name}").to_pandas()["n"][0] for name in panel})

# Stop at expiry. The panel keeps quoting after the result is observable, and a
# resolved book sits at ~1.00 / ~0.00: not an opportunity, just an answer.
expiry_ns = int(db.sql("SELECT max(expiration_ns) AS e FROM instruments").to_pandas()["e"][0])
quotes = db.sql(
    f"""
    SELECT ts_init AS ts, instrument_id, outcome,
           max(CASE WHEN side = 'buy'  THEN price END) AS bid,
           max(CASE WHEN side = 'sell' THEN price END) AS ask
    FROM h5i('book_deltas', 'panel-v1')
    WHERE ts_init <= to_timestamp_nanos({expiry_ns})
    GROUP BY ts_init, instrument_id, outcome
    """
).to_arrow()
db.create_table("quotes", quotes.schema, time_column="ts")
db.append("quotes", quotes.sort_by([("ts", "ascending")]))

yes = db.table("quotes").filter(col("outcome") == 0)
no = db.table("quotes").filter(col("outcome") == 1)
pairs = (
    yes.join(no, on=["ts", "instrument_id"], how="inner")
    .select(
        ts=col("ts", relation="l"),
        instrument_id=col("instrument_id", relation="l"),
        ask_yes=col("ask", relation="l"),
        ask_no=col("ask", relation="r"),
        mid_yes=(col("bid", relation="l") + col("ask", relation="l")) / 2,
    )
    .with_columns(pair_cost=col("ask_yes") + col("ask_no"))
    .with_columns(gross_edge=1.0 - col("pair_cost"))
    .to_pandas()
)
print(f"{len(pairs):,} market-instants")
print(f"pair cost: min {pairs.pair_cost.min():.4f}  median {pairs.pair_cost.median():.4f}  max {pairs.pair_cost.max():.4f}")
print(f"below 1.00 (gross arbitrage): {(pairs.gross_edge > 0).mean():.1%}")

FEE_RATE = 0.07


def pair_fee(mid_yes: float, rate: float = FEE_RATE) -> float:
    """Kalshi-style quadratic fee for one contract of each leg."""
    mid_no = 1.0 - mid_yes
    return rate * (mid_yes * (1.0 - mid_yes) + mid_no * (1.0 - mid_no))


pairs["fee_hurdle"] = pairs.mid_yes.map(pair_fee)
pairs["net_edge"] = pairs.gross_edge - pairs.fee_hurdle
print(f"clears the hurdle: {(pairs.net_edge > 0).mean():.1%} of market-instants")
for level in (0.05, 0.15, 0.30, 0.50):
    print(f"  hurdle at p={level:.2f}: {pair_fee(level):.4f}")

fig, ax = plt.subplots(figsize=(8, 4.5))
survivors = pairs[pairs.net_edge > 0]
ax.scatter(pairs.mid_yes, pairs.gross_edge * 100, s=6, alpha=0.25, label="gross basis")
ax.scatter(survivors.mid_yes, survivors.gross_edge * 100, s=10, color="#c0392b", label="clears fees")
grid = [i / 100 for i in range(2, 99)]
ax.plot(grid, [pair_fee(p) * 100 for p in grid], color="black", lw=1.5, label="fee hurdle")
ax.axhline(0.0, color="grey", lw=0.8)
ax.set_title("A basis is only tradeable where the fee curve is low")
ax.set_xlabel("YES mid (probability)")
ax.set_ylabel("edge, cents per contract")
ax.legend(loc="upper right", fontsize=8)
fig.tight_layout()

QUANTITY = 20.0
picks = (
    pairs[pairs.net_edge > 0]
    .sort_values(["instrument_id", "ts"])
    .groupby("instrument_id", as_index=False)
    .first()
)
rows = []
for pick in picks.itertuples():
    submit = pick.ts.to_pydatetime() + dt.timedelta(microseconds=1)
    for outcome in (0, 1):
        rows.append(
            {
                "ts": submit,
                "instrument_id": pick.instrument_id,
                "outcome": outcome,
                "side": "buy",
                "quantity": QUANTITY,
                "tag": f"parity-{'yes' if outcome == 0 else 'no'}",
            }
        )
signals = backtest.signal_table(rows).sort_by([("ts", "ascending")])
backtest.create_signal_table(db, "signals")
db.append("signals", signals)
print(f"{len(picks)} markets, {signals.num_rows} legs")
print(f"mean gross edge selected: {picks.gross_edge.mean() * 100:.2f} cents")

def configure(run_id: str, fee_kind: str | None) -> backtest.BacktestConfig:
    execution = (
        backtest.ExecutionConfig(fee_kind=fee_kind, fee_rate=FEE_RATE)
        if fee_kind
        else backtest.ExecutionConfig()
    )
    return backtest.BacktestConfig(
        run_id=run_id,
        data=backtest.DataConfig(signals="signals", snapshot="panel-v1"),
        portfolio=backtest.PortfolioConfig(starting_cash=100_000.0),
        execution=execution,
        metadata={"study": "binary-parity"},
    )


gross = backtest.execute(db, configure("parity-gross", None))
net = backtest.execute(db, configure("parity-kalshi", "kalshi"))


def account(result: backtest.BacktestResult) -> dict[str, float]:
    """Total the run. `realized_pnl` covers closed trades only; these pairs are
    held to resolution, so the result lives in `bt_positions.settlement_pnl`."""
    positions = result.positions.to_pandas()
    summary = result.summary()
    settled = float(positions.settlement_pnl.fillna(0.0).sum())
    fees = float(summary["commissions"])
    # realized_pnl already carries the commissions, so the total is
    # realized + settlement. Here nothing closes before resolution, which makes
    # realized exactly -fees and the two spellings agree; they part company the
    # moment a rule round-trips, so use the one that always holds.
    net = float(summary["realized_pnl"]) + settled
    return {"fills": summary["fills"], "settled": settled, "fees": fees, "net": net}


for label, result in (("gross", gross), ("kalshi fees", net)):
    row = account(result)
    print(
        f"{label:12} fills={row['fills']:>4}  settlement={row['settled']:>8,.2f}  "
        f"fees={row['fees']:>7,.2f}  net={row['net']:>8,.2f}"
    )

fills = gross.fills.to_pandas()
realized_cost = fills.groupby("instrument_id").price.sum()
expected = float((QUANTITY * (1.0 - realized_cost)).sum())
positions = gross.positions.to_pandas()
booked = float(positions.settlement_pnl.sum())
print(f"pairs filled: {len(realized_cost)} of {len(picks)}")
print(f"expected from arithmetic: {expected:,.2f}")
print(f"booked as settlement:     {booked:,.2f}")
assert abs(expected - booked) < 1e-6, (expected, booked)
after_fees = account(net)
print(f"the same pairs after the real fee curve: {after_fees['net']:,.2f}")
print(f"fees consumed {after_fees['fees'] / expected:.0%} of the gross edge")

naive = pairs[pairs.gross_edge > 0]
print(f"gross-positive opportunities: {len(naive):,}")
print(f"  mean gross edge: {naive.gross_edge.mean() * 100:+.2f} cents")
print(f"  mean net edge:   {naive.net_edge.mean() * 100:+.2f} cents")
print(f"  fraction still positive after fees: {(naive.net_edge > 0).mean():.1%}")
by_level = naive.assign(
    level=lambda f: f.mid_yes.round(1).clip(0.1, 0.9)
).groupby("level").net_edge.mean() * 100
print("\nmean net edge by price level, cents:")
print(by_level.round(2).to_string())