The whole loop, once

panel_tables = cu.make_prediction_markets(n_markets=60, steps=32, seed=11)
mirror = Path("data/cache/e2e-mirror")
files = cu.write_polymarket_archive(panel_tables, mirror)
specs = venues.polymarket_markets_from_json(cu.polymarket_market_payloads(panel_tables))

venues.write_markets(db, specs, note="market definitions")
ingest = venues.ingest_archive(
    db, files=files, markets=specs, layout=venues.PMXT_LAYOUT, note="backfill"
)
db.snapshot("mirror-v1", tables=list(venues.CANONICAL_SCHEMAS), note="study input")
print(f"{len(files)} files -> {ingest.rows:,} rows across {len(ingest.tables)} tables")
print(f"gaps: {len(ingest.gaps)}, unrecognised events: {len(ingest.skipped)}")
print({name: write.rows for name, write in sorted(ingest.tables.items())})

panel = backtest.quote_panel(db, snapshot="mirror-v1")
print(f"{len(panel):,} rows x {len(panel.columns)} columns")
print(f"{panel.instrument_id.nunique()} markets, {panel.ts.nunique()} instants")
expiry = int(db.sql("SELECT max(expiration_ns) AS e FROM instruments").to_pandas()["e"][0])
print(f"panel ends {panel.ts.max()}, expiry {pd.Timestamp(expiry, unit='ns')}")
panel.head()

print("available:", ", ".join(sorted(backtest.STRATEGIES)))
counts = {}
for name, generator in sorted(backtest.STRATEGIES.items()):
    counts[name] = generator(panel).num_signals
print("\nsignals each rule produces on this panel:")
for name, count in sorted(counts.items(), key=lambda item: -item[1]):
    print(f"  {name:24} {count:>5}")

plan = backtest.strategies.ema_crossover(
    panel, fast=4, slow=12, quantity=20.0, long_only=True
)
print(f"strategy   {plan.strategy}")
print(f"parameters {plan.parameters}")
print(f"signals    {plan.num_signals}")
db.create_table("signals", plan.signals.schema, time_column="ts")
db.append("signals", plan.signals, note=f"{plan.strategy} {plan.parameters}")
plan.signals.to_pandas().head(3)

def config(
    run_id: str, *, fee_rate_override: float | None = None, **execution
) -> backtest.BacktestConfig:
    return backtest.BacktestConfig(
        run_id=run_id,
        data=backtest.DataConfig(signals="signals", snapshot="mirror-v1"),
        portfolio=backtest.PortfolioConfig(starting_cash=250_000.0),
        execution=backtest.ExecutionConfig(
            fee_kind="kalshi",
            fee_rate=FEE_RATE if fee_rate_override is None else fee_rate_override,
            **execution,
        ),
        output=backtest.OutputConfig(equity_interval_nanos=15 * 60 * 1_000_000_000),
        metadata=plan.to_metadata(),
    )


accepted = backtest.inspect(db, config("preflight"))
print(f"fidelity {accepted.fidelity}, accepted {accepted.ok}")
for issue in accepted.warnings:
    print(f"  warning {issue.code}: {issue.message}")
try:
    backtest.inspect(db, config("queue", queue_position=True)).raise_for_errors()
except ValueError as error:
    print(f"\nqueue claim refused: {str(error).splitlines()[0]}")

stamps = list(
    db.sql("SELECT DISTINCT ts_init FROM book_deltas ORDER BY ts_init").to_pandas().ts_init
)
edges = [0, 8, 12, 16, 20, 24, 28]
walk = backtest.WalkForward.of(
    *[
        backtest.ValidationWindows(
            train=(stamps[edges[index]], stamps[edges[index + 1]]),
            holdout=(stamps[edges[index + 1]], stamps[edges[index + 2]]),
        )
        for index in range(0, len(edges) - 2, 2)
    ]
)
for index, fold in enumerate(walk.folds):
    print(f"fold {index}: train {fold.train[0].time()}-{fold.train[1].time()} "
          f"holdout {fold.holdout[0].time()}-{fold.holdout[1].time()} "
          f"embargo {fold.embargo_ns / 1e9:.0f}s")

study = backtest.study(
    db,
    study_id="e2e-fees",
    base=config("e2e-fees"),
    parameters={"execution.fee_rate": [0.0, 0.02, 0.04, FEE_RATE]},
    validation=walk,
    selection=backtest.TopK(k=2, metric="realized_pnl"),
)
board = pd.DataFrame(study.ranked())
columns = ["trial", "parameters", "train_median_realized_pnl", "holdout_median_realized_pnl"]
print(board[columns].to_string(index=False))
print(f"\ntrials run: {len(study.trials)}, reached the holdout: {len(study.selected)}")
print(f"attention: {study.attention_state}, unseen warnings: {study.warning_badge}")

for row in sorted(study.trials, key=lambda item: item["trial"]):
    has_holdout = any(key.startswith("fold0_holdout_") for key in row)
    print(f"trial {row['trial']}  selected={row['selected']!s:5} "
          f"holdout_columns={has_holdout}")

absorbed = []
for row in sorted(study.trials, key=lambda item: item["trial"]):
    if row["status"] != "ok":
        continue
    rate = json.loads(row["parameters"])["execution.fee_rate"]
    train = row["train_median_realized_pnl"]
    absorbed.append(
        {
            "fee_rate": rate,
            "train_median_realized_pnl": train,
            "profitable": train > 0.0,
            "reached_holdout": row["selected"],
        }
    )
