Replay any 5-minute market

#!/usr/bin/env python3
"""
replay.py — replay a Polymarket crypto up/down market's price action + order
book straight from your terminal, combining two free/open data sources:

  - Vatic API (api.vatic.trading)       -> Chainlink target price + TWAP resolution
  - PendulumFlow archive (archive.pendulumflow.com) -> the real historical order book

Usage:
    python3 replay.py btc-updown-5m-1790244900
    python3 replay.py btc-updown-5m-1790244900 --video     # also render an mp4

No API key needed for either source. Only dependencies: requests, pandas,
pyarrow (all pip-installable). Reads PendulumFlow's remote Parquet file via
plain synchronous HTTP Range requests (no fsspec/async) for reliable timeouts.
"""
import argparse
import json
import re
import sys
import time
from pathlib import Path

import pandas as pd
import requests

sys.stdout.reconfigure(line_buffering=True)

HTTP_TIMEOUT = 30  # seconds per range request; plain `requests`, no async/fsspec involved.


class HTTPRangeFile:
    """
    Minimal synchronous random-access file-like object over HTTP Range requests,
    using plain `requests` (reliable timeouts, no async event loop). This is what
    we hand to pyarrow instead of fsspec's HTTP filesystem, which has shown
    intermittent hangs against this archive that its own timeout config didn't
    catch.
    """

    def __init__(self, url: str, session: requests.Session | None = None):
        self.url = url
        self.session = session or requests.Session()
        head = self.session.head(url, timeout=HTTP_TIMEOUT)
        head.raise_for_status()
        self.size = int(head.headers["Content-Length"])
        self._pos = 0

    def seek(self, offset, whence=0):
        if whence == 0:
            self._pos = offset
        elif whence == 1:
            self._pos += offset
        elif whence == 2:
            self._pos = self.size + offset
        return self._pos

    def tell(self):
        return self._pos

    def read(self, n=-1):
        if n is None or n < 0:
            end = self.size - 1
        else:
            end = min(self._pos + n, self.size) - 1
        if self._pos > end:
            return b""

        last_err = None
        for attempt in range(4):
            try:
                resp = self.session.get(
                    self.url, headers={"Range": f"bytes={self._pos}-{end}"}, timeout=HTTP_TIMEOUT,
                )
                resp.raise_for_status()
                data = resp.content
                if len(data) != (end - self._pos + 1):
                    raise IOError(f"short read: expected {end - self._pos + 1} bytes, got {len(data)}")
                self._pos += len(data)
                return data
            except Exception as err:  # noqa: BLE001 - flaky connection drops from the archive server
                last_err = err
                time.sleep(0.5 * (attempt + 1))
        raise last_err

    def readable(self):
        return True

    def seekable(self):
        return True

    def writable(self):
        return False

    @property
    def closed(self):
        return False

    def close(self):
        pass

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()

GAMMA_BASE = "https://gamma-api.polymarket.com"
VATIC_BASE = "https://api.vatic.trading"
PENDULUM_BASE = "https://archive.pendulumflow.com"

SLUG_RE = re.compile(r"^(?P<asset>[a-z0-9]+)-updown-(?P<interval>\d+[mh]|daily)-(?P<ts>\d+)$")
INTERVAL_TO_MARKET_TYPE = {"5m": "5min", "15m": "15min", "1h": "1hour", "4h": "4hour", "daily": "daily"}
MARKET_SECONDS = {"5min": 300, "15min": 900, "1hour": 3600, "4hour": 14400, "daily": 86400}

RESET, DIM, BOLD = "\x1b[0m", "\x1b[2m", "\x1b[1m"
GREEN, RED, YELLOW, CYAN, BLUE = "\x1b[32m", "\x1b[31m", "\x1b[33m", "\x1b[36m", "\x1b[34m"
BLOCKS = " ▁▂▃▄▅▆▇█"


