Trading the favorite-longshot bias

panel = cu.make_prediction_markets(n_markets=180, steps=32, seed=11)
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="favorite-longshot study")
print(f"{panel['book_deltas'].num_rows:,} book rows, {panel['resolutions'].num_rows} markets")
panel["book_deltas"].to_pandas().head()

stamps = sorted({value.as_py() for value in panel["book_deltas"].column("ts_init")})
decision = stamps[8]
decision_ns = int(pd.Timestamp(decision, tz="UTC").value)
quotes = db.sql(
    f"""
    SELECT 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({decision_ns})
    GROUP BY instrument_id, outcome
    """
).to_pandas()
truth = cu.market_truth(panel).to_pandas()

yes = quotes[quotes.outcome == 0].merge(truth, on="instrument_id", validate="one_to_one")
yes["payoff"] = yes.yes_won.astype(float)
yes["gross_return"] = (yes.payoff - yes.ask) / yes.ask
yes["fee_per_contract"] = FEE_RATE * yes.ask * (1.0 - yes.ask)
yes["net_return"] = (yes.payoff - yes.ask - yes.fee_per_contract) / yes.ask
print(f"decision instant {decision}, {len(yes)} markets")

EDGES = [0.0, 0.15, 0.30, 0.45, 0.60, 0.75, 0.88, 1.0]
yes["bucket"] = pd.cut(yes.ask, EDGES)
buckets = (
    yes.groupby("bucket", observed=True)
    .agg(
        n=("ask", "size"),
        mean_ask=("ask", "mean"),
        hit_rate=("payoff", "mean"),
        gross=("gross_return", "mean"),
        net=("net_return", "mean"),
    )
    .assign(edge_pp=lambda f: (f.hit_rate - f.mean_ask) * 100)
)
print(buckets.round(3).to_string())

fig, ax = plt.subplots(figsize=(8, 4.5))
centres = buckets.mean_ask
width = 0.05
ax.bar(centres - width / 2, buckets.gross * 100, width=width, label="gross", color="#2c7fb8")
ax.bar(centres + width / 2, buckets.net * 100, width=width, label="net of fees", color="#c0392b")
ax.axhline(0.0, color="black", lw=0.8)
ax.set_title("Hold-to-resolution return on a long YES position, by price")
ax.set_xlabel("YES ask at the decision instant")
ax.set_ylabel("return on capital, %")
ax.legend(fontsize=8)
fig.tight_layout()

levels = pd.DataFrame({"price": [0.10, 0.25, 0.50, 0.75, 0.90]})
levels["fee_per_contract"] = FEE_RATE * levels.price * (1 - levels.price)
levels["fee_pct_of_capital"] = levels.fee_per_contract / levels.price * 100
print(levels.round(4).to_string(index=False))

FAVORITE, LONGSHOT = 0.75, 0.30
submit = decision + dt.timedelta(microseconds=1)


def signals_for(members: pd.DataFrame, tag: str) -> pa.Table:
    rows = [
        {
            "ts": submit,
            "instrument_id": row.instrument_id,
            "outcome": 0,
            "side": "buy",
            "quantity": QUANTITY,
            "tag": tag,
        }
        for row in members.itertuples()
    ]
    return backtest.signal_table(rows).sort_by([("ts", "ascending")])


legs = {
    "favorites": yes[yes.ask >= FAVORITE],
    "longshots": yes[yes.ask <= LONGSHOT],
}
for name, members in legs.items():
    table = signals_for(members, name)
    db.create_table(f"signals_{name}", table.schema, time_column="ts")
    db.append(f"signals_{name}", table)
    print(f"{name:10} {len(members):>3} markets, mean ask {members.ask.mean():.3f}")

def run(name: str, fee_kind: str | None) -> backtest.BacktestResult:
    execution = (
        backtest.ExecutionConfig(fee_kind=fee_kind, fee_rate=FEE_RATE)
        if fee_kind
        else backtest.ExecutionConfig()
    )
    return backtest.execute(
        db,
        backtest.BacktestConfig(
            run_id=f"flb-{name}-{fee_kind or 'gross'}",
            data=backtest.DataConfig(signals=f"signals_{name}", snapshot="panel-v1"),
            portfolio=backtest.PortfolioConfig(starting_cash=100_000.0),
            execution=execution,
            metadata={"study": "favorite-longshot", "leg": name},
        ),
    )


def account(result: backtest.BacktestResult) -> dict[str, float]:
    """Positions are held to resolution, so the result is in settlement_pnl."""
    positions = result.positions.to_pandas()
    fills = result.fills.to_pandas()
    settled = float(positions.settlement_pnl.fillna(0.0).sum())
    fees = float(result.summary()["commissions"])
    capital = float((fills.price * fills.quantity).sum())
    # realized_pnl already nets out commissions, so total = realized + settled.
    # These entries are held to resolution, so realized is exactly -fees and the
    # two agree; a rule that closes positions would make them differ.
    net = float(result.summary()["realized_pnl"]) + settled
    return {
        "fills": len(fills),
        "capital": capital,
        "settled": settled,
        "fees": fees,
        "net": net,
        "net_return_pct": net / capital * 100 if capital else float("nan"),
    }


report = {}
for name in legs:
    for fee_kind in (None, "kalshi"):
        label = f"{name} / {'gross' if fee_kind is None else 'net'}"
        report[label] = account(run(name, fee_kind))
print(pd.DataFrame(report).T.round(2).to_string())

sweep = []
for threshold in np.arange(0.40, 0.92, 0.04):
    members = yes[yes.ask >= threshold]
    if len(members) < 5:
        continue
    sweep.append(
        {
            "threshold": round(float(threshold), 2),
            "markets": len(members),
            "mean_ask": members.ask.mean(),
            "gross_pct": members.gross_return.mean() * 100,
            "net_pct": members.net_return.mean() * 100,
        }
    )
sweep = pd.DataFrame(sweep)
print(sweep.round(2).to_string(index=False))

expiry_ns = int(db.sql("SELECT max(expiration_ns) AS e FROM instruments").to_pandas()["e"][0])
paths = db.sql(
    f"""
    SELECT instrument_id, ts_init,
           (max(CASE WHEN side = 'buy'  THEN price END)
          + max(CASE WHEN side = 'sell' THEN price END)) / 2 AS mid
    FROM h5i('book_deltas', 'panel-v1')
    WHERE outcome = 0 AND ts_init <= to_timestamp_nanos({expiry_ns})
    GROUP BY instrument_id, ts_init
    ORDER BY instrument_id, ts_init
    """
).to_pandas()
paths["step_move"] = paths.groupby("instrument_id").mid.diff()
paths["level"] = paths.groupby("instrument_id").mid.transform("mean").round(1).clip(0.1, 0.9)
vol = (
    paths.dropna(subset=["step_move"])
    .groupby("level")
    .agg(observations=("step_move", "size"), step_vol_bp=("step_move", lambda s: s.std() * 10_000))
    .assign(sqrt_p1mp=lambda f: np.sqrt(f.index * (1 - f.index)))
)
vol["vol_over_sqrt"] = vol.step_vol_bp / vol.sqrt_p1mp
print(vol.round(1).to_string())
print("\nthe last column is roughly flat, which is what p*(1-p) scaling looks like")