frame = pd.DataFrame(absorbed)
print(frame.to_string(index=False))
survivors = frame[frame.profitable]
if len(survivors) and len(survivors) < len(frame):
    print(f"\nthe edge survives up to a {survivors.fee_rate.max():.2%} fee and "
          f"not at {frame[~frame.profitable].fee_rate.min():.2%}")
else:
    print("\nno crossover inside the swept range: widen it before concluding")

winner = study.ranked()[0]
best_fee = float(json.loads(winner["parameters"])["execution.fee_rate"])
final = backtest.execute(db, config("e2e-final", fee_rate_override=best_fee))
positions = final.positions.to_pandas()
summary = final.summary()
settled = float(positions.settlement_pnl.fillna(0.0).sum())
marked = float(positions.market_exit_pnl.fillna(0.0).sum())
print(f"winning fee_rate on holdout: {best_fee}")
print(f"fills                 {summary['fills']}")
print(f"realized (round trips){summary['realized_pnl']:>12,.2f}")
print(f"marked to last quote  {marked:>12,.2f}")
print(f"settled at resolution {settled:>12,.2f}")
print(f"settlement adjustment {settled - marked:>+12,.2f}")
print(f"fees                  {summary['commissions']:>12,.2f}")
# realized_pnl is already net of commissions, so the total is realized +
# settlement. Subtracting commissions again double-counts them, and settlement
# alone drops every closed round trip.
print(f"net                   {summary['realized_pnl'] + settled:>12,.2f}")
print(f"settlement applied:   {bool(final.run.to_pandas().settlement_applied.iloc[0])}")

fills = final.fills.to_pandas()
truth = cu.market_truth(panel_tables).to_pandas().set_index("instrument_id")
mids = panel.groupby("instrument_id").apply(
    lambda group: group.sort_values("ts").iloc[len(group) // 2], include_groups=False
)
scored = pd.DataFrame(
    {
        "strategy": fills.groupby("instrument_id").price.mean(),
        "market": ((mids.bid + mids.ask) / 2),
    }
).dropna()
scored["outcome"] = truth.yes_won.astype(float)
scored = scored.dropna()
advantage = quant.brier_advantage(
    scored.strategy.tolist(), scored.market.tolist(), scored.outcome.tolist()
)
print(f"markets scored     {advantage.observations}")
print(f"strategy Brier     {advantage.strategy_brier:.4f}")
print(f"market   Brier     {advantage.market_brier:.4f}")
print(f"advantage          {advantage.advantage:+.4f}")
print(f"skill score        {advantage.skill_score:+.3f}")
print(f"beat the market on {advantage.win_rate:.0%} of markets")

runs = {"final": final}
for row in study.selected:
    fork = row.get("fold0_holdout_fork") or row.get("fold0_train_fork")
    if fork:
        runs[f"trial{row['trial']}"] = backtest.open_result(db, fork.removeprefix("bt-"))
report = quant.basket_payload(
    db,
    runs,
    basket_id="late-favorite basket",
    panels=quant.PORTFOLIO_PANELS + ("equity", "price", "allocation"),
    snapshot="mirror-v1",
    brier={
        "final": {
            "strategy": scored.strategy.tolist(),
            "market": scored.market.tolist(),
            "outcome": scored.outcome.tolist(),
        }
    },
)
print(f"runs      {report.totals['runs']}")
print(f"net       {report.totals['net']:,.2f}")
print(f"drawdown  {report.totals.get('max_drawdown', 0):.4f}")
print(f"panels    {', '.join(report.drawn)}")
if report.skipped:
    print(f"not drawn {[item['panel'] for item in report.skipped]}")
path = report.to_html("data/cache/e2e-basket.html")
print(f"\nwrote {len(path):,} bytes of self-contained HTML")

equity = final.equity.to_pandas().sort_values("ts")
levels = equity.equity.tolist()
returns = [
    (levels[index] - levels[index - 1]) / levels[index - 1]
    for index in range(1, len(levels))
    if levels[index - 1]
]
deflated = quant.deflated_sharpe(returns, trials=len(study.trials))
track = quant.minimum_track_record_length(returns)
print(f"observed Sharpe            {deflated.sharpe:.3f}")
print(f"benchmark after {deflated.trials} trials  {deflated.benchmark:.3f}")
print(f"P(true Sharpe > benchmark) {deflated.probability:.3f}")
if track == float("inf"):
    print("observations needed        no finite length suffices: the observed")
    print("                          Sharpe sits below the deflated benchmark")
else:
    print(f"observations needed        {track:.0f}, available {len(returns)}")
print(f"skew {deflated.skew:+.2f}, excess kurtosis {deflated.kurtosis:.2f}")

fig, axes = plt.subplots(1, 2, figsize=(11, 4))
axes[0].plot(equity.ts, equity.equity, color="#2c7fb8")
axes[0].set_title("Winner equity, mark to settlement")
axes[0].set_xlabel("simulated time")
axes[0].set_ylabel("equity")
axes[0].tick_params(axis="x", labelrotation=30)
axes[1].plot(advantage.cumulative, color="#c0392b")
axes[1].axhline(0.0, color="black", lw=0.8)
axes[1].set_title(f"Cumulative Brier advantage ({advantage.observations} markets)")
axes[1].set_xlabel("market")
axes[1].set_ylabel("market Brier - strategy Brier")
fig.tight_layout()

verified = final.verify()
print(f"verified: {verified['verified']}")
print(f"tables compared: {list(verified['tables_equal'])}")
print(f"all equal: {all(verified['tables_equal'].values())}")
print(f"\nconfig digest {final.config.digest[:16]}")
print(f"trial digest  {final.config.trial_digest[:16]} (excludes run_id and metadata)")
print(f"data pin      {final.config.data.snapshot}")