""" Export an interactive orderbook replay viewer for one recorded LPL/LOL market. Example: python3 scripts/export_lpl_orderbook_viewer.py \ --slug lol-tes-al-2026-05-01-game2 \ --depth 12 """ from __future__ import annotations import html import json import os import sys from datetime import datetime, timezone from pathlib import Path from typing import Any import click from loguru import logger ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if ROOT not in sys.path: sys.path.insert(0, ROOT) from analysis.safe_btc5.types import parse_dt # noqa: E402 def _parse_time(value: Any) -> datetime | None: parsed = parse_dt(value) if parsed is None: return None return parsed.astimezone(timezone.utc) def _levels(rows: Any, *, depth: int, reverse: bool) -> list[dict[str, float]]: if not isinstance(rows, list): return [] levels: list[dict[str, float]] = [] for row in rows: try: price = float(row["price"]) size = float(row["size"]) except (KeyError, TypeError, ValueError): continue if size <= 0: continue levels.append({"price": price, "size": size}) levels.sort(key=lambda item: item["price"], reverse=reverse) return levels[:depth] def _load_last_trades(events_path: Path) -> list[dict[str, Any]]: trades: list[dict[str, Any]] = [] with events_path.open("r", encoding="utf-8") as f: for line in f: if not line.strip(): continue row = json.loads(line) if row.get("event_type") != "last_trade_price": continue ts = _parse_time(row.get("received_at_wall")) raw = row.get("raw_json") or {} if ts is None: continue try: price = float(raw.get("price")) size = float(raw.get("size") or 0) except (TypeError, ValueError): continue trades.append( { "ts": ts, "asset_id": str(row.get("asset_id") or raw.get("asset_id") or ""), "price": price, "size": size, "side": str(raw.get("side") or ""), } ) trades.sort(key=lambda item: item["ts"]) return trades def export_orderbook_viewer( *, slug: str, data_root: Path, output_path: Path | None, depth: int, frame_stride: int, ) -> dict[str, Any]: market_dir = data_root / slug meta_path = market_dir / "recording_meta.json" snapshots_path = market_dir / "orderbook_snapshots.jsonl" events_path = market_dir / "orderbook_events.jsonl" if not meta_path.exists(): raise FileNotFoundError(meta_path) if not snapshots_path.exists(): raise FileNotFoundError(snapshots_path) if not events_path.exists(): raise FileNotFoundError(events_path) meta = json.loads(meta_path.read_text(encoding="utf-8")) token_ids = [str(item) for item in meta.get("token_ids", [])] outcomes = [str(item) for item in meta.get("outcomes", [])] token_to_outcome = { token_id: outcomes[idx] if idx < len(outcomes) else token_id for idx, token_id in enumerate(token_ids) } trades = _load_last_trades(events_path) trade_index = 0 last_trade_by_asset: dict[str, dict[str, Any]] = {} latest_book: dict[str, dict[str, Any]] = {} frames: list[dict[str, Any]] = [] candidate_frame = 0 with snapshots_path.open("r", encoding="utf-8") as f: for line in f: if not line.strip(): continue row = json.loads(line) ts = _parse_time(row.get("snapshot_at_wall")) asset_id = str(row.get("asset_id") or "") if ts is None or not asset_id: continue while trade_index < len(trades) and trades[trade_index]["ts"] <= ts: trade = trades[trade_index] last_trade_by_asset[trade["asset_id"]] = { "price": trade["price"], "size": trade["size"], "side": trade["side"], "ts": trade["ts"].isoformat(), } trade_index += 1 bid_levels = _levels(row.get("bid_levels"), depth=depth, reverse=True) ask_levels = _levels(row.get("ask_levels"), depth=depth, reverse=False) latest_book[asset_id] = { "asset_id": asset_id, "outcome": str(row.get("outcome") or token_to_outcome.get(asset_id, "")), "best_bid": row.get("best_bid"), "best_ask": row.get("best_ask"), "bids": bid_levels, "asks": ask_levels, "last_trade": last_trade_by_asset.get(asset_id), } if not token_ids or not all(token_id in latest_book for token_id in token_ids): continue candidate_frame += 1 if candidate_frame % max(1, frame_stride) != 0: continue books = { token_id: json.loads(json.dumps(latest_book[token_id])) for token_id in token_ids } frames.append( { "t": ts.isoformat(), "message_index": row.get("message_index"), "books": books, } ) if output_path is None: output_path = market_dir / "orderbook_replay.html" output_path.parent.mkdir(parents=True, exist_ok=True) payload = { "meta": meta, "token_to_outcome": token_to_outcome, "trades": [ { "ts": trade["ts"].isoformat(), "asset_id": trade["asset_id"], "outcome": token_to_outcome.get(trade["asset_id"], trade["asset_id"]), "price": trade["price"], "size": trade["size"], "side": trade["side"], } for trade in trades ], "frames": frames, "depth": depth, "frame_stride": frame_stride, "generated_at": datetime.now(timezone.utc).isoformat(), } output_path.write_text(_render_html(payload), encoding="utf-8") return { "slug": slug, "frames": len(frames), "output_path": str(output_path), "started_at": frames[0]["t"] if frames else "", "ended_at": frames[-1]["t"] if frames else "", } def _render_html(payload: dict[str, Any]) -> str: data = ( json.dumps(payload, ensure_ascii=False, separators=(",", ":")) .replace(" __TITLE__

Order Book Replay

""" return template.replace("__TITLE__", title).replace("__DATA__", data) @click.command() @click.option("--slug", required=True, help="Recorded Polymarket market slug") @click.option( "--data-root", default="data/lpl", show_default=True, type=click.Path(file_okay=False, path_type=Path), help="Recording data root", ) @click.option( "--output", "output_path", default=None, type=click.Path(dir_okay=False, path_type=Path), help="Output HTML path. Defaults to data/lpl//orderbook_replay.html", ) @click.option("--depth", default=5, show_default=True, type=int, help="Levels per side") @click.option( "--frame-stride", default=1, show_default=True, type=int, help="Keep every Nth reconstructed snapshot frame", ) def main(slug: str, data_root: Path, output_path: Path | None, depth: int, frame_stride: int) -> None: result = export_orderbook_viewer( slug=slug, data_root=data_root, output_path=output_path, depth=depth, frame_stride=frame_stride, ) logger.info( "viewer exported: " f"frames={result['frames']} " f"started_at={result['started_at']} " f"ended_at={result['ended_at']} " f"output={result['output_path']}" ) if __name__ == "__main__": main()