def sparkline(values, width=60, color=""):
    vals = [v for v in values if v is not None]
    if not vals:
        return ""
    lo, hi = min(vals), max(vals)
    span = (hi - lo) or 1.0
    # Resample to `width` buckets by simple stride sampling.
    n = len(values)
    idxs = [int(i * (n - 1) / max(width - 1, 1)) for i in range(width)]
    chars = []
    for i in idxs:
        v = values[i]
        if v is None:
            chars.append(" ")
            continue
        level = int(((v - lo) / span) * (len(BLOCKS) - 1))
        chars.append(BLOCKS[level])
    return f"{color}{''.join(chars)}{RESET}"


def build_time_grid(times_values, t_min, t_max, width):
    """Resample a (t, value) series onto `width` evenly-spaced columns over [t_min, t_max],
    using the last-known value at or before each column's time (step interpolation)."""
    import bisect

    pairs = sorted(((t, v) for t, v in times_values if v is not None), key=lambda tv: tv[0])
    ts = [p[0] for p in pairs]
    vals = [p[1] for p in pairs]
    grid_times = [t_min + (t_max - t_min) * i / max(width - 1, 1) for i in range(width)]
    grid_vals = []
    for gt in grid_times:
        idx = bisect.bisect_right(ts, gt) - 1
        grid_vals.append(vals[idx] if idx >= 0 else None)
    return grid_times, grid_vals


def partial_sparkline(grid_vals, revealed_count, lo, hi, color=""):
    span = (hi - lo) or 1.0
    chars = []
    for i, v in enumerate(grid_vals):
        if i >= revealed_count or v is None:
            chars.append(" ")
            continue
        level = int(((v - lo) / span) * (len(BLOCKS) - 1))
        level = max(0, min(level, len(BLOCKS) - 1))
        chars.append(BLOCKS[level])
    return f"{color}{''.join(chars)}{RESET}"


def format_depth_ladder(bids, asks, n, ask_color, bid_color):
    """Real trading-ladder layout: worst asks at top, best ask just above the
    spread, best bid just below the spread, worst bids at the bottom. Always
    returns exactly 2n lines so the animation's line count stays constant."""
    ask_levels = top_levels(asks, n)  # best-to-worst
    ask_display = list(reversed(ask_levels))  # worst-to-best, top-to-bottom
    missing_asks = n - len(ask_display)
    ask_display = [None] * missing_asks + ask_display  # pad the far (worst) end

    bid_levels = top_levels(bids, n)  # best-to-worst, top-to-bottom already
    missing_bids = n - len(bid_levels)
    bid_display = bid_levels + [None] * missing_bids  # pad the far (worst) end

    def row(lv, label, color):
        if lv is None:
            return ""
        return f"  {color}{label}{RESET}  ${lv['price']:.2f}   x {lv['size']:>9,.0f}"

    lines = [row(lv, "ASK", ask_color) for lv in ask_display]
    lines += [row(lv, "BID", bid_color) for lv in bid_display]
    return lines


def value_at_or_before(times_values, t):
    import bisect

    pairs = sorted(((tv[0], tv[1]) for tv in times_values if tv[1] is not None), key=lambda tv: tv[0])
    if not pairs:
        return None
    ts = [p[0] for p in pairs]
    idx = bisect.bisect_right(ts, t) - 1
    return pairs[idx][1] if idx >= 0 else None


def parse_slug(slug: str):
    m = SLUG_RE.match(slug)
    if not m:
        raise SystemExit(
            f"Could not parse slug '{slug}'. Expected format like 'btc-updown-5m-1790244900'."
        )
    interval = m.group("interval")
    market_type = INTERVAL_TO_MARKET_TYPE.get(interval)
    if not market_type:
        raise SystemExit(f"Unsupported interval '{interval}' in slug '{slug}'.")
    return m.group("asset"), market_type, int(m.group("ts"))


