""" Export an event-driven orderbook replay viewer for a recorded LPL/LOL market. This viewer is built from orderbook_events.jsonl, not periodic snapshots. It is intended for microstructure analysis around video-marked game events. """ from __future__ import annotations import html import json import os import sys from datetime import datetime, timezone from pathlib import Path 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.lpl_orderbook.event_replay import iter_event_frames # noqa: E402 def _safe_json(value: object) -> str: return ( json.dumps(value, ensure_ascii=False, separators=(",", ":")) .replace(" dict[str, object]: meta, frames = iter_event_frames( slug=slug, data_root=data_root, depth=depth, market_kind=market_kind, ) if frame_stride > 1: frames = frames[::frame_stride] market_dir = data_root / slug if output_path is None: suffix = f"_{market_kind}" if market_kind else "" output_path = market_dir / f"event_orderbook_replay{suffix}.html" output_path.parent.mkdir(parents=True, exist_ok=True) payload = { "meta": meta, "frames": frames, "depth": depth, "frame_stride": frame_stride, "market_kind": market_kind or "", "time_basis": { "primary": "received_at_wall", "secondary": "exchange_ts", "delay": "received_at_wall - exchange_ts", }, "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]["received_at_wall"] if frames else "", "ended_at": frames[-1]["received_at_wall"] if frames else "", } def _render_html(payload: dict[str, object]) -> str: data = _safe_json(payload) title = html.escape(str((payload.get("meta") or {}).get("question") or "Event Replay")) template = """ __TITLE__

Event 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//event_orderbook_replay.html", ) @click.option("--depth", default=5, show_default=True, type=int, help="Levels per side") @click.option( "--market-kind", default=None, type=click.Choice(["moneyline", "game1", "game2", "game3", "game4"]), help="Export one target market from a unified event recording.", ) @click.option( "--frame-stride", default=1, show_default=True, type=int, help="Keep every Nth event frame. Use this only for very large browser payloads.", ) def main( slug: str, data_root: Path, output_path: Path | None, depth: int, market_kind: str | None, frame_stride: int, ) -> None: result = export_event_viewer( slug=slug, data_root=data_root, output_path=output_path, depth=depth, market_kind=market_kind, frame_stride=frame_stride, ) logger.info( "event 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()