如何回测 Polymarket
基于当时真实存在的订单簿。

import time, requests
 
API = "https://tickfoundry.com/api/v1"           # Explorer and up
S = requests.Session()
S.headers["Authorization"] = "Bearer tf_live_…"   # minted in the dashboard
 
unit = S.get(f"{API}/catalog", params={"q": "bitcoin up or down", "kind": "series"}).json()["units"][0]["id"]
job  = S.post(f"{API}/downloads", json={"unit_id": unit, "date": "2026-08-19"}).json()
while job["status"] == "staging":
    time.sleep(job.get("retry_after", 5))
    job = S.get(f"{API}/downloads/{job['id']}").json()
open("bundle.zip", "wb").write(S.get(job["url"]).content)

import pandas as pd
 
# the free sample: https://tickfoundry.com/samples (no account needed)
l1     = pd.read_parquet("l1.parquet")       # every top-of-book change, ns receive time
book   = pd.read_parquet("l2.parquet")       # full 25-level snapshot per book update
trades = pd.read_parquet("trades.parquet")   # every fill the venue reported
 
# token_id is a 77-digit integer — read it as a string or the join matches nothing
tokens  = pd.read_csv("reference/tokens.csv", dtype={"token_id": str})
markets = pd.read_csv("reference/markets.csv")
 
l1 = l1.merge(tokens[["token_id", "market_id", "outcome_label"]], on="token_id")
l1["mid"] = (l1.best_bid + l1.best_ask) / 2
l1["ts"]  = pd.to_datetime(l1.ts_recv_ns, unit="ns", utc=True)

# slippage.py — walk the historical L2 book for a market buy and compare
# the price you would actually have paid with the price a naive backtest assumes.
import numpy as np
import pandas as pd
 
l2 = pd.read_parquet("l2.parquet")                 # one full snapshot per book update
markets = pd.read_csv("reference/markets.csv")
 
# Pick one outcome token: the YES leg of the busiest market of the day.
busiest = l2.groupby("condition_id").size().idxmax()
yes = str(markets.set_index("condition_id").loc[busiest, "yes_token_id"])
book = l2[l2.token_id == yes].sort_values("ts_recv_ns").reset_index(drop=True)
 
ask_px = book[[f"ask_px_{i}" for i in range(1, 26)]].to_numpy()   # per level, NaN past depth
ask_sz = book[[f"ask_sz_{i}" for i in range(1, 26)]].to_numpy()   # size is in shares
 
def buy_vwap(notional_usd):
    """Average price paid for a market buy of notional_usd, walking the ask
    ladder snapshot by snapshot. NaN where the visible book is too thin."""
    cost = ask_px * ask_sz                                    # $ resting at each level
    cum = np.nancumsum(cost, axis=1)
    filled = np.minimum(cost, np.maximum(notional_usd - (cum - cost), 0))
    shares = np.nansum(filled / ask_px, axis=1)
    enough = np.nanmax(cum, axis=1) >= notional_usd
    with np.errstate(divide="ignore"):
        return np.where(enough, notional_usd / shares, np.nan)
 
mid = (book.best_bid + book.best_ask) / 2
for usd in (100, 500, 2000):
    vwap = buy_vwap(usd)
    print(usd, f"fillable {np.mean(~np.isnan(vwap)):.1%}",
          f"median slip vs mid {np.nanmedian((vwap - mid) * 100):.2f}¢",
          f"vs best ask {np.nanmedian((vwap - book.best_ask) * 100):.2f}¢")

样本只是比特币市场的一天。免费账户可以从平台上的全部市场中自选 5 个市场·日，数据结构完全相同，这个脚本无需修改就能在其中任何一个上运行。