def fetch_gamma_event(slug: str):
    resp = requests.get(f"{GAMMA_BASE}/events/slug/{slug}", timeout=30)
    resp.raise_for_status()
    data = resp.json()
    market = data["markets"][0]
    token_ids = json.loads(market["clobTokenIds"])
    outcomes = json.loads(market["outcomes"])
    return {
        "conditionId": market["conditionId"],
        "tokenIds": dict(zip(outcomes, token_ids)),
        "closed": market.get("closed", False),
        "outcomePrices": json.loads(market.get("outcomePrices", "[]")),
        "outcomes": outcomes,
    }


def resolved_outcome_from_gamma(gamma_event):
    prices = gamma_event["outcomePrices"]
    outcomes = gamma_event["outcomes"]
    if not gamma_event["closed"] or not prices:
        return None
    for outcome, price in zip(outcomes, prices):
        if float(price) >= 0.5:
            return outcome
    return None


def fetch_vatic_resolve(asset, market_type, window_start):
    resp = requests.get(
        f"{VATIC_BASE}/api/v1/twap/resolve",
        params={"asset": asset, "type": market_type, "marketStart": window_start},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


def fetch_vatic_price_series(asset, window_start, window_end, step_seconds=5, pad_seconds=30, workers=15):
    from concurrent.futures import ThreadPoolExecutor, as_completed

    start = window_start - pad_seconds
    end = window_end + pad_seconds
    timestamps = list(range(start, end + 1, step_seconds))
    session = requests.Session()

    def fetch_one(ts):
        try:
            resp = session.get(
                f"{VATIC_BASE}/api/v1/twap/report",
                params={"asset": asset, "window": 30, "timestamp": ts},
                timeout=30,
            )
            return ts, (float(resp.json()["value"]) if resp.ok else None)
        except requests.RequestException:
            return ts, None

    results = {}
    done = 0
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = [pool.submit(fetch_one, ts) for ts in timestamps]
        for future in as_completed(futures):
            ts, price = future.result()
            results[ts] = price
            done += 1
            if done % 20 == 0 or done == len(timestamps):
                print(f"{DIM}  ...{done}/{len(timestamps)} price samples fetched{RESET}")

    return [{"timestamp": ts, "price": results[ts]} for ts in timestamps]


def hour_urls_for_window(window_start, window_end):
    import datetime as dt
    start_hour = dt.datetime.fromtimestamp(window_start, tz=dt.timezone.utc).replace(minute=0, second=0, microsecond=0)
    end_hour = dt.datetime.fromtimestamp(window_end, tz=dt.timezone.utc).replace(minute=0, second=0, microsecond=0)
    hours = []
    h = start_hour
    while h <= end_hour:
        date_str = h.strftime("%Y-%m-%d")
        hour_str = h.strftime("%H")
        hours.append((h, f"{PENDULUM_BASE}/v3/{date_str}/{hour_str}/{date_str}T{hour_str}.parquet"))
        h += dt.timedelta(hours=1)
    return hours


def fetch_pendulum_events(condition_id_hex: str, window_start: int, window_end: int, retries: int = 3):
    import time

    import pyarrow.parquet as pq

    market_bytes = bytes.fromhex(condition_id_hex.removeprefix("0x"))
    columns = [
        "event_type", "timestamp", "market", "asset_id",
        "best_bid", "best_ask", "spread", "price", "size",
        "bids", "asks",
        "winning_asset_id", "winning_outcome",
    ]

    frames = []
    for _, url in hour_urls_for_window(window_start, window_end):
        last_err = None
        for attempt in range(retries):
            try:
                resp = requests.head(url, timeout=HTTP_TIMEOUT)
                if resp.status_code == 404:
                    last_err = None
                    break
                resp.raise_for_status()
                with HTTPRangeFile(url) as f:
                    table = pq.read_table(f, columns=columns, filters=[("market", "=", market_bytes)], use_threads=False)
                frames.append(table.to_pandas())
                last_err = None
                break
            except Exception as err:  # noqa: BLE001 - transient network/HTTP flakiness from the archive
                last_err = err
                if attempt < retries - 1:
                    time.sleep(2 * (attempt + 1))
        if last_err is not None:
            raise last_err
    if not frames:
        return pd.DataFrame(columns=columns)
    df = pd.concat(frames, ignore_index=True)
    from decimal import Decimal

    def levels_to_float(levels):
        if levels is None:
            return None
        try:
            return [{"price": float(lv["price"]), "size": float(lv["size"])} for lv in levels]
        except (TypeError, KeyError):
            return None

    for col in ["best_bid", "best_ask", "spread", "price", "size"]:
        if col in df.columns:
            df[col] = df[col].apply(lambda v: float(v) if isinstance(v, Decimal) else v)
    for col in ["bids", "asks"]:
        if col in df.columns:
            df[col] = df[col].apply(levels_to_float)
    df["asset_id_hex"] = df["asset_id"].apply(lambda b: b.hex() if isinstance(b, (bytes, bytearray)) else None)
    return df.sort_values("timestamp").reset_index(drop=True)


def top_levels(levels, n=3):
    """Best-to-worst top n levels. Archive stores bids ascending / asks descending
    by price, with the best (touch) price as the LAST element in both cases."""
    if not levels:
        return []
    return list(reversed(levels))[:n]


def compute_strategy(price_points, book_df, window_start, target_price, resolved_outcome, signal_window_seconds=40):
    early_prices = [p for p in price_points if 0 <= p["timestamp"] - window_start <= signal_window_seconds and p["price"] is not None]
    if not early_prices:
        return None
    btc_edge = early_prices[-1]["price"] - target_price

    book_df = book_df.copy()
    book_df["t"] = (book_df["timestamp"].astype("int64") // 1000) - window_start
    early_book = book_df[(book_df["t"] >= 0) & (book_df["t"] <= signal_window_seconds)]
    if early_book.empty:
        return None
    cheapest = early_book.loc[early_book["best_ask"].idxmin()]
    entry_ask = float(cheapest["best_ask"])
    payout = 1.0 if resolved_outcome == "Up" else 0.0
    pnl_pct = ((payout - entry_ask) / entry_ask) * 100 if entry_ask else None
    return {
        "btcEdgeVsTargetUsd": btc_edge,
        "cheapestUpAskSecondsAfterOpen": int(cheapest["t"]),
        "cheapestUpAsk": entry_ask,
        "payoutPerShare": payout,
        "pnlPct": pnl_pct,
    }


def print_report(slug, asset, market_type, window_start, window_end, resolve, resolved_outcome, price_points, book_df, depth_df, strategy, depth_levels=5):
    print()
    print(f"{BOLD}{CYAN}== {slug} =={RESET}")
    print(f"{DIM}{asset.upper()} {market_type} window: "
          f"{pd.Timestamp(window_start, unit='s', tz='UTC')} -> {pd.Timestamp(window_end, unit='s', tz='UTC')}{RESET}")
    print()

    target = resolve.get("targetPrice")
    twap_close = resolve.get("resolutionTwap")
    print(f"  target (open, Vatic):     ${target:,.2f}" if target else "  target: n/a")
    if twap_close:
        print(f"  TWAP close (Vatic):       ${float(twap_close):,.2f}")
    if resolved_outcome:
        color = GREEN if resolved_outcome == "Up" else RED
        print(f"  resolved:                 {color}{BOLD}{resolved_outcome.upper()}{RESET}")
    print()

    prices = [p["price"] for p in price_points]
    print(f"  {BOLD}BTC/USD (Vatic Chainlink TWAP){RESET}")
    print(f"  {sparkline(prices, color=YELLOW)}")
    if prices:
        valid = [p for p in prices if p is not None]
        print(f"  {DIM}min ${min(valid):,.2f}  max ${max(valid):,.2f}{RESET}")
    print()

    up_book = book_df[(book_df["event_type"] == "best_bid_ask")]
    if up_book is not None and not up_book.empty:
        asks = up_book["best_ask"].tolist()
        print(f"  {BOLD}Polymarket \"Up\" share — best ask (PendulumFlow order book){RESET}")
        print(f"  {sparkline(asks, color=BLUE)}")
        if asks:
            print(f"  {DIM}open ${asks[0]:.2f}  close ${asks[-1]:.2f}{RESET}")
    print()

    if depth_df is not None and not depth_df.empty:
        depth_times = (depth_df["timestamp"].astype("int64") // 1000) - window_start
        asks_tv = list(zip(depth_times, depth_df["asks"]))
        bids_tv = list(zip(depth_times, depth_df["bids"]))
        # Show depth at the strategy-entry moment (a genuinely two-sided book) rather than
        # the final snapshot, which is often one-sided by the time a market fully resolves.
        snapshot_t = strategy["cheapestUpAskSecondsAfterOpen"] if strategy else (window_end - window_start)
        snap_asks = value_at_or_before(asks_tv, snapshot_t)
        snap_bids = value_at_or_before(bids_tv, snapshot_t)
        print(f"  {BOLD}Order book depth at T+{snapshot_t}s (top {depth_levels}){RESET}")
        ladder = format_depth_ladder(snap_bids, snap_asks, depth_levels, RED, GREEN)
        mid = depth_levels
        for line in ladder[:mid]:
            print(line)
        print(f"  {DIM}{'-' * 10} spread {'-' * 10}{RESET}")
        for line in ladder[mid:]:
            print(line)
        print()

    if strategy:
        print(f"  {BOLD}{YELLOW}illustrative example (not trading advice):{RESET}")
        print(f"    at T+{strategy['cheapestUpAskSecondsAfterOpen']}s, BTC was already "
              f"${strategy['btcEdgeVsTargetUsd']:.0f} above target (Vatic),")
        print(f"    while \"Up\" shares were still asking ${strategy['cheapestUpAsk']:.2f} (order book).")
        pct = strategy["pnlPct"]
        pct_color = GREEN if pct and pct > 0 else RED
        print(f"    held to resolution: {pct_color}{BOLD}{pct:+.0f}%{RESET}")
    print()


def animate_report(slug, asset, market_type, window_start, window_end, resolve,
                    resolved_outcome, price_points, book_df, depth_df, strategy,
                    speed=2.0, fps=10, width=60, pad_seconds=30, depth_levels=5):
    target = resolve.get("targetPrice")
    t_min, t_max = -pad_seconds, (window_end - window_start) + pad_seconds

    price_tv = [(p["timestamp"] - window_start, p["price"]) for p in price_points]
    price_vals_all = [v for _, v in price_tv if v is not None]
    p_lo, p_hi = (min(price_vals_all), max(price_vals_all)) if price_vals_all else (0, 1)
    price_grid_times, price_grid_vals = build_time_grid(price_tv, t_min, t_max, width)

    book_tv = []
    if not book_df.empty:
        book_tv = list(zip(
            (book_df["timestamp"].astype("int64") // 1000) - window_start,
            book_df["best_ask"],
        ))
    ask_vals_all = [v for _, v in book_tv if v is not None]
    a_lo, a_hi = (min(ask_vals_all), max(ask_vals_all)) if ask_vals_all else (0, 1)
    ask_grid_times, ask_grid_vals = build_time_grid(book_tv, t_min, t_max, width)

    depth_bids_tv, depth_asks_tv = [], []
    if not depth_df.empty:
        depth_times = (depth_df["timestamp"].astype("int64") // 1000) - window_start
        depth_bids_tv = list(zip(depth_times, depth_df["bids"]))
        depth_asks_tv = list(zip(depth_times, depth_df["asks"]))

    total_playback_seconds = (t_max - t_min) / speed
    total_frames = max(1, int(total_playback_seconds * fps))
    frame_times = [t_min + (t_max - t_min) * i / max(total_frames - 1, 1) for i in range(total_frames)]

    entry_shown_from = strategy["cheapestUpAskSecondsAfterOpen"] if strategy else None

    def fmt_clock(t):
        sign = "" if t >= 0 else "-"
        return f"T{sign}{abs(int(t)):>3d}s"

    def render_lines(t_now, final=False):
        revealed_price = sum(1 for gt in price_grid_times if gt <= t_now)
        revealed_ask = sum(1 for gt in ask_grid_times if gt <= t_now)
        cur_price = value_at_or_before(price_tv, t_now)
        cur_ask = value_at_or_before(book_tv, t_now)

        lines = []
        lines.append(f"{BOLD}{CYAN}== {slug} =={RESET}   {DIM}{fmt_clock(t_now)}{RESET}")
        lines.append(f"{DIM}{asset.upper()} {market_type} window: "
                      f"{pd.Timestamp(window_start, unit='s', tz='UTC')} -> {pd.Timestamp(window_end, unit='s', tz='UTC')}{RESET}")
        lines.append("")
        if target:
            diff = (cur_price - target) if cur_price is not None else None
            diff_str = f"   ({'+' if diff and diff >= 0 else ''}{diff:,.2f} vs target)" if diff is not None else ""
            cur_str = f"${cur_price:,.2f}" if cur_price is not None else "..."
            lines.append(f"  target (open, Vatic): ${target:,.2f}    current BTC: {cur_str}{diff_str}")
        else:
            lines.append("  target: n/a")
        lines.append("")
        lines.append(f"  {BOLD}BTC/USD (Vatic Chainlink TWAP){RESET}")
        lines.append(f"  {partial_sparkline(price_grid_vals, revealed_price, p_lo, p_hi, color=YELLOW)}")
        lines.append("")
        lines.append(f"  {BOLD}Polymarket \"Up\" share — best ask (PendulumFlow order book){RESET}")
        ask_str = f"${cur_ask:.2f}" if cur_ask is not None else "..."
        lines.append(f"  {partial_sparkline(ask_grid_vals, revealed_ask, a_lo, a_hi, color=BLUE)}   current: {ask_str}")
        lines.append("")

        cur_asks = value_at_or_before(depth_asks_tv, t_now)
        cur_bids = value_at_or_before(depth_bids_tv, t_now)
        lines.append(f"  {BOLD}Order book depth (top {depth_levels}){RESET}")
        ladder = format_depth_ladder(cur_bids, cur_asks, depth_levels, RED, GREEN)
        mid = depth_levels
        lines.extend(ladder[:mid])
        lines.append(f"  {DIM}{'-' * 10} spread {'-' * 10}{RESET}")
        lines.extend(ladder[mid:])
        lines.append("")

        if strategy and t_now >= entry_shown_from:
            pct = strategy["pnlPct"]
            pct_color = GREEN if pct and pct > 0 else RED
            lines.append(
                f"  {YELLOW}at T+{entry_shown_from}s: BTC ${strategy['btcEdgeVsTargetUsd']:+.0f} vs target, "
                f"\"Up\" asking ${strategy['cheapestUpAsk']:.2f} -> held to resolution: "
                f"{pct_color}{BOLD}{pct:+.0f}%{RESET}"
            )
        else:
            lines.append("")

        if final and resolved_outcome:
            color = GREEN if resolved_outcome == "Up" else RED
            twap_close = resolve.get("resolutionTwap")
            twap_str = f", TWAP close ${float(twap_close):,.2f}" if twap_close else ""
            lines.append(f"  {color}{BOLD}RESOLVED: {resolved_outcome.upper()}{RESET}{twap_str}")
        else:
            lines.append("")
        return lines

    prev_line_count = 0
    start_wall = time.time()
    for i, t_now in enumerate(frame_times):
        target_wall = start_wall + i / fps
        sleep_for = target_wall - time.time()
        if sleep_for > 0:
            time.sleep(sleep_for)

        is_final = i == len(frame_times) - 1
        lines = render_lines(t_now, final=is_final)

        out = []
        if prev_line_count:
            out.append(f"\x1b[{prev_line_count}A")
        for line in lines:
            out.append("\x1b[2K" + line + "\n")
        sys.stdout.write("".join(out))
        sys.stdout.flush()
        prev_line_count = len(lines)

    print()


def main():
    parser = argparse.ArgumentParser(description="Replay a Polymarket crypto up/down market from Vatic + PendulumFlow.")
    parser.add_argument("slug", nargs="?", default="btc-updown-5m-1790244900",
                         help="Polymarket event slug, e.g. btc-updown-5m-1790244900")
    parser.add_argument("--signal-window-seconds", type=int, default=40)
    parser.add_argument("--step-seconds", type=int, default=5)
    parser.add_argument("--video", action="store_true", help="Also render output/<slug>.mp4 (needs matplotlib+ffmpeg)")
    parser.add_argument("--data-dir", default="data")
    parser.add_argument("--instant", action="store_true", help="Skip the live terminal animation, print the final report immediately.")
    parser.add_argument("--speed", type=float, default=2.0, help="Playback speed multiplier for the live animation (default 2x).")
    parser.add_argument("--fps", type=int, default=10, help="Animation frame rate.")
    parser.add_argument("--width", type=int, default=60, help="Sparkline width in characters.")
    parser.add_argument("--depth-levels", type=int, default=5, help="Number of order-book levels to show on each side.")
    args = parser.parse_args()

    asset, market_type, window_start = parse_slug(args.slug)
    window_end = window_start + MARKET_SECONDS[market_type]

    print(f"{DIM}Resolving {args.slug} via Polymarket Gamma...{RESET}")
    gamma_event = fetch_gamma_event(args.slug)
    resolved_outcome = resolved_outcome_from_gamma(gamma_event)

    print(f"{DIM}Fetching target/TWAP from Vatic...{RESET}")
    resolve = fetch_vatic_resolve(asset, market_type, window_start)

    print(f"{DIM}Sampling continuous price series from Vatic...{RESET}")
    price_points = fetch_vatic_price_series(asset, window_start, window_end, step_seconds=args.step_seconds)

    print(f"{DIM}Fetching order book from PendulumFlow...{RESET}")
    events_df = fetch_pendulum_events(gamma_event["conditionId"], window_start, window_end)

    up_token_hex = format(int(gamma_event["tokenIds"]["Up"]), "x").zfill(64) if "Up" in gamma_event["tokenIds"] else None
    book_df = events_df[
        (events_df["event_type"] == "best_bid_ask") & (events_df["asset_id_hex"] == up_token_hex)
    ].sort_values("timestamp") if up_token_hex else pd.DataFrame()
    depth_df = events_df[
        (events_df["event_type"] == "book") & (events_df["asset_id_hex"] == up_token_hex)
    ].sort_values("timestamp") if up_token_hex else pd.DataFrame()

    strategy = compute_strategy(
        price_points, book_df, window_start, resolve.get("targetPrice"), resolved_outcome,
        signal_window_seconds=args.signal_window_seconds,
    ) if not book_df.empty else None

    print()
    if args.instant:
        print_report(args.slug, asset, market_type, window_start, window_end, resolve, resolved_outcome, price_points, book_df, depth_df, strategy)
    else:
        animate_report(
            args.slug, asset, market_type, window_start, window_end, resolve, resolved_outcome,
            price_points, book_df, depth_df, strategy, speed=args.speed, fps=args.fps, width=args.width,
            depth_levels=args.depth_levels,
        )

    data_dir = Path(args.data_dir)
    data_dir.mkdir(parents=True, exist_ok=True)
    (data_dir / f"{args.slug}.json").write_text(json.dumps({
        "slug": args.slug, "asset": asset, "marketType": market_type,
        "windowStart": window_start, "windowEnd": window_end,
        "conditionId": gamma_event["conditionId"], "tokenIds": gamma_event["tokenIds"],
        "resolvedOutcome": resolved_outcome, "resolve": resolve,
        "strategy": strategy,
    }, indent=2))

    if args.video:
        print(f"{DIM}--video not wired into this script yet — use render_video.py with this market's data.{RESET}")


if __name__ == "__main__":
    import os

    try:
        main()
    except Exception:  # noqa: BLE001 - print, then hard-exit so no lingering fsspec/asyncio threads hang the process
        import traceback
        traceback.print_exc()
        os._exit(1)
    os._exit(0)