> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vatic.trading/llms.txt
> Use this file to discover all available pages before exploring further.
# Replay any 5-minute market
> Combine Vatic's Chainlink price truth with PendulumFlow's open order-book archive to replay a closed Polymarket crypto up/down market — price action and full order book, side by side.
Vatic tells you *what the price was* and *how a market resolved* (target, TWAP close, direction). It doesn't store the order book itself. [PendulumFlow](https://archive.pendulumflow.com) is a free, open, no-auth archive of Polymarket's full historical order book (bids, asks, trades) going back to February 2026.
Put the two together and you can replay any closed 5-minute (or 15-minute, 1-hour, 4-hour) crypto up/down market: the real BTC/ETH/SOL price line next to the real order book, second by second, ending on the actual resolution.
## See it in action
A real closed BTC 5-minute market, replayed live in the terminal at 2x speed — Chainlink price truth from Vatic on top, the real order book (5 levels each side) from PendulumFlow below.
## Three ways to get this data
Pick based on what you're actually building:
1. **Just want the combined JSON, no code to run?** Use our hosted endpoint: [`GET /api/v1/replay/:slug`](/api-reference/replay/replay). It does the exact same fetch server-side (Vatic price/TWAP truth + PendulumFlow order book) and hands you the result. It's a polling endpoint — fetch time varies too much by market to block on — so a first request returns `status: "pending"` and you poll the same URL again after `retryAfterSeconds` until you get `status: "ready"`.
2. **Want the live terminal animation** (order book + price line playing out at 2x speed, like the video above)? Run the [reference implementation](#reference-implementation) below yourself — it talks to PendulumFlow's raw Parquet archive directly plus Vatic's endpoints, and renders the terminal UI. There's no hosted version of the animation itself, only the underlying data (`/api/v1/replay/:slug` above).
3. **Only need Vatic's side** (price-to-beat / TWAP resolution, no order book at all)? Skip PendulumFlow entirely and call Vatic's own primitives directly: [`/api/v1/targets/timestamp`](/api-reference/targets/timestamp) for the opening target, [`/api/v1/twap/resolve`](/api-reference/twap/resolve) for target + snapshot + TWAP close in one call, or [`/api/v1/twap/report`](/api-reference/twap/report) to sample a continuous price series across a window (see [Chainlink TWAP](/concepts/chainlink-twap) for which `window` to use).
If you're pointing a coding agent at this page: give it the [agent prompt](#give-this-to-your-coding-agent) below when you want it to build the *animation* from scratch (PendulumFlow + raw Parquet), or just tell it to call `/api/v1/replay/:slug` directly when you only need the data.
## Give this to your coding agent
The block below is a complete, self-contained spec — paste it into Claude Code, Cursor, or any coding agent and it has everything needed to build a working replay tool without rediscovering the gotchas (byte encodings, ordering conventions, a TWAP window edge case, and an HTTP client that hangs on this archive).
```text theme={null}
Build a script that replays a Polymarket crypto up/down market (e.g. slug
"btc-updown-5m-1790244900") by combining two free, public data sources:
1. VATIC API (https://api.vatic.trading, no auth, no key needed)
- GET /api/v1/twap/resolve?asset=btc&type=5min&marketStart=
Returns targetPrice (opening price-to-beat), resolutionTwap (the actual
Chainlink TWAP used to resolve the market), and snapshotClosePrice.
Only works for markets that have already closed.
- GET /api/v1/twap/report?asset=btc&window=60×tamp=
Returns the Chainlink TWAP ending at that exact timestamp - call this
repeatedly (every 5s) across the window to build a continuous price line,
not just two points. window must be 60 for markets on/after 2026-08-14,
30 before that date (5-minute markets migrated from a 30s to 60s TWAP on
2026-08-14T00:00:00Z - pass the correct one or you'll get a value that's
off by a fraction of a percent and won't match what the market actually
resolved on).
- Parallelize the repeated /twap/report calls (e.g. 15 concurrent requests)
- sequential calls make this step take 20-30s for no reason, Vatic's
rate limit is 120 req/60s so this is well within budget.
2. POLYMARKET GAMMA API (https://gamma-api.polymarket.com, no auth)
- GET /events/slug/ to resolve the market slug into:
- conditionId (0x... hex string)
- clobTokenIds (array of two big decimal-integer strings: [Up, Down])
- outcomePrices + closed (to know how it resolved)
3. PENDULUMFLOW ARCHIVE (https://archive.pendulumflow.com, no auth, no key,
no rate limit - it's free and donation-funded, credit it if you use it)
- Data for hour on date (UTC) lives at one URL:
https://archive.pendulumflow.com/v3///T.parquet
(v3 is PendulumFlow's own capture, covers 2026-08-18 onward and is the
highest-quality corpus - use it, don't reach for pmxt/v1 or v2 unless
you need an earlier date)
- This is ONE Parquet file per hour but with a UNION schema: every event
type (best_bid_ask, book, last_trade_price, price_change,
market_resolved, new_market, tick_size_change) lives in the same file,
discriminated by an `event_type` column, with unused columns null per row.
- Do NOT download the whole file (it can be 1GB+). Use a Parquet reader
that supports column projection + row-group filter pushdown (e.g.
pyarrow's read_table(columns=[...], filters=[("market","=",...)])) so
only the relevant row groups are fetched.
- `market` and `asset_id` columns are raw big-endian BYTES, not hex
strings: convert conditionId via bytes.fromhex(conditionId[2:]), and
each clobTokenId via bytes.fromhex(format(int(token_id), 'x').zfill(64)).
Filtering by the hex-string form directly will silently match nothing.
- The `book` event type carries full depth: `bids` and `asks` columns are
each a list of {price, size}. Both arrays are ordered so the BEST price
(closest to the spread) is the LAST element - bids ascend low-to-high,
asks descend high-to-low. Reverse as needed to print a normal ladder.
- Do NOT use fsspec's async HTTP filesystem to read the remote Parquet
file - it has an intermittent hang against this archive that its own
timeout config doesn't catch, and it can also drop a connection
mid-read. Implement a small synchronous file-like object over plain
`requests` (HEAD for Content-Length, GET with a Range header per read,
with a short retry-on-short-read loop) and hand that to pyarrow instead.
This is much more reliable and it's fewer dependencies.
- That custom file-like object is stateful (it tracks a current byte
position) and NOT thread-safe. Pass `use_threads=False` to
`pq.read_table(...)` - pyarrow's default multi-threaded scanner issues
concurrent seek()/read() calls against it, which corrupts its position
and hangs the whole read indefinitely (no exception, no timeout, just
silence) rather than failing cleanly. This is easy to miss because it
can work fine on some machines/pyarrow versions and hang forever on
others - always set it explicitly.
- When sorting/deduplicating rows by timestamp, sort by the timestamp
alone (e.g. `key=lambda tv: tv[0]`), never by the raw (timestamp, value)
tuple - ties on timestamp are common, and if value is a list of dicts
(like an order-book snapshot) Python can't compare it and will crash.
Output: replay the window in the terminal (or render a chart/video) with the
BTC/ETH/etc price line and the order book (best bid/ask, or a few levels of
real depth) moving together from window open to resolution. Optionally
compress the real window into a shorter playback (e.g. 2x speed) so a 5-minute
market plays back in ~2.5 minutes.
NOTE: if you only need the combined data (not a terminal/video animation),
skip building all of the above and just call Vatic's own hosted endpoint
instead: GET https://api.vatic.trading/api/v1/replay/. It does this
exact Vatic+PendulumFlow fetch server-side. It's a polling endpoint (returns
status:"pending" with a retryAfterSeconds hint until the fetch finishes, then
status:"ready" with the full payload) - no API key needed.
```
## What you get
* **Price truth**: Vatic's target price + Chainlink TWAP close, exactly what the market resolved on.
* **Market microstructure**: PendulumFlow's real order book — best bid/ask over time, full depth snapshots, actual trade prints.
* **No API keys, no cost.** Both sources are free and public.
PendulumFlow is run by an independent maintainer on donations, with no account, key, or rate limit. If you build something with it, credit it and consider [supporting it](https://x.com/PendulumFlow).
## A worked example
`btc-updown-5m-1790244900` (Sep 24, 2026, resolved Up):
* Target (open): `$83,204.32`
* TWAP close (resolution): `$83,289.67`
* The order book's implied "Up" probability lagged the real Chainlink price for the first \~10 seconds of the window — the price was already `$52` above target while the "Up" share was still asking `$0.66` — then converged to `$1.00` by resolution, with the ask side fully depleted well before the window closed.
That lag between price-truth and order-book-implied-probability is exactly the kind of thing this combination makes visible.
## Reference implementation
If you'd rather start from working code than a spec, expand the accordion below for the actual script used to produce the recording above — battle-tested against the gotchas listed in the agent prompt (the fsspec hang, the byte-encoding traps, the TWAP window edge case). Three dependencies: `requests`, `pandas`, `pyarrow`. Copy it out with the button in the top-right of the code block, save it as `replay.py`, and run:
```bash theme={null}
pip install requests pandas pyarrow
python3 replay.py btc-updown-5m-1790244900 # live 2x-speed animation
python3 replay.py btc-updown-5m-1790244900 --instant # one-shot summary, no animation
```
The `--video` flag in this script is a placeholder for a separate matplotlib/ffmpeg renderer that isn't included here — the terminal output above is the fully working path.
```python theme={null}
#!/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[a-z0-9]+)-updown-(?P\d+[mh]|daily)-(?P\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/.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)
```
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.