diff --git a/benchmarks/pm-data-freshness.yml b/benchmarks/pm-data-freshness.yml deleted file mode 100644 index c6fb7bd1..00000000 --- a/benchmarks/pm-data-freshness.yml +++ /dev/null @@ -1,138 +0,0 @@ -# OpenChainBench. Bench № 032 - -slug: pm-data-freshness -number: "032" -title: Fastest prediction market data API, live freshness across venues -seo_title: "Fastest prediction market data API 2026" -seo_description: "Fastest prediction market data API ranked live across Polymarket and Kalshi. Milliseconds Codex and Predexon lag the venue gateway publish on top markets." -subtitle: Per event delay between provider arrival and the venue gateway publish, measured every minute on the top markets by 24 hour volume across Polymarket and Kalshi. - -category: Aggregators -status: live -metric: Freshness delta vs venue -unit: ms -higher_is_better: false - -seo_intro: | - Prediction markets generate the most time sensitive event stream in crypto. - An election market settles in seconds, a sports book moves on every play. - Builders that integrate Polymarket or Kalshi through a data provider - rather than hitting each venue directly need to know how many - milliseconds that provider adds between when the venue itself publishes - a trade and when the provider relays the same trade to its WebSocket - subscribers. This benchmark measures exactly that. The harness - subscribes to the same basket of top volume markets on each venue's own - canonical source (T0) and on Codex GraphQL subscriptions. Each trade - is cross correlated across the streams, and - the per provider lag versus the venue gateway publish time is recorded - as a Prometheus histogram. The leaderboard sorts by p50 freshness delta - in milliseconds, lower is better. Use the venue tab at the top of the - page to switch between Polymarket and Kalshi. - -abstract: | - Three WebSocket subscribers ride the same rotating basket of ~20 top - volume markets simultaneously on each venue. For every trade event - published on the venue's own canonical WebSocket gateway (the T0), we - record the moment it lands and the moment each provider relays the - same trade. The signature used to match a trade across providers is a - tuple based on the market identifier, price and size of the fill, and - a small time bucket, which is robust against the minor clock skew - between gateways. Providers that fail to relay a trade within 90 - seconds are not counted toward their p50, only toward their receive - total, so a provider can look fresh on the leaderboard while silently - dropping events. The success rate column flags that. Coverage today: - Codex on Polymarket and Kalshi. - -methodology: - - "Anchor rows: on each venue tab the venue itself (Polymarket, Kalshi) is the canonical T0, so its row reads as the harness network round trip, near zero by construction. It is kept on the board to document the reference point, not as a competing feed; the comparison rows are the data providers measured against it." - - "Polymarket T0. `wss://ws-subscriptions-clob.polymarket.com/ws/market` is public, no auth, sub 50ms gateway publish latency from EU West. Cross correlation uses (conditionId, priceUSD rounded to 3 decimals, sizeUSD micros, 5s time bucket)." - - "Kalshi T0. `https://api.elections.kalshi.com/v1/social/trades` (REST), polled every 5s with cursor pagination. Every trade carries `create_date` at microsecond precision; that timestamp is the canonical T0. Poll cadence affects only correlation timing, not the freshness number. The official Kalshi WebSocket needs RSA PSS signed headers from a US KYC account and 403s from non US IPs, not viable for a public benchmark." - - "Codex GraphQL subscriptions. `wss://graph.codex.io/graphql` with the `graphql-transport-ws` subprotocol. Firehose `onPredictionTradesCreated` subscription filtered client side to the venue's marketIds. Covers both Polymarket and Kalshi." - - "Basket. Top 20 active markets by 24h volume per venue, refreshed every 5 minutes. Polymarket pulls from `gamma-api.polymarket.com`, Kalshi pulls from `api.elections.kalshi.com/trade-api/v2/markets`." - - "Histogram buckets: 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000, 60000 ms. The venue's own arrivals always land in the smallest bucket because by construction we measure other providers against it." - - "Region. `eu-west` (Railway europe-west4). Both venue gateways are geographically distributed; deltas reflect EU client to EU gateway latency." - - "Venue clock asymmetry. Polymarket CLOB trades carry no venue timestamp so T0 is when our harness receives the WS event; the Polymarket-to-harness RTT cancels inside the Codex delta. Kalshi's REST `create_date` is venue-side so the Codex Kalshi delta includes a Codex-to-harness RTT (50 to 100 ms) the Polymarket math cancels out. Kalshi rows are slightly stricter for the same provider, not more favorable; the gap is small versus the multi-second deltas the chart highlights." - -findings: - - "Each venue's own gateway is by definition the freshest source on the leaderboard for that venue. The lag versus its own publish time is on the order of the network round trip from the harness to the gateway, typically below 100 ms p50 from EU West." - - "{{name:codex}} sits at p50 {{p50:codex}} on the Polymarket tab because Codex indexes the chain event (Polygon block confirmation) rather than the off chain orderbook publish. The lag includes Polygon block time (~2s) plus ingestion. On Kalshi Codex ingests Kalshi's own WebSocket feed directly, so the delta is purely pipeline latency." - - "The spread between providers reflects integration depth: native gateway vs chain indexed pipeline. None of these is wrong, they answer different questions. For live trading UIs the gateway path is the only viable one." - -faq: - - q: "Which prediction market data API has the lowest latency right now?" - a: "{{best_name}} currently leads at {{best_p50}} (p50 over the last 24h) measured as time from the venue's own canonical WebSocket publish to provider relay. The leaderboard re sorts every minute on fresh Prometheus samples, so the answer reflects the actual measured lag on the active market basket, not a marketing claim. Switch the venue tab at the top to see Polymarket vs Kalshi independently." - - q: "What does 'freshness delta' mean for a prediction market API?" - a: "We connect to the venue's own canonical WebSocket and to the provider's WebSocket simultaneously, subscribe to the same markets, and for every trade event we record how many milliseconds the provider takes to relay the event after the venue itself publishes it. Lower is better. The venue's own gateway publish time is the canonical T0 because by construction nothing downstream can be faster than the source." - - q: "Why don't you include Polymarket REST polling on this benchmark?" - a: "When the Polymarket tab is selected, freshness is a WebSocket question. REST polling at 1s would have a floor freshness around 500ms (poll interval / 2) plus RTT, dominated by how often you poll. The Polymarket gateway WebSocket exists for exactly this reason, to avoid that floor. Adding REST as a row would make the leaderboard noisy without changing the conclusion: WebSocket beats polling by definition for real time data." - - q: "What does the Kalshi T0 source measure?" - a: "Kalshi's own public WebSocket at `wss://external-api-ws.kalshi.com/trade-api/ws/v2`, channel `trade`. The harness reads the `msg.ts_ms` field of each fill as the canonical publish timestamp and uses it as T0 for every Kalshi side comparison. Just like Polymarket's CLOB WebSocket on the Polymarket tab, it is the source against which provider latency is measured." - - q: "Are these numbers comparable to Kalshi or Limitless?" - a: "Kalshi is now its own tab on this page, click the Venue selector at the top to switch. Each venue uses its own canonical WebSocket as T0, so the absolute numbers are not directly comparable across tabs (different gateways, different network paths), but the relative provider ordering within each tab is honest. Limitless is not yet covered, we will add a tab if and when meaningful provider coverage exists for it." - - q: "How does OpenChainBench measure freshness?" - a: "Three WebSocket clients run in parallel inside the harness, all subscribed to the same basket of top volume markets on the active venue. Every 5 minutes we refresh the basket from the venue's own markets API. For each trade event, we compute a signature based on market id, price, size and a 5 second time bucket, and record the wall clock receive time on each provider. The freshness delta is `recv_provider - recv_venue` for the same signature. We export the histogram to Prometheus, the leaderboard reads the 24h p50." - -source: https://github.com/ChainBench/OpenChainBench/tree/main/harnesses/pm-freshness-bench - -prometheus: - window: 24h - expected_freshness_seconds: 300 - -# Venue selector. tabs at the top of the page. Server injects -# `venue="X"` into every PromQL query for the active tab. There is no -# "all" entry because cross venue averaging would mix two different -# canonical T0 gateways, which is not a meaningful comparison; the -# default landing tab is Polymarket (the first entry below). -dimensions: - venue: - - { value: polymarket, label: Polymarket } - - { value: kalshi, label: Kalshi } - -providers: - - slug: polymarket - name: Polymarket - tag: Native gateway WebSocket, no auth, sub 50ms publish latency - formula: "Polymarket gateway publish time is the canonical T0 on the Polymarket tab. The near zero row is the network round trip from the harness to the gateway, not a comparison against another source." - queries: - # Polymarket is T0 by construction on the Polymarket tab. Multiply - # the 0.5 ms floor by pm_health{provider="polymarket"} so the - # dimension framework's injected venue label gates the row: - # pm_health{provider="polymarket", venue="polymarket"} = 1 -> 0.5, - # pm_health{provider="polymarket", venue="kalshi"} = 0 -> 0, - # and the leaderboard hides empty rows. - p50: pm_health{provider="polymarket"} * 0.5 - p90: pm_health{provider="polymarket"} * 0.5 - p99: pm_health{provider="polymarket"} * 0.5 - mean: pm_health{provider="polymarket"} * 0.5 - success: avg_over_time(pm_health{provider="polymarket"}[24h]) - sample_size: sum(increase(pm_events_total{provider="polymarket"}[24h])) - series: pm_health{provider="polymarket"} * 0.5 - - - slug: kalshi - name: Kalshi - tag: Native venue REST, create_date is canonical T0 - formula: "Kalshi gateway publish time is the canonical T0 on the Kalshi tab. The near zero row is the network round trip from the harness to the Kalshi endpoint, not a comparison against another source." - queries: - # Symmetric to the Polymarket row, gated by pm_health{provider="kalshi"} - # so the venue label injected at runtime zeros out the row on the - # Polymarket tab. - p50: pm_health{provider="kalshi"} * 0.5 - p90: pm_health{provider="kalshi"} * 0.5 - p99: pm_health{provider="kalshi"} * 0.5 - mean: pm_health{provider="kalshi"} * 0.5 - success: avg_over_time(pm_health{provider="kalshi"}[24h]) - sample_size: sum(increase(pm_events_total{provider="kalshi"}[24h])) - series: pm_health{provider="kalshi"} * 0.5 - - - slug: codex - name: Codex - tag: Chain indexer on Polymarket, native ingestion on Kalshi - formula: "Median ms lag versus the venue gateway. On Polymarket Codex indexes the on chain Polygon settlement of each trade (block time ~2s plus ingestion). On Kalshi Codex ingests Kalshi's own WebSocket directly so the lag is pure pipeline latency." - queries: - p50: histogram_quantile(0.50, sum(rate(pm_freshness_delta_ms_bucket{provider="codex",kind="trade"}[24h])) by (le)) - p90: histogram_quantile(0.90, sum(rate(pm_freshness_delta_ms_bucket{provider="codex",kind="trade"}[24h])) by (le)) - p99: histogram_quantile(0.99, sum(rate(pm_freshness_delta_ms_bucket{provider="codex",kind="trade"}[24h])) by (le)) - mean: sum(rate(pm_freshness_delta_ms_sum{provider="codex",kind="trade"}[24h])) / sum(rate(pm_freshness_delta_ms_count{provider="codex",kind="trade"}[24h])) - success: clamp_max(sum(rate(pm_matched_total{provider="codex",kind="trade"}[24h])) / sum(rate(pm_events_total{provider="polymarket",kind="trade"}[24h])), 1) - sample_size: sum(increase(pm_matched_total{provider="codex",kind="trade"}[24h])) - series: histogram_quantile(0.50, sum(rate(pm_freshness_delta_ms_bucket{provider="codex",kind="trade"}[1h])) by (le)) diff --git a/benchmarks/pm-ws-latency.yml b/benchmarks/pm-ws-latency.yml index 992b388f..cd7accfb 100644 --- a/benchmarks/pm-ws-latency.yml +++ b/benchmarks/pm-ws-latency.yml @@ -5,7 +5,7 @@ number: "114" title: Polymarket WebSocket latency, live connection and update speed from 3 regions seo_title: "Polymarket WebSocket latency 2026: connection time and update speed" seo_description: "How fast does the Polymarket WebSocket connect and deliver market updates? Live p50 from us-east, eu-west and Singapore measured continuously by OpenChainBench." -subtitle: Time from WebSocket dial to first market snapshot plus inter-message update rate on the Polymarket CLOB feed, probed from three regions 24 hours a day. +subtitle: Connect-to-snapshot latency, inter-message update rate, and trade publication lag on the Polymarket and Kalshi WebSocket feeds, probed from three regions 24 hours a day. category: Trading status: live @@ -99,6 +99,20 @@ metric_panels: label_key: venue unit: ms higher_is_better: false + - id: trade_pub_lag_p50 + label: Trade pub lag p50 + description: "Median delay between venue server-side trade creation and WS subscriber receipt. Polymarket: CLOB WS timestamp field vs receive time (all regions). Kalshi: REST create_date vs WS ticker receive time (us-east only). Lower is better." + metric: 1000 * histogram_quantile(0.50, sum by (venue) (rate(pmapi_ws_trade_pub_lag_seconds_bucket{source="direct",venue=~"polymarket|kalshi"}[24h]))) + label_key: venue + unit: ms + higher_is_better: false + - id: trade_pub_lag_p90 + label: Trade pub lag p90 + description: "90th percentile trade publication lag. The tail cost a subscriber pays waiting for the venue to push a confirmed trade. Kalshi us-east only." + metric: 1000 * histogram_quantile(0.90, sum by (venue) (rate(pmapi_ws_trade_pub_lag_seconds_bucket{source="direct",venue=~"polymarket|kalshi"}[24h]))) + label_key: venue + unit: ms + higher_is_better: false - id: disconnect_rate label: Disconnects 24h description: "Number of WebSocket disconnects recorded across all regions in the last 24 hours. Kalshi probes us-east only; Polymarket probes all three regions." diff --git a/harnesses/pm-freshness-bench/.env.example b/harnesses/pm-freshness-bench/.env.example deleted file mode 100644 index 53c28cc7..00000000 --- a/harnesses/pm-freshness-bench/.env.example +++ /dev/null @@ -1,18 +0,0 @@ -# Codex (defined.fi) JWT scraping. Cookie rotated every ~7 days, see -# aggregator-latency-benchmark/CODEX_SESSION_GUIDE.md for refresh. -DEFINED_SESSION_COOKIE= - -# Webshare rotating proxy — required for Codex JWT mint (Vercel bans direct IPs). -HTTP_PROXY= -HTTPS_PROXY= - -# Mobula PM API key. Needs Growth or Enterprise plan for PM streams. -MOBULA_API_KEY= - -# Optional: token-gate /logs and /debug endpoints. -LOGS_TOKEN= - -# Optional: bench cadence. Defaults are sane. -# REFRESH_MARKETS_INTERVAL_SEC=300 # how often to re-poll gamma-api for the basket -# BASKET_SIZE=20 # how many top-vol markets to subscribe per provider -# KALSHI_POLL_INTERVAL_SEC=5 # /v1/social/trades cadence (CloudFront max-age=10) diff --git a/harnesses/pm-freshness-bench/Dockerfile b/harnesses/pm-freshness-bench/Dockerfile deleted file mode 100644 index d8ce6a93..00000000 --- a/harnesses/pm-freshness-bench/Dockerfile +++ /dev/null @@ -1,13 +0,0 @@ -FROM golang:1.24-alpine AS build -WORKDIR /src -COPY go.mod go.sum ./ -RUN go mod download -COPY cmd ./cmd -RUN CGO_ENABLED=0 GOOS=linux go build -o /out/script ./cmd/script - -FROM alpine:3.20 -RUN apk add --no-cache ca-certificates && update-ca-certificates -WORKDIR /app -COPY --from=build /out/script /app/script -EXPOSE 2112 -ENTRYPOINT ["/app/script"] diff --git a/harnesses/pm-freshness-bench/README.md b/harnesses/pm-freshness-bench/README.md deleted file mode 100644 index 1477a774..00000000 --- a/harnesses/pm-freshness-bench/README.md +++ /dev/null @@ -1,84 +0,0 @@ -# pm-freshness-bench - -Bench №032 — prediction market data freshness across providers. - -Subscribes to the same basket of high-volume Polymarket markets on three -data providers simultaneously, cross-correlates each trade event by -`(conditionId, outcomeId, price, size, time-window)` and measures how -many milliseconds each provider lags the canonical Polymarket CLOB WS -gateway. - -## Providers (v2) - -The bench now compares per-venue. Polymarket events flow through 3 providers; -Kalshi events flow through 2 (no Mobula Kalshi coverage). - -| Venue | Provider | Endpoint | Auth | Notes | -|---|---|---|---|---| -| Polymarket | Polymarket CLOB | `wss://ws-subscriptions-clob.polymarket.com/ws/market` | none | Canonical T0 for Polymarket | -| Polymarket | Codex / defined.fi | `wss://graph.codex.io/graphql` | scraped JWT + Webshare proxy | Firehose, branch on marketId suffix | -| Polymarket | Mobula PM | `wss://pm-api-prod-eu.mobula.io` | API key | Browser UA required | -| Kalshi | Kalshi public REST | `https://api.elections.kalshi.com/v1/social/trades` | none | Canonical T0 via `create_date` (venue clock, µs precision) | -| Kalshi | Codex / defined.fi | (same firehose) | (same auth) | Filtered to `:Kalshi` marketIds | - -## Run locally - -```bash -cp .env.example .env -# fill in keys -go build ./cmd/script -./script -``` - -## Endpoints - -- `:2112/metrics` — Prometheus scrape (hardcoded — Railway $PORT ignored) -- `:2112/logs?tail=N` — last N log lines (token-gated by `LOGS_TOKEN` when set) - -## Metrics - -All metrics now carry a `venue` label (`polymarket` | `kalshi`). The T0 -reference clock per venue is the direct venue feed (`polymarket` for -Polymarket events, `kalshi` for Kalshi events). - -- `pm_freshness_delta_ms_bucket{provider, venue, kind}` histogram — per-event - delta vs that venue's T0, kind ∈ {trade, price} -- `pm_events_total{provider, venue, kind}` counter — raw events received -- `pm_matched_total{provider, venue, kind}` counter — events successfully matched to T0 -- `pm_health{provider, venue}` gauge — 1 if (provider, venue) published at least - one event in the last 60 s, else 0 -- `pm_fetch_errors_total{provider, venue, error_type}` counter -- `pm_basket_size{venue}` gauge - -## Kalshi support - -Kalshi runs as a second venue alongside Polymarket. Codex's -prediction-trades firehose already carries Kalshi trades; we just stopped -filtering them out. The Kalshi T0 (canonical publish time) comes from the -public `/v1/social/trades` REST endpoint — the same one that powers the -trade ticker on kalshi.com homepage. No account, no KYC, no key. - -The freshness measurement uses the `create_date` field embedded in the -Kalshi response, which is the venue's own publish timestamp at microsecond -precision. The poll cadence (default 5s, tunable via -`KALSHI_POLL_INTERVAL_SEC`) only governs when correlation happens, not -the measurement itself: a Codex relay event that lands at T+100ms is -still credited with a 100ms delta even if our poller surfaces the matching -Kalshi trade up to 5 seconds later. - -CloudFront fronts the endpoint with `max-age=10s`; polling faster than -~5s just hits cache. Tested live from Paris with 200ms RTT, no geo -block, no auth header required. - -## Methodology - -Every 5 minutes the harness polls `gamma-api.polymarket.com` for the top-20 -active markets by 24h volume, then opens / updates subscriptions on all -three providers simultaneously. Trades are indexed in memory for 60 s -after first arrival; deltas are computed against the earliest receive -time for that `(conditionId, outcomeId, price, size)` signature, which -is always the Polymarket T0 if Polymarket emitted the trade at all. - -Polymarket trades that no other provider relays in the 60 s window are -counted in `pm_events_total{provider="polymarket"}` but not in any -matched counter. diff --git a/harnesses/pm-freshness-bench/cmd/script/codex.go b/harnesses/pm-freshness-bench/cmd/script/codex.go deleted file mode 100644 index b4db8351..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/codex.go +++ /dev/null @@ -1,323 +0,0 @@ -package main - -import ( - "context" - "encoding/json" - "fmt" - "net/http" - "strings" - "sync" - "time" - - "nhooyr.io/websocket" -) - -const codexWS = "wss://graph.codex.io/graphql" - -func runCodex(ctx context.Context, cfg Config, basketCh <-chan []Market, kalshiBasketCh <-chan []Market) { - if cfg.DefinedSessionCookie == "" { - appendLog("[codex] DEFINED_SESSION_COOKIE not set — skipping Codex provider") - return - } - - // Codex uses its own composite marketId format. For Polymarket events: - // :Polymarket:: - // For Kalshi events the trailing segment is `:Kalshi` and the leading - // segment is the market_ticker. We branch on the suffix to pick venue - // and which basket to check against. - - var ( - cidMu sync.RWMutex - knownPoly = map[string]bool{} - knownKalshi = map[string]bool{} - ) - updateKnownPoly := func(ms []Market) { - cidMu.Lock() - defer cidMu.Unlock() - knownPoly = map[string]bool{} - for _, m := range ms { - knownPoly[m.ConditionId] = true - } - } - updateKnownKalshi := func(ms []Market) { - cidMu.Lock() - defer cidMu.Unlock() - knownKalshi = map[string]bool{} - for _, m := range ms { - knownKalshi[m.ConditionId] = true - } - } - isKnown := func(venue, cid string) bool { - cidMu.RLock() - defer cidMu.RUnlock() - if venue == "kalshi" { - return knownKalshi[cid] - } - return knownPoly[cid] - } - - updateKnownPoly(currentBasket()) - updateKnownKalshi(currentKalshiBasket()) - - // basket update consumers — Polymarket + Kalshi basket channels are - // independent because each venue has its own refresh loop. - go func() { - for { - select { - case <-ctx.Done(): - return - case ms := <-basketCh: - updateKnownPoly(ms) - case ms := <-kalshiBasketCh: - updateKnownKalshi(ms) - } - } - }() - - backoff := 10 * time.Second - consecutiveFails := 0 - for ctx.Err() == nil { - err := codexConnect(ctx, cfg, isKnown) - if ctx.Err() != nil { - return - } - consecutiveFails++ - errType := classify(err) - // Errors are reported on the polymarket venue to keep historical - // series; Codex transport errors don't have a per-venue semantic. - fetchErrors.WithLabelValues("codex", "polymarket", errType).Inc() - appendLog("[codex] disconnected: %v — backing off %v", err, backoff) - - // Mirror the head-lag bench's recovery rules. - if errType == "auth" || errType == "rate_limit" { - InvalidateCodexJWT() - backoff = 30 * time.Second - } - if consecutiveFails >= 10 { - InvalidateCodexJWT() - backoff = 10 * time.Second - consecutiveFails = 0 - } - - select { - case <-ctx.Done(): - return - case <-time.After(backoff): - } - if backoff < 60*time.Second { - backoff *= 2 - } - } -} - -func codexConnect(ctx context.Context, cfg Config, isKnown func(venue, cid string) bool) error { - jwt, err := GetCodexJWT(cfg.DefinedSessionCookie) - if err != nil { - return fmt.Errorf("mint: %w", err) - } - - opts := &websocket.DialOptions{ - Subprotocols: []string{"graphql-transport-ws"}, - HTTPHeader: http.Header{ - "Origin": []string{"https://www.defined.fi"}, - "User-Agent": []string{"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Chrome/131.0.0.0"}, - }, - } - conn, _, err := websocket.Dial(ctx, codexWS, opts) - if err != nil { - return fmt.Errorf("dial: %w", err) - } - defer conn.Close(websocket.StatusInternalError, "") - conn.SetReadLimit(8 * 1024 * 1024) - - init := map[string]interface{}{ - "type": "connection_init", - "payload": map[string]interface{}{"Authorization": "Bearer " + jwt}, - } - ib, _ := json.Marshal(init) - if err := conn.Write(ctx, websocket.MessageText, ib); err != nil { - return fmt.Errorf("init: %w", err) - } - _, ackB, err := conn.Read(ctx) - if err != nil { - return fmt.Errorf("ack read: %w", err) - } - var ack struct { - Type string `json:"type"` - } - json.Unmarshal(ackB, &ack) - if ack.Type != "connection_ack" { - return fmt.Errorf("expected connection_ack got %s", ack.Type) - } - appendLog("[codex] connection_ack") - - // Firehose subscription. We accept both Polymarket AND Kalshi trades - // client-side because Codex's input doesn't take a protocol filter on - // the subscription itself (only on the discovery query). The volume is - // manageable — observed ~33 msg/sec mixing both venues. - subQuery := `subscription Fire { onPredictionTradesCreated { marketId trades { timestamp transactionHash priceUsd amountUsd tradeType outcomeId outcomeLabel } } }` - subMsg := map[string]interface{}{ - "type": "subscribe", - "id": "firehose", - "payload": map[string]interface{}{"query": subQuery, "variables": map[string]interface{}{}}, - } - sb, _ := json.Marshal(subMsg) - if err := conn.Write(ctx, websocket.MessageText, sb); err != nil { - return fmt.Errorf("subscribe write: %w", err) - } - - pingCtx, pingCancel := context.WithCancel(ctx) - defer pingCancel() - go func() { - t := time.NewTicker(20 * time.Second) - defer t.Stop() - for { - select { - case <-pingCtx.Done(): - return - case <-t.C: - conn.Write(pingCtx, websocket.MessageText, []byte(`{"type":"ping"}`)) - } - } - }() - - for { - _, b, err := conn.Read(ctx) - if err != nil { - return err - } - var m struct { - Type string `json:"type"` - Id string `json:"id"` - Payload json.RawMessage `json:"payload"` - } - json.Unmarshal(b, &m) - if m.Type != "next" { - continue - } - var p struct { - Data struct { - OnPredictionTradesCreated struct { - MarketId string `json:"marketId"` - Trades []struct { - Timestamp int64 `json:"timestamp"` - TransactionHash string `json:"transactionHash"` - PriceUsd string `json:"priceUsd"` - AmountUsd string `json:"amountUsd"` - TradeType string `json:"tradeType"` - } `json:"trades"` - } `json:"onPredictionTradesCreated"` - } `json:"data"` - } - if err := json.Unmarshal(m.Payload, &p); err != nil { - continue - } - marketId := p.Data.OnPredictionTradesCreated.MarketId - venue, cid := codexParseMarketId(marketId) - logCodexMarketIDSample(marketId, venue, cid, isKnown(venue, cid)) - if cid == "" || venue == "" || !isKnown(venue, cid) { - continue // out-of-basket, unknown venue, or unparsable id - } - health.WithLabelValues("codex", venue).Set(1) - markAlive("codex", venue) - now := nowSec() - for _, t := range p.Data.OnPredictionTradesCreated.Trades { - eventsTotal.WithLabelValues("codex", venue, "trade").Inc() - sig := EventSig{ - Venue: venue, - ConditionId: cid, - PriceMilli: priceToMilli(t.PriceUsd), - BucketSec: bucketSec(float64(t.Timestamp)), - } - logSigSample("codex", sig) - correlator.Add(sig, Arrival{Provider: "codex", Kind: "trade", RecvUnix: now}) - } - } -} - -// codexShapesMu + codexShapesLogged track which composite marketId -// shapes we've already logged a sample for. The goal is to surface every -// distinct shape Codex's firehose can emit in the first few minutes after -// boot — particularly important for Kalshi where the suffix format wasn't -// verified live. We log each (venue, known) tuple once per unique cid -// prefix shape, capped at 50 samples total. -var ( - codexShapesMu sync.Mutex - codexShapesSeen = map[string]struct{}{} - codexShapesLogged int - codexShapesCap = 50 -) - -func logCodexMarketIDSample(marketId, venue, cid string, known bool) { - if marketId == "" { - return - } - codexShapesMu.Lock() - defer codexShapesMu.Unlock() - if codexShapesLogged >= codexShapesCap { - return - } - // Shape key = venue + known + shape(marketId). Shape collapses hex - // runs to "" and decimal runs to "" so we don't log every - // individual market. - shape := shapeMarketID(marketId) - key := fmt.Sprintf("%s|%t|%s", venue, known, shape) - if _, ok := codexShapesSeen[key]; ok { - return - } - codexShapesSeen[key] = struct{}{} - codexShapesLogged++ - appendLog("[codex-debug] marketId=%s venue=%q cid=%q known=%t (sample %d/%d, shape=%s)", - marketId, venue, cid, known, codexShapesLogged, codexShapesCap, shape) -} - -// shapeMarketID replaces hex and decimal runs with placeholders so two -// otherwise-equivalent marketIds collapse to one shape string. -func shapeMarketID(s string) string { - var b strings.Builder - i := 0 - for i < len(s) { - c := s[i] - switch { - case c == '0' && i+1 < len(s) && (s[i+1] == 'x' || s[i+1] == 'X'): - b.WriteString("") - i += 2 - for i < len(s) && isHexDigit(s[i]) { - i++ - } - case c >= '0' && c <= '9': - b.WriteString("") - for i < len(s) && s[i] >= '0' && s[i] <= '9' { - i++ - } - default: - b.WriteByte(c) - i++ - } - } - return b.String() -} - -func isHexDigit(c byte) bool { - return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F') -} - -// codexParseMarketId splits Codex's composite marketId into (venue, id). -// Polymarket events look like 0x:Polymarket:0x: -// Kalshi events look like :Kalshi -// Returns ("", "") if the format isn't recognised. -func codexParseMarketId(s string) (venue string, id string) { - idx := strings.Index(s, ":") - if idx <= 0 { - return "", "" - } - id = strings.ToLower(s[:idx]) - rest := s[idx+1:] - switch { - case strings.HasPrefix(rest, "Polymarket"): - return "polymarket", id - case strings.HasPrefix(rest, "Kalshi"): - return "kalshi", id - } - return "", "" -} diff --git a/harnesses/pm-freshness-bench/cmd/script/codex_auth.go b/harnesses/pm-freshness-bench/cmd/script/codex_auth.go deleted file mode 100644 index 0720a305..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/codex_auth.go +++ /dev/null @@ -1,151 +0,0 @@ -package main - -// JWT mint logic vendored from -// miniapps/aggregator-latency-benchmark/cmd/script/defined_auth.go -// keeping the same Webshare-proxy + 7-day-cookie flow. Reusing the working -// production pattern instead of re-inventing — the head-lag bench has been -// minting JWTs against defined.fi this way for months. - -import ( - "bytes" - "encoding/base64" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" - "sync" - "time" -) - -type definedTokenResponse struct { - Data struct { - CreateApiTokens []struct { - Token string `json:"token"` - } `json:"createApiTokens"` - } `json:"data"` -} - -type definedTokenCache struct { - mu sync.RWMutex - token string - expiresAt time.Time - lastRefresh time.Time -} - -var codexTokenCache = &definedTokenCache{} - -func decodeJWTExpiration(token string) (time.Time, error) { - parts := strings.Split(token, ".") - if len(parts) != 3 { - return time.Time{}, fmt.Errorf("invalid JWT format") - } - payload, err := base64.RawURLEncoding.DecodeString(parts[1]) - if err != nil { - return time.Time{}, err - } - var claims struct { - Exp int64 `json:"exp"` - } - if err := json.Unmarshal(payload, &claims); err != nil { - return time.Time{}, err - } - if claims.Exp == 0 { - return time.Time{}, fmt.Errorf("no exp") - } - return time.Unix(claims.Exp, 0), nil -} - -// GetCodexJWT returns a cached short-lived JWT, minting a new one when the -// cached one is within 1h of expiry. -func GetCodexJWT(sessionCookie string) (string, error) { - codexTokenCache.mu.RLock() - if codexTokenCache.token != "" && time.Now().Before(codexTokenCache.expiresAt.Add(-1*time.Hour)) { - t := codexTokenCache.token - codexTokenCache.mu.RUnlock() - return t, nil - } - codexTokenCache.mu.RUnlock() - - codexTokenCache.mu.Lock() - defer codexTokenCache.mu.Unlock() - if codexTokenCache.token != "" && time.Now().Before(codexTokenCache.expiresAt.Add(-1*time.Hour)) { - return codexTokenCache.token, nil - } - tok, err := mintCodexJWT(sessionCookie) - if err != nil { - return "", err - } - exp, err := decodeJWTExpiration(tok) - if err != nil { - exp = time.Now().Add(24 * time.Hour) - } - codexTokenCache.token = tok - codexTokenCache.expiresAt = exp - codexTokenCache.lastRefresh = time.Now() - appendLog("[codex-auth] JWT refreshed, expires in %.1fh", time.Until(exp).Hours()) - return tok, nil -} - -func InvalidateCodexJWT() { - codexTokenCache.mu.Lock() - defer codexTokenCache.mu.Unlock() - codexTokenCache.token = "" - codexTokenCache.expiresAt = time.Time{} - appendLog("[codex-auth] JWT cache invalidated") -} - -func mintCodexJWT(sessionCookie string) (string, error) { - // CRITICAL: route through HTTP_PROXY / HTTPS_PROXY (Webshare). Direct - // Railway IPs get stuck in Vercel's bot-ban loop after a few mints. - tr := &http.Transport{DisableKeepAlives: true, Proxy: http.ProxyFromEnvironment} - client := &http.Client{Timeout: 15 * time.Second, Transport: tr} - - body := map[string]interface{}{ - "operationName": "CreateApiToken", - "query": "mutation CreateApiToken { createApiTokens(input: { count: 1 }) { token } }", - "variables": map[string]interface{}{}, - } - bb, _ := json.Marshal(body) - req, _ := http.NewRequest("POST", "https://www.defined.fi/api", bytes.NewBuffer(bb)) - req.Header.Set("Accept", "application/json") - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Origin", "https://www.defined.fi") - req.Header.Set("Referer", "https://www.defined.fi/") - req.Header.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36") - req.Header.Set("sec-ch-ua", `"Not_A Brand";v="8", "Chromium";v="131", "Google Chrome";v="131"`) - req.Header.Set("sec-ch-ua-mobile", "?0") - req.Header.Set("sec-ch-ua-platform", `"macOS"`) - req.Header.Set("sec-fetch-dest", "empty") - req.Header.Set("sec-fetch-mode", "cors") - req.Header.Set("sec-fetch-site", "same-origin") - req.AddCookie(&http.Cookie{Name: "session", Value: sessionCookie}) - - resp, err := client.Do(req) - if err != nil { - return "", err - } - defer resp.Body.Close() - rb, _ := io.ReadAll(resp.Body) - if resp.StatusCode == 429 { - return "", fmt.Errorf("rate limited (429)") - } - if resp.StatusCode != 200 { - return "", fmt.Errorf("status=%d body=%s", resp.StatusCode, snippet(rb)) - } - var tr2 definedTokenResponse - if err := json.Unmarshal(rb, &tr2); err != nil { - return "", err - } - if len(tr2.Data.CreateApiTokens) == 0 { - return "", fmt.Errorf("no token in response") - } - return tr2.Data.CreateApiTokens[0].Token, nil -} - -func snippet(b []byte) string { - if len(b) > 200 { - return string(b[:200]) - } - return string(b) -} diff --git a/harnesses/pm-freshness-bench/cmd/script/config.go b/harnesses/pm-freshness-bench/cmd/script/config.go deleted file mode 100644 index fb5784e4..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/config.go +++ /dev/null @@ -1,43 +0,0 @@ -package main - -import ( - "os" - "strconv" - "strings" -) - -type Config struct { - DefinedSessionCookie string - MobulaApiKey string - LogsToken string - - // Kalshi T0 polls /v1/social/trades (public, unauthenticated). - // The default 5s cadence matches the upstream CloudFront max-age=10. - KalshiPollIntervalSec int - - RefreshMarketsIntervalSec int - BasketSize int -} - -func loadConfig() Config { - return Config{ - DefinedSessionCookie: strings.TrimSpace(os.Getenv("DEFINED_SESSION_COOKIE")), - MobulaApiKey: strings.TrimSpace(os.Getenv("MOBULA_API_KEY")), - LogsToken: strings.TrimSpace(os.Getenv("LOGS_TOKEN")), - KalshiPollIntervalSec: intEnv("KALSHI_POLL_INTERVAL_SEC", 5), - RefreshMarketsIntervalSec: intEnv("REFRESH_MARKETS_INTERVAL_SEC", 300), - BasketSize: intEnv("BASKET_SIZE", 20), - } -} - -func intEnv(key string, dflt int) int { - v := strings.TrimSpace(os.Getenv(key)) - if v == "" { - return dflt - } - n, err := strconv.Atoi(v) - if err != nil { - return dflt - } - return n -} diff --git a/harnesses/pm-freshness-bench/cmd/script/correlator.go b/harnesses/pm-freshness-bench/cmd/script/correlator.go deleted file mode 100644 index d825e265..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/correlator.go +++ /dev/null @@ -1,199 +0,0 @@ -package main - -import ( - "fmt" - "math" - "strings" - "sync" - "time" -) - -// EventSig is the cross-provider signature we use to match a single trade -// across the 3 streams. After empirical sampling of live traffic we know: -// - transactionHash: not on Mobula's payload, can't use -// - outcomeId: encoding differs (Codex wraps in composite scheme) -// - amountUSD: SEMANTICS differ — Mobula appears to aggregate multiple -// fills into one event while Polymarket fires one event per fill, -// so the dollar amount on the same trade ranges 1.4× to 38× off -// -// What ALL three providers report consistently for the same market -// activity: conditionId + the 3-decimal trade price + a coarse 5s time -// bucket. That's the signature. Trade-off: two distinct trades in the -// same market at the same price within a 5-second window collapse into -// one key — acceptable because we record time-to-first-arrival per -// provider, which remains representative of the freshness even when -// collapsed. -type EventSig struct { - Venue string // "polymarket" or "kalshi" — distinct venues never match each other - ConditionId string // Polymarket conditionId OR Kalshi market_ticker, lowercased - PriceMilli int // priceUSD * 1000 rounded - BucketSec int64 // floor(trade time / 5s) — coarsens skew between providers -} - -func (s EventSig) Key() string { - v := s.Venue - if v == "" { - v = "polymarket" - } - return fmt.Sprintf("%s|%s|%d|%d", v, s.ConditionId, s.PriceMilli, s.BucketSec) -} - -// Arrival records one provider's receipt of an event keyed by EventSig. -type Arrival struct { - Provider string // "polymarket", "codex", "mobula" - Kind string // "trade" or "price" - RecvUnix float64 -} - -type Correlator struct { - mu sync.Mutex - // per-signature, per-provider earliest arrival within the retention window - byKey map[string]map[string]Arrival - // signature key -> T0 arrival time, where T0 is the direct-venue source - // (Polymarket CLOB for polymarket venue, Kalshi WS for kalshi venue). - // Used for fast delta computation when a follower reports later. - t0 map[string]float64 - // LRU eviction list (key + insertion time) - insertOrder []keyAt -} - -type keyAt struct { - Key string - At float64 -} - -const retention = 90.0 // seconds; trades that don't match within 90s are forgotten - -func NewCorrelator() *Correlator { - return &Correlator{ - byKey: map[string]map[string]Arrival{}, - t0: map[string]float64{}, - } -} - -// isT0Provider returns true when the given provider is the direct-venue -// source for the given venue (i.e. the reference clock we measure lag from). -func isT0Provider(provider, venue string) bool { - switch venue { - case "kalshi": - return provider == "kalshi" - default: - return provider == "polymarket" - } -} - -func (c *Correlator) Add(sig EventSig, a Arrival) { - c.mu.Lock() - defer c.mu.Unlock() - c.evictExpired(a.RecvUnix) - - venue := sig.Venue - if venue == "" { - venue = "polymarket" - } - - key := sig.Key() - if _, ok := c.byKey[key]; !ok { - c.byKey[key] = map[string]Arrival{} - c.insertOrder = append(c.insertOrder, keyAt{Key: key, At: a.RecvUnix}) - } - if existing, ok := c.byKey[key][a.Provider]; ok && existing.RecvUnix <= a.RecvUnix { - return - } - c.byKey[key][a.Provider] = a - - if isT0Provider(a.Provider, venue) { - c.t0[key] = a.RecvUnix - for prov, follower := range c.byKey[key] { - if isT0Provider(prov, venue) { - continue - } - observeDelta(prov, venue, follower.Kind, follower.RecvUnix-a.RecvUnix) - } - return - } - - if t0, ok := c.t0[key]; ok { - observeDelta(a.Provider, venue, a.Kind, a.RecvUnix-t0) - } -} - -func observeDelta(provider, venue, kind string, deltaSec float64) { - if deltaSec < 0 { - deltaSec = 0 - } - ms := deltaSec * 1000 - if math.IsNaN(ms) || math.IsInf(ms, 0) { - return - } - freshnessDelta.WithLabelValues(provider, venue, kind).Observe(ms) - matchedTotal.WithLabelValues(provider, venue, kind).Inc() -} - -func (c *Correlator) evictExpired(now float64) { - cutoff := now - retention - cut := 0 - for cut < len(c.insertOrder) && c.insertOrder[cut].At < cutoff { - k := c.insertOrder[cut].Key - delete(c.byKey, k) - delete(c.t0, k) - cut++ - } - if cut > 0 { - c.insertOrder = c.insertOrder[cut:] - } -} - -func priceToMilli(price string) int { - f := parseF(price) - return int(math.Round(f * 1000)) -} - -func amountToCents(s string) int64 { - return int64(math.Round(parseF(s) * 100)) -} - -func amountFloatToCents(f float64) int64 { - return int64(math.Round(f * 100)) -} - -func parseF(s string) float64 { - if s == "" { - return 0 - } - var f float64 - fmt.Sscanf(strings.TrimSpace(s), "%f", &f) - return f -} - -func bucketSec(t float64) int64 { - return int64(t / 5) -} - -func nowSec() float64 { - return float64(time.Now().UnixNano()) / 1e9 -} - -// logSigSample prints the first 5 signatures per provider so we can eyeball -// cross-provider alignment when matched_total stays at zero. Pure debug -// instrumentation, dropped from output once we trust the matching. -var ( - sigSampleMu sync.Mutex - sigSamples = map[string]int{} -) - -func logSigSample(provider string, s EventSig) { - sigSampleMu.Lock() - defer sigSampleMu.Unlock() - key := provider + "/" + s.Venue - if sigSamples[key] >= 5 { - return - } - sigSamples[key]++ - cid := s.ConditionId - if len(cid) > 14 { - cid = cid[:14] - } - appendLog("[sig %s/%s #%d] cond=%s price=%d bucket=%d", - provider, s.Venue, sigSamples[key], cid, s.PriceMilli, s.BucketSec) -} diff --git a/harnesses/pm-freshness-bench/cmd/script/kalshi.go b/harnesses/pm-freshness-bench/cmd/script/kalshi.go deleted file mode 100644 index 13d576f5..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/kalshi.go +++ /dev/null @@ -1,270 +0,0 @@ -package main - -import ( - "context" - "encoding/json" - "fmt" - "net/http" - "sort" - "strings" - "sync" - "time" -) - -// Kalshi public trade feed. The /v1/social/trades endpoint is the same one -// kalshi.com homepage trade ticker hits; it is unauthenticated, CloudFront -// fronted (max-age=10s), no geo block on read, no auth header required. -// -// Why REST and not the official WebSocket: the WS at -// external-api-ws.kalshi.com requires RSA-PSS signed headers from an -// account with US KYC, and returns 403 from non-US IPs even with valid -// auth. The social/trades JSON carries `create_date` at microsecond -// precision -- that IS the venue T0, independent of our poll cadence. -// The poller's only job is to surface trade signatures so the correlator -// can match them against Codex's Kalshi stream. -// -// Calling without `series_ticker` returns a global feed of the most -// recent ~100 trades across all series, so a single HTTP call per cycle -// covers every active market. The Kalshi basket (used by codex.go to -// filter Kalshi events on its firehose) is built dynamically from the -// tickers actually seen in those trades, which auto-targets the active -// markets without needing a separate /markets call that returns zero -// volumes on most rows. -const kalshiSocialTradesURL = "https://api.elections.kalshi.com/v1/social/trades" - -func runKalshi(ctx context.Context, cfg Config, codexKalshiCh chan<- []Market) { - every := time.Duration(cfg.KalshiPollIntervalSec) * time.Second - if every < 2*time.Second { - every = 5 * time.Second - } - - seen := newSeenCache(8192) - tickerFreq := newTickerFreq(cfg.BasketSize) - - appendLog("[kalshi] REST poller started (global feed), interval=%s", every) - tick := time.NewTicker(every) - defer tick.Stop() - for { - select { - case <-ctx.Done(): - return - case <-tick.C: - } - if err := pollKalshiGlobal(ctx, seen, tickerFreq, codexKalshiCh); err != nil { - fetchErrors.WithLabelValues("kalshi", "kalshi", classify(err)).Inc() - health.WithLabelValues("kalshi", "kalshi").Set(0) - appendLog("[kalshi] poll err: %v", err) - } - } -} - -// pollKalshiGlobal pages through /v1/social/trades following the -// returned cursor until either every trade in a page is already in the -// seen cache (we caught up with last cycle) or we hit the page cap. -// One call returns ~100 trades covering ~3 seconds of venue time; with -// a 5 s poll cadence two pages cover the gap with overlap, the cap of -// 5 is generous headroom for traffic spikes. -const kalshiMaxPagesPerPoll = 5 - -func pollKalshiGlobal(ctx context.Context, seen *seenCache, freq *tickerFreq, codexKalshiCh chan<- []Market) error { - cursor := "" - for page := 0; page < kalshiMaxPagesPerPoll; page++ { - newOnThisPage, nextCursor, err := fetchKalshiPage(ctx, cursor, seen, freq) - if err != nil { - return fmt.Errorf("page %d: %w", page+1, err) - } - // Stop once a whole page returns zero new trades -- means we - // crossed back into the previous poll's window. - if newOnThisPage == 0 { - break - } - if nextCursor == "" { - break - } - cursor = nextCursor - } - // Push the latest observed-tickers basket so codex.go's isKnownKalshi - // accepts the same set of markets we're polling. The channel send is - // non-blocking; codex.go re-reads currentKalshiBasket on (re)connect - // anyway, so dropping an update is safe — the next poll will retry. - basket := freq.topTickers() - basketMu.Lock() - kalshiBasket = basket - basketMu.Unlock() - basketSize.WithLabelValues("kalshi").Set(float64(len(basket))) - if codexKalshiCh != nil && len(basket) > 0 { - select { - case codexKalshiCh <- basket: - default: - } - } - return nil -} - -func fetchKalshiPage(ctx context.Context, cursor string, seen *seenCache, freq *tickerFreq) (int, string, error) { - url := kalshiSocialTradesURL - if cursor != "" { - url += "?cursor=" + cursor - } - req, _ := http.NewRequestWithContext(ctx, "GET", url, nil) - req.Header.Set("Accept", "application/json") - req.Header.Set("User-Agent", "openchainbench-pm-freshness-bench") - client := &http.Client{Timeout: 8 * time.Second} - resp, err := client.Do(req) - if err != nil { - return 0, "", err - } - defer resp.Body.Close() - if resp.StatusCode != 200 { - return 0, "", fmt.Errorf("status_%d", resp.StatusCode) - } - var payload struct { - Trades []struct { - TradeID string `json:"trade_id"` - Ticker string `json:"ticker"` - Price int `json:"price"` - CreateDate string `json:"create_date"` - } `json:"trades"` - Cursor string `json:"cursor"` - } - if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil { - return 0, "", err - } - newCount := 0 - for _, t := range payload.Trades { - if t.TradeID == "" || t.Ticker == "" { - continue - } - if seen.contains(t.TradeID) { - continue - } - seen.add(t.TradeID) - ts, err := time.Parse(time.RFC3339Nano, t.CreateDate) - if err != nil { - continue - } - newCount++ - eventsTotal.WithLabelValues("kalshi", "kalshi", "trade").Inc() - health.WithLabelValues("kalshi", "kalshi").Set(1) - markAlive("kalshi", "kalshi") - - tickerLower := strings.ToLower(t.Ticker) - freq.bump(tickerLower) - - sig := EventSig{ - Venue: "kalshi", - ConditionId: tickerLower, - PriceMilli: t.Price * 10, - BucketSec: bucketSec(float64(ts.UnixNano()) / 1e9), - } - logSigSample("kalshi", sig) - correlator.Add(sig, Arrival{Provider: "kalshi", Kind: "trade", RecvUnix: float64(ts.UnixNano()) / 1e9}) - } - return newCount, payload.Cursor, nil -} - -// tickerFreq tracks how often each market_ticker has appeared in the -// recent trade stream, with exponential decay so quiet markets fade. -// We expose the top N as the "active basket" -- Codex's firehose filter -// reads from this set. -type tickerFreq struct { - mu sync.Mutex - counts map[string]float64 - cap int - lastDec time.Time -} - -func newTickerFreq(cap int) *tickerFreq { - return &tickerFreq{counts: map[string]float64{}, cap: cap, lastDec: time.Now()} -} - -func (f *tickerFreq) bump(ticker string) { - f.mu.Lock() - defer f.mu.Unlock() - f.decayLocked() - f.counts[ticker]++ -} - -// decayLocked applies an exponential decay so a market that stops -// trading drops out of the top-N after a few minutes. Tuned so a market -// at "1 trade per minute" decays past a 0.1 floor in ~10 minutes. -func (f *tickerFreq) decayLocked() { - now := time.Now() - elapsed := now.Sub(f.lastDec).Seconds() - if elapsed < 30 { - return - } - f.lastDec = now - decay := 0.95 * (elapsed / 60) // ~5% per minute - if decay > 0.5 { - decay = 0.5 - } - factor := 1 - decay - for k, v := range f.counts { - nv := v * factor - if nv < 0.1 { - delete(f.counts, k) - } else { - f.counts[k] = nv - } - } -} - -func (f *tickerFreq) topTickers() []Market { - f.mu.Lock() - defer f.mu.Unlock() - f.decayLocked() - type kv struct { - ticker string - score float64 - } - all := make([]kv, 0, len(f.counts)) - for t, c := range f.counts { - all = append(all, kv{t, c}) - } - sort.Slice(all, func(i, j int) bool { return all[i].score > all[j].score }) - if len(all) > f.cap { - all = all[:f.cap] - } - out := make([]Market, len(all)) - for i, kv := range all { - out[i] = Market{Slug: kv.ticker, ConditionId: kv.ticker, Vol24h: kv.score} - } - return out -} - -// seenCache is a small FIFO set just enough to dedup trade_id across polls. -// Each poll typically returns ~100 trades and we re-fetch every 5s, so 8k -// entries covers ~7 minutes of unique IDs at peak. -type seenCache struct { - mu sync.Mutex - set map[string]struct{} - order []string - cap int -} - -func newSeenCache(cap int) *seenCache { - return &seenCache{set: make(map[string]struct{}, cap), cap: cap} -} - -func (s *seenCache) contains(k string) bool { - s.mu.Lock() - defer s.mu.Unlock() - _, ok := s.set[k] - return ok -} - -func (s *seenCache) add(k string) { - s.mu.Lock() - defer s.mu.Unlock() - if _, ok := s.set[k]; ok { - return - } - s.set[k] = struct{}{} - s.order = append(s.order, k) - if len(s.order) > s.cap { - old := s.order[0] - s.order = s.order[1:] - delete(s.set, old) - } -} diff --git a/harnesses/pm-freshness-bench/cmd/script/log_buffer.go b/harnesses/pm-freshness-bench/cmd/script/log_buffer.go deleted file mode 100644 index 2fb7e3df..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/log_buffer.go +++ /dev/null @@ -1,58 +0,0 @@ -package main - -import ( - "fmt" - "net/http" - "sync" - "time" -) - -// Ring buffer mirroring the pattern used in the aggregator-latency-benchmark -// and wallet-labels miniapps. Captures the last ~5000 log lines for the -// `/logs?tail=N` endpoint. -const ringCapacity = 5000 - -type logEntry struct { - At time.Time - Line string -} - -var ( - ringMu sync.Mutex - ring = make([]logEntry, 0, ringCapacity) -) - -func appendLog(format string, args ...interface{}) { - line := fmt.Sprintf(format, args...) - ts := time.Now().UTC().Format("15:04:05.000") - fmt.Printf("[%s] %s\n", ts, line) - ringMu.Lock() - if len(ring) >= ringCapacity { - ring = ring[1:] - } - ring = append(ring, logEntry{At: time.Now(), Line: line}) - ringMu.Unlock() -} - -func setupLogsEndpoint(mux *http.ServeMux, token string) { - mux.HandleFunc("/logs", func(w http.ResponseWriter, r *http.Request) { - if token != "" && r.Header.Get("X-Logs-Token") != token { - http.Error(w, "forbidden", http.StatusForbidden) - return - } - tail := 200 - if q := r.URL.Query().Get("tail"); q != "" { - fmt.Sscanf(q, "%d", &tail) - } - ringMu.Lock() - defer ringMu.Unlock() - start := 0 - if len(ring) > tail { - start = len(ring) - tail - } - w.Header().Set("Content-Type", "text/plain; charset=utf-8") - for _, e := range ring[start:] { - fmt.Fprintf(w, "[%s] %s\n", e.At.UTC().Format("15:04:05.000"), e.Line) - } - }) -} diff --git a/harnesses/pm-freshness-bench/cmd/script/main.go b/harnesses/pm-freshness-bench/cmd/script/main.go deleted file mode 100644 index 75ce8e63..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/main.go +++ /dev/null @@ -1,142 +0,0 @@ -package main - -import ( - "context" - "os" - "os/signal" - "strings" - "sync" - "syscall" - "time" -) - -var correlator = NewCorrelator() - -func main() { - cfg := loadConfig() - appendLog("[boot] pm-freshness-bench starting (basket=%d, refresh=%ds)", cfg.BasketSize, cfg.RefreshMarketsIntervalSec) - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - // graceful shutdown - sigCh := make(chan os.Signal, 1) - signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) - go func() { - <-sigCh - appendLog("[boot] shutdown signal received") - cancel() - }() - - // fan-out: every Polymarket basket refresh broadcasts on these per-consumer channels. - polyCh := make(chan []Market, 4) - codexPolyCh := make(chan []Market, 4) - mobulaCh := make(chan []Market, 4) - onChange := func(ms []Market) { - for _, ch := range []chan []Market{polyCh, codexPolyCh, mobulaCh} { - select { - case ch <- ms: - default: - // channel full — provider will pick up the change on the next refresh - } - } - } - - // Kalshi has its own basket which is now built dynamically by the - // Kalshi T0 client itself from the global /v1/social/trades feed. The - // /trade-api/v2/markets endpoint returns zero volume on every row so - // the previous markets-based refresh produced a basket of random - // inactive series; the observation-based basket auto-targets the - // active markets instead. - codexKalshiCh := make(chan []Market, 4) - - // health gauges decay: any (provider, venue) pair that hasn't published in 60s drops to 0. - go healthDecayLoop(ctx) - - var wg sync.WaitGroup - wg.Add(5) - go func() { - defer wg.Done() - refreshLoop(ctx, time.Duration(cfg.RefreshMarketsIntervalSec)*time.Second, cfg.BasketSize, onChange) - }() - go func() { defer wg.Done(); runPolymarket(ctx, polyCh) }() - go func() { defer wg.Done(); runCodex(ctx, cfg, codexPolyCh, codexKalshiCh) }() - go func() { defer wg.Done(); runMobula(ctx, cfg, mobulaCh) }() - go func() { defer wg.Done(); runKalshi(ctx, cfg, codexKalshiCh) }() - - // :2112 hardcoded per the OCB convention — Railway $PORT is intentionally ignored - // so the shared Prom can scrape every harness on the same well-known port. - go func() { - if err := startMetricsServer(":2112", cfg.LogsToken); err != nil { - appendLog("[metrics] server err: %v", err) - } - }() - - wg.Wait() -} - -// lastEventAt is updated by every WS client when it observes a real event, -// keyed by (provider, venue). The decay loop reads it and drops the -// corresponding health gauge to 0 when nothing has arrived in the last 60s. -type provVenue struct{ provider, venue string } - -var ( - lastEventMu sync.Mutex - lastEventAt = map[provVenue]time.Time{} -) - -// healthPairs is the static list of (provider, venue) tuples we track. -// Polymarket venue: 3 providers (polymarket T0, codex follower, mobula follower). -// Kalshi venue: 2 providers (kalshi T0, codex follower). -var healthPairs = []provVenue{ - {"polymarket", "polymarket"}, - {"codex", "polymarket"}, - {"mobula", "polymarket"}, - {"kalshi", "kalshi"}, - {"codex", "kalshi"}, -} - -func markAlive(provider, venue string) { - lastEventMu.Lock() - lastEventAt[provVenue{provider, venue}] = time.Now() - lastEventMu.Unlock() -} - -func healthDecayLoop(ctx context.Context) { - t := time.NewTicker(10 * time.Second) - defer t.Stop() - for { - select { - case <-ctx.Done(): - return - case <-t.C: - lastEventMu.Lock() - for _, pv := range healthPairs { - last, ok := lastEventAt[pv] - if !ok || time.Since(last) > 60*time.Second { - health.WithLabelValues(pv.provider, pv.venue).Set(0) - } - } - lastEventMu.Unlock() - } - } -} - -func classify(err error) string { - if err == nil { - return "none" - } - s := strings.ToLower(err.Error()) - switch { - case strings.Contains(s, "429") || strings.Contains(s, "rate limited"): - return "rate_limit" - case strings.Contains(s, "401") || strings.Contains(s, "403") || strings.Contains(s, "4403") || strings.Contains(s, "auth"): - return "auth" - case strings.Contains(s, "timeout"): - return "timeout" - case strings.Contains(s, "eof") || strings.Contains(s, "reset"): - return "conn_drop" - default: - return "other" - } -} diff --git a/harnesses/pm-freshness-bench/cmd/script/markets.go b/harnesses/pm-freshness-bench/cmd/script/markets.go deleted file mode 100644 index 2fd4d799..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/markets.go +++ /dev/null @@ -1,220 +0,0 @@ -package main - -import ( - "context" - "encoding/json" - "fmt" - "io" - "net/http" - "sort" - "strings" - "sync" - "time" -) - -// A market we subscribe to across every provider. Each Polymarket market -// has a conditionId (0x-prefixed hex) shared by both outcomes, and a pair -// of clobTokenIds (decimal strings) — one per Yes/No outcome. -// For Kalshi the ConditionId field holds the market_ticker and ClobTokenIds -// is empty — the rest of the pipeline only needs ConditionId for matching. -type Market struct { - Slug string - ConditionId string // Polymarket conditionId OR Kalshi market_ticker (lowercased) - ClobTokenIds []string // 2 ids, the Polymarket WS `asset_id` per outcome (Polymarket only) - Vol24h float64 -} - -var ( - basketMu sync.RWMutex - basket []Market // Polymarket basket - kalshiBasket []Market -) - -// currentBasket returns a snapshot of the currently-subscribed Polymarket -// markets. Callers should treat the slice as read-only. -func currentBasket() []Market { - basketMu.RLock() - defer basketMu.RUnlock() - out := make([]Market, len(basket)) - copy(out, basket) - return out -} - -// currentKalshiBasket returns a snapshot of the currently-subscribed Kalshi -// markets. Same read-only contract as currentBasket. -func currentKalshiBasket() []Market { - basketMu.RLock() - defer basketMu.RUnlock() - out := make([]Market, len(kalshiBasket)) - copy(out, kalshiBasket) - return out -} - -// refreshLoop polls gamma-api on an interval and updates the shared basket. -// Connections are not torn down here — providers re-subscribe diff-style -// on every refresh (see their respective files). -func refreshLoop(ctx context.Context, every time.Duration, size int, onChange func([]Market)) { - tick := time.NewTicker(every) - defer tick.Stop() - if err := refreshBasket(ctx, size, onChange); err != nil { - appendLog("[markets] initial refresh err: %v", err) - } - for { - select { - case <-ctx.Done(): - return - case <-tick.C: - if err := refreshBasket(ctx, size, onChange); err != nil { - appendLog("[markets] refresh err: %v", err) - } - } - } -} - -func refreshBasket(ctx context.Context, size int, onChange func([]Market)) error { - url := fmt.Sprintf("https://gamma-api.polymarket.com/markets?active=true&closed=false&order=volume24hr&ascending=false&limit=%d", size) - req, _ := http.NewRequestWithContext(ctx, "GET", url, nil) - resp, err := http.DefaultClient.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() - body, _ := io.ReadAll(resp.Body) - if resp.StatusCode != 200 { - return fmt.Errorf("gamma-api status=%d", resp.StatusCode) - } - var raw []struct { - Slug string `json:"slug"` - ConditionId string `json:"conditionId"` - ClobTokenIds string `json:"clobTokenIds"` // JSON-encoded array of strings - Volume24hr float64 `json:"volume24hr"` - } - if err := json.Unmarshal(body, &raw); err != nil { - return err - } - out := make([]Market, 0, len(raw)) - for _, r := range raw { - if r.ConditionId == "" || r.ClobTokenIds == "" { - continue - } - var ids []string - if err := json.Unmarshal([]byte(r.ClobTokenIds), &ids); err != nil { - continue - } - if len(ids) < 2 { - continue - } - out = append(out, Market{ - Slug: r.Slug, - ConditionId: strings.ToLower(r.ConditionId), - ClobTokenIds: ids, - Vol24h: r.Volume24hr, - }) - } - - basketMu.Lock() - basket = out - basketMu.Unlock() - basketSize.WithLabelValues("polymarket").Set(float64(len(out))) - appendLog("[markets] refreshed basket: %d markets, top vol=$%.0fk", len(out), firstVol(out)/1000) - if onChange != nil { - onChange(out) - } - return nil -} - -func firstVol(ms []Market) float64 { - if len(ms) == 0 { - return 0 - } - return ms[0].Vol24h -} - -// refreshKalshiLoop mirrors refreshLoop for Kalshi. Kalshi's public markets -// endpoint is a separate host so we drive it on its own goroutine. -func refreshKalshiLoop(ctx context.Context, every time.Duration, size int, onChange func([]Market)) { - tick := time.NewTicker(every) - defer tick.Stop() - if err := refreshKalshiBasket(ctx, size, onChange); err != nil { - appendLog("[kalshi-markets] initial refresh err: %v", err) - } - for { - select { - case <-ctx.Done(): - return - case <-tick.C: - if err := refreshKalshiBasket(ctx, size, onChange); err != nil { - appendLog("[kalshi-markets] refresh err: %v", err) - } - } - } -} - -func refreshKalshiBasket(ctx context.Context, size int, onChange func([]Market)) error { - // Kalshi's elections host is the public unauthenticated read endpoint. - // We pull a wider page than `size` because the API doesn't expose a - // sort-by-volume parameter — we rank client-side and trim. - url := "https://api.elections.kalshi.com/trade-api/v2/markets?status=open&limit=200" - req, _ := http.NewRequestWithContext(ctx, "GET", url, nil) - resp, err := http.DefaultClient.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() - body, _ := io.ReadAll(resp.Body) - if resp.StatusCode != 200 { - return fmt.Errorf("kalshi markets status=%d", resp.StatusCode) - } - var raw struct { - Markets []struct { - Ticker string `json:"ticker"` - Volume24h float64 `json:"volume_24h"` - Volume float64 `json:"volume"` - LastPrice float64 `json:"last_price"` - } `json:"markets"` - } - if err := json.Unmarshal(body, &raw); err != nil { - return err - } - // Rank by 24h volume when available, fall back to all-time volume. - type scored struct { - Market - score float64 - } - scoredList := make([]scored, 0, len(raw.Markets)) - for _, r := range raw.Markets { - if r.Ticker == "" { - continue - } - s := r.Volume24h - if s == 0 { - s = r.Volume - } - scoredList = append(scoredList, scored{ - Market: Market{ - Slug: r.Ticker, - ConditionId: strings.ToLower(r.Ticker), - Vol24h: r.Volume24h, - }, - score: s, - }) - } - sort.Slice(scoredList, func(i, j int) bool { return scoredList[i].score > scoredList[j].score }) - if len(scoredList) > size { - scoredList = scoredList[:size] - } - out := make([]Market, len(scoredList)) - for i, s := range scoredList { - out[i] = s.Market - } - - basketMu.Lock() - kalshiBasket = out - basketMu.Unlock() - basketSize.WithLabelValues("kalshi").Set(float64(len(out))) - appendLog("[kalshi-markets] refreshed basket: %d markets, top vol24h=$%.0fk", len(out), firstVol(out)/1000) - if onChange != nil { - onChange(out) - } - return nil -} diff --git a/harnesses/pm-freshness-bench/cmd/script/metrics.go b/harnesses/pm-freshness-bench/cmd/script/metrics.go deleted file mode 100644 index ce9090f4..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/metrics.go +++ /dev/null @@ -1,59 +0,0 @@ -package main - -import ( - "net/http" - - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" - "github.com/prometheus/client_golang/prometheus/promhttp" -) - -var commonLabels = prometheus.Labels{"benchmark": "pm-freshness"} - -var ( - // Freshness delta vs Polymarket T0, in ms. Buckets sized for sub-second - // to ~30s lag (Codex p50 is around 4s). - freshnessDelta = promauto.NewHistogramVec(prometheus.HistogramOpts{ - Name: "pm_freshness_delta_ms", - Help: "Per-event delta in ms between provider arrival and venue T0 arrival, matched by (conditionId, price, time-bucket, venue).", - Buckets: []float64{10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000, 60000}, - ConstLabels: commonLabels, - }, []string{"provider", "venue", "kind"}) - - eventsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ - Name: "pm_events_total", - Help: "Raw events received per provider/venue/kind.", - ConstLabels: commonLabels, - }, []string{"provider", "venue", "kind"}) - - matchedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ - Name: "pm_matched_total", - Help: "Provider events that matched a venue T0 event by signature.", - ConstLabels: commonLabels, - }, []string{"provider", "venue", "kind"}) - - fetchErrors = promauto.NewCounterVec(prometheus.CounterOpts{ - Name: "pm_fetch_errors_total", - Help: "Provider errors (connection, auth, parse).", - ConstLabels: commonLabels, - }, []string{"provider", "venue", "error_type"}) - - health = promauto.NewGaugeVec(prometheus.GaugeOpts{ - Name: "pm_health", - Help: "1 if the (provider, venue) published at least one event in the last 60s.", - ConstLabels: commonLabels, - }, []string{"provider", "venue"}) - - basketSize = promauto.NewGaugeVec(prometheus.GaugeOpts{ - Name: "pm_basket_size", - Help: "Number of markets currently subscribed per venue.", - ConstLabels: commonLabels, - }, []string{"venue"}) -) - -func startMetricsServer(addr, logsToken string) error { - mux := http.NewServeMux() - mux.Handle("/metrics", promhttp.Handler()) - setupLogsEndpoint(mux, logsToken) - return http.ListenAndServe(addr, mux) -} diff --git a/harnesses/pm-freshness-bench/cmd/script/mobula.go b/harnesses/pm-freshness-bench/cmd/script/mobula.go deleted file mode 100644 index 65a73c5c..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/mobula.go +++ /dev/null @@ -1,198 +0,0 @@ -package main - -import ( - "context" - "encoding/json" - "fmt" - "net/http" - "strings" - "sync" - "time" - - "nhooyr.io/websocket" -) - -const mobulaWS = "wss://pm-api-prod-eu.mobula.io" - -// Cloudflare on the Mobula PM gateway blocks the default Go-http-client/1.1 -// User-Agent silently — the subscribe ack comes back but no data frame -// ever fires. Spoofing a normal browser UA on the WS upgrade unblocks it. -const mobulaUA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" - -func runMobula(ctx context.Context, cfg Config, basketCh <-chan []Market) { - if cfg.MobulaApiKey == "" { - appendLog("[mobula] MOBULA_API_KEY not set — skipping Mobula provider") - return - } - - backoff := 5 * time.Second - for ctx.Err() == nil { - err := mobulaConnect(ctx, cfg.MobulaApiKey, basketCh) - if ctx.Err() != nil { - return - } - appendLog("[mobula] disconnected: %v — backing off %v", err, backoff) - fetchErrors.WithLabelValues("mobula", "polymarket", classify(err)).Inc() - select { - case <-ctx.Done(): - return - case <-time.After(backoff): - } - if backoff < 60*time.Second { - backoff *= 2 - } - } -} - -func mobulaConnect(ctx context.Context, apiKey string, basketCh <-chan []Market) error { - opts := &websocket.DialOptions{ - HTTPHeader: http.Header{ - "User-Agent": []string{mobulaUA}, - "Origin": []string{"https://mobula.io"}, - }, - } - conn, _, err := websocket.Dial(ctx, mobulaWS, opts) - if err != nil { - return fmt.Errorf("dial: %w", err) - } - defer conn.Close(websocket.StatusInternalError, "") - conn.SetReadLimit(8 * 1024 * 1024) - - var ( - subMu sync.Mutex - subscribed = map[string]bool{} // conditionId - ) - subscribeMissing := func(ms []Market) { - subMu.Lock() - defer subMu.Unlock() - for _, m := range ms { - if subscribed[m.ConditionId] { - continue - } - subscribed[m.ConditionId] = true - for _, channel := range []string{"pm-market-trade", "pm-market-price"} { - sub := map[string]interface{}{ - "event": channel, - "data": map[string]interface{}{ - "platform": "polymarket", - "marketId": m.ConditionId, - "authorization": apiKey, - }, - } - sb, _ := json.Marshal(sub) - if err := conn.Write(ctx, websocket.MessageText, sb); err != nil { - appendLog("[mobula] subscribe err: %v", err) - return - } - // pace the writes so Mobula doesn't drop them - time.Sleep(40 * time.Millisecond) - } - } - appendLog("[mobula] subscribed total=%d markets (trade+price)", len(subscribed)) - } - - if initial := currentBasket(); len(initial) > 0 { - subscribeMissing(initial) - } - go func() { - for { - select { - case <-ctx.Done(): - return - case ms := <-basketCh: - subscribeMissing(ms) - } - } - }() - - for { - _, b, err := conn.Read(ctx) - if err != nil { - return err - } - var m map[string]interface{} - if err := json.Unmarshal(b, &m); err != nil { - continue - } - evt, _ := m["event"].(string) - switch evt { - case "subscribed": - continue - case "error": - s := string(b) - if len(s) > 200 { - s = s[:200] - } - fetchErrors.WithLabelValues("mobula", "polymarket", "server_error").Inc() - appendLog("[mobula] error frame: %s", s) - continue - case "pm-market-trade": - handleMobulaTrade(m) - case "pm-market-price": - handleMobulaPrice(m) - } - } -} - -// Mobula's pm-market-trade payload is a flat object (no nested `data` on -// outbound), e.g. -// {"event":"pm-market-trade","marketId":"0xabc...","outcomeId":"60526...", -// "type":"buy","priceUSD":0.7,"sizeToken":12.34,"amountUsd":"8.64","date":1780...} -// Mobula's pm-market-price is similar but carries priceUSD/bestBid/bestAsk -// without size — we still record it as a "price" arrival for correlation -// against Polymarket `price_change`. -func handleMobulaTrade(m map[string]interface{}) { - eventsTotal.WithLabelValues("mobula", "polymarket", "trade").Inc() - health.WithLabelValues("mobula", "polymarket").Set(1) - markAlive("mobula", "polymarket") - - mid, _ := m["marketId"].(string) - cid := strings.ToLower(mid) - if cid == "" { - return - } - priceUSD, _ := m["priceUSD"].(float64) - ts := readMs(m, "date", "timestamp") - now := nowSec() - sig := EventSig{ - Venue: "polymarket", - ConditionId: cid, - PriceMilli: int(priceUSD*1000 + 0.5), - BucketSec: bucketSec(ts / 1000.0), - } - logSigSample("mobula", sig) - correlator.Add(sig, Arrival{Provider: "mobula", Kind: "trade", RecvUnix: now}) -} - -func handleMobulaPrice(m map[string]interface{}) { - eventsTotal.WithLabelValues("mobula", "polymarket", "price").Inc() - health.WithLabelValues("mobula", "polymarket").Set(1) - markAlive("mobula", "polymarket") - - mid, _ := m["marketId"].(string) - cid := strings.ToLower(mid) - if cid == "" { - return - } - priceUSD, _ := m["priceUSD"].(float64) - ts := readMs(m, "timestamp", "date") - now := nowSec() - correlator.Add(EventSig{ - Venue: "polymarket", - ConditionId: cid, - PriceMilli: int(priceUSD*1000 + 0.5), - BucketSec: bucketSec(ts / 1000.0), - }, Arrival{Provider: "mobula", Kind: "price", RecvUnix: now}) -} - -func readMs(m map[string]interface{}, keys ...string) float64 { - for _, k := range keys { - if v, ok := m[k].(float64); ok { - return v - } - if s, ok := m[k].(string); ok { - return parseF(s) - } - } - return 0 -} diff --git a/harnesses/pm-freshness-bench/cmd/script/polymarket.go b/harnesses/pm-freshness-bench/cmd/script/polymarket.go deleted file mode 100644 index 16a00953..00000000 --- a/harnesses/pm-freshness-bench/cmd/script/polymarket.go +++ /dev/null @@ -1,216 +0,0 @@ -package main - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "strings" - "sync" - "time" - - "nhooyr.io/websocket" -) - -const polymarketWS = "wss://ws-subscriptions-clob.polymarket.com/ws/market" - -func runPolymarket(ctx context.Context, basketCh <-chan []Market) { - // Maps clobTokenId → conditionId, refreshed from each basket update. - // Polymarket WS keys events by `asset_id` (= clobTokenId). The correlator - // keys by conditionId, so we need this mapping in O(1) on the hot path. - var ( - idMu sync.RWMutex - assetToCID = map[string]string{} - ) - - updateMap := func(ms []Market) { - idMu.Lock() - defer idMu.Unlock() - assetToCID = map[string]string{} - for _, m := range ms { - for _, aid := range m.ClobTokenIds { - assetToCID[aid] = strings.ToLower(m.ConditionId) - } - } - } - - lookupCID := func(assetId string) string { - idMu.RLock() - defer idMu.RUnlock() - return assetToCID[assetId] - } - - // Reconnect loop. On every connection we read the current basket and - // resubscribe with the full asset list. - backoff := 5 * time.Second - for ctx.Err() == nil { - err := dialAndStream(ctx, basketCh, updateMap, lookupCID) - if ctx.Err() != nil { - return - } - appendLog("[poly] connection ended: %v — reconnecting in %v", err, backoff) - fetchErrors.WithLabelValues("polymarket", "polymarket", classify(err)).Inc() - select { - case <-ctx.Done(): - return - case <-time.After(backoff): - } - if backoff < 30*time.Second { - backoff *= 2 - } - } -} - -func dialAndStream(ctx context.Context, basketCh <-chan []Market, onBasket func([]Market), lookupCID func(string) string) error { - conn, _, err := websocket.Dial(ctx, polymarketWS, nil) - if err != nil { - return fmt.Errorf("dial: %w", err) - } - defer conn.Close(websocket.StatusInternalError, "") - conn.SetReadLimit(8 * 1024 * 1024) - - subscribed := map[string]bool{} // assetId -> true - - subscribeMissing := func(ms []Market) { - assetIds := []string{} - for _, m := range ms { - for _, aid := range m.ClobTokenIds { - if !subscribed[aid] { - assetIds = append(assetIds, aid) - subscribed[aid] = true - } - } - } - if len(assetIds) == 0 { - return - } - sub := map[string]interface{}{ - "assets_ids": assetIds, - "type": "market", - "custom_feature_enabled": true, - } - sb, _ := json.Marshal(sub) - if err := conn.Write(ctx, websocket.MessageText, sb); err != nil { - appendLog("[poly] subscribe write err: %v", err) - return - } - appendLog("[poly] subscribed +%d assets (total=%d)", len(assetIds), len(subscribed)) - } - - // initial basket - initial := currentBasket() - if len(initial) > 0 { - onBasket(initial) - subscribeMissing(initial) - } - - // ping loop (server expects literal text "PING" every 10s) - pingCtx, pingCancel := context.WithCancel(ctx) - defer pingCancel() - go func() { - t := time.NewTicker(10 * time.Second) - defer t.Stop() - for { - select { - case <-pingCtx.Done(): - return - case <-t.C: - if err := conn.Write(pingCtx, websocket.MessageText, []byte("PING")); err != nil { - return - } - } - } - }() - - // basket-update goroutine — react to new markets coming in - go func() { - for { - select { - case <-pingCtx.Done(): - return - case ms := <-basketCh: - onBasket(ms) - subscribeMissing(ms) - } - } - }() - - for { - _, b, err := conn.Read(ctx) - if err != nil { - return err - } - if bytes.Equal(b, []byte("PONG")) { - continue - } - t := bytes.TrimSpace(b) - var arr []map[string]interface{} - if len(t) > 0 && t[0] == '[' { - json.Unmarshal(b, &arr) - } else { - var one map[string]interface{} - if json.Unmarshal(b, &one) == nil { - arr = []map[string]interface{}{one} - } - } - for _, ev := range arr { - eventType, _ := ev["event_type"].(string) - tsMs, _ := parseTimestampMsField(ev["timestamp"]) - assetId, _ := ev["asset_id"].(string) - cid := lookupCID(assetId) - if cid == "" { - continue - } - now := nowSec() - switch eventType { - case "last_trade_price": - eventsTotal.WithLabelValues("polymarket", "polymarket", "trade").Inc() - price, _ := ev["price"].(string) - health.WithLabelValues("polymarket", "polymarket").Set(1) - markAlive("polymarket", "polymarket") - sig := EventSig{ - Venue: "polymarket", - ConditionId: cid, - PriceMilli: priceToMilli(price), - BucketSec: bucketSec(tsMs / 1000.0), - } - logSigSample("polymarket", sig) - correlator.Add(sig, Arrival{Provider: "polymarket", Kind: "trade", RecvUnix: now}) - case "price_change": - eventsTotal.WithLabelValues("polymarket", "polymarket", "price").Inc() - health.WithLabelValues("polymarket", "polymarket").Set(1) - markAlive("polymarket", "polymarket") - if changes, ok := ev["price_changes"].([]interface{}); ok { - for _, ch := range changes { - chm, _ := ch.(map[string]interface{}) - if chm == nil { - continue - } - price, _ := chm["price"].(string) - correlator.Add(EventSig{ - Venue: "polymarket", - ConditionId: cid, - PriceMilli: priceToMilli(price), - BucketSec: bucketSec(tsMs / 1000.0), - }, Arrival{Provider: "polymarket", Kind: "price", RecvUnix: now}) - } - } - } - } - } -} - -// Polymarket emits `timestamp` as a string of ms-since-epoch. Codex emits -// it as an int seconds-since-epoch. Mobula emits it as a number of ms. -// Normalize to ms float here. -func parseTimestampMsField(v interface{}) (float64, error) { - switch t := v.(type) { - case string: - return parseF(t), nil - case float64: - return t, nil - case int64: - return float64(t), nil - } - return 0, fmt.Errorf("unknown timestamp type") -} diff --git a/harnesses/pm-freshness-bench/go.mod b/harnesses/pm-freshness-bench/go.mod deleted file mode 100644 index b2c2cec0..00000000 --- a/harnesses/pm-freshness-bench/go.mod +++ /dev/null @@ -1,18 +0,0 @@ -module pm-freshness-bench - -go 1.22 - -require ( - github.com/prometheus/client_golang v1.19.1 - nhooyr.io/websocket v1.8.11 -) - -require ( - github.com/beorn7/perks v1.0.1 // indirect - github.com/cespare/xxhash/v2 v2.2.0 // indirect - github.com/prometheus/client_model v0.5.0 // indirect - github.com/prometheus/common v0.48.0 // indirect - github.com/prometheus/procfs v0.12.0 // indirect - golang.org/x/sys v0.17.0 // indirect - google.golang.org/protobuf v1.33.0 // indirect -) diff --git a/harnesses/pm-freshness-bench/go.sum b/harnesses/pm-freshness-bench/go.sum deleted file mode 100644 index 0874f900..00000000 --- a/harnesses/pm-freshness-bench/go.sum +++ /dev/null @@ -1,22 +0,0 @@ -github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= -github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= -github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= -github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= -github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= -github.com/prometheus/client_golang v1.19.1 h1:wZWJDwK+NameRJuPGDhlnFgx8e8HN3XHQeLaYJFJBOE= -github.com/prometheus/client_golang v1.19.1/go.mod h1:mP78NwGzrVks5S2H6ab8+ZZGJLZUq1hoULYBAYBw1Ho= -github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw= -github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI= -github.com/prometheus/common v0.48.0 h1:QO8U2CdOzSn1BBsmXJXduaaW+dY/5QLjfB8svtSzKKE= -github.com/prometheus/common v0.48.0/go.mod h1:0/KsvlIEfPQCQ5I2iNSAWKPZziNCvRs5EC6ILDTlAPc= -github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo= -github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo= -golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y= -golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI= -google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= -nhooyr.io/websocket v1.8.11 h1:f/qXNc2/3DpoSZkHt1DQu6rj4zGC8JmkkLkWss0MgN0= -nhooyr.io/websocket v1.8.11/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/harnesses/pm-rate-limits/cmd/script/main.go b/harnesses/pm-rate-limits/cmd/script/main.go index bd9eb9ec..e046a7b0 100644 --- a/harnesses/pm-rate-limits/cmd/script/main.go +++ b/harnesses/pm-rate-limits/cmd/script/main.go @@ -51,6 +51,9 @@ func main() { if v.Slug == "kalshi" && cfg.KalshiAPIKeyID != "" && cfg.KalshiPrivateKeyPEM != "" { go runKalshiWS(ctx, v.state, cfg.KalshiAPIKeyID, cfg.KalshiPrivateKeyPEM) } + if v.Slug == "kalshi" && cfg.KalshiAPIKeyID != "" && cfg.KalshiPrivateKeyPEM != "" { + go runKalshiTradeRestPoll(ctx, v.state) + } } // Aggregator probes. One goroutine per (venue, source) pair; reuses diff --git a/harnesses/pm-rate-limits/cmd/script/metrics.go b/harnesses/pm-rate-limits/cmd/script/metrics.go index 9545699e..139dfcfa 100644 --- a/harnesses/pm-rate-limits/cmd/script/metrics.go +++ b/harnesses/pm-rate-limits/cmd/script/metrics.go @@ -67,6 +67,12 @@ var ( Help: "WebSocket connections dropped after being established.", }, []string{"venue", "region", "source"}) + wsTradePublishLag = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "pmapi_ws_trade_pub_lag_seconds", + Help: "Delay between venue server-side trade creation and WS subscriber receipt. Polymarket: CLOB WS timestamp field vs recv time. Kalshi: REST create_date vs WS ticker recv time (us-east only).", + Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}, + }, []string{"venue", "region", "source"}) + rampDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ Name: "pmapi_ramp_request_duration_seconds", Help: "Round-trip during the daily rate-limit ramp, by tier (requests per 10 s window). Successful requests only.", diff --git a/harnesses/pm-rate-limits/cmd/script/trade_pub_lag.go b/harnesses/pm-rate-limits/cmd/script/trade_pub_lag.go new file mode 100644 index 00000000..b5188f65 --- /dev/null +++ b/harnesses/pm-rate-limits/cmd/script/trade_pub_lag.go @@ -0,0 +1,126 @@ +package main + +// Kalshi correlation state + REST poller for trade pub lag. +// Polymarket pub lag is computed inline in wsSession (timestamp in WS frame). + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "sync" + "time" +) + +type wsTickerEvent struct { + priceCents int + recvAt time.Time +} + +var kalshiWSTickerState = struct { + mu sync.Mutex + byMkt map[string][]wsTickerEvent +}{byMkt: map[string][]wsTickerEvent{}} + +func recordKalshiWSTick(mkt string, priceCents int, recvAt time.Time) { + kalshiWSTickerState.mu.Lock() + defer kalshiWSTickerState.mu.Unlock() + cutoff := recvAt.Add(-30 * time.Second) + existing := kalshiWSTickerState.byMkt[mkt] + pruned := existing[:0] + for _, e := range existing { + if e.recvAt.After(cutoff) { + pruned = append(pruned, e) + } + } + kalshiWSTickerState.byMkt[mkt] = append(pruned, wsTickerEvent{priceCents: priceCents, recvAt: recvAt}) +} + +func findKalshiWSMatch(mkt string, priceCents int, createDate time.Time) (time.Time, bool) { + kalshiWSTickerState.mu.Lock() + defer kalshiWSTickerState.mu.Unlock() + lo := createDate.Add(-2 * time.Second) + hi := createDate.Add(15 * time.Second) + for _, e := range kalshiWSTickerState.byMkt[mkt] { + if e.priceCents == priceCents && e.recvAt.After(lo) && e.recvAt.Before(hi) { + return e.recvAt, true + } + } + return time.Time{}, false +} + +func runKalshiTradeRestPoll(ctx context.Context, st *venueState) { + if currentRegion != "us-east" { + return + } + client := &http.Client{Timeout: 8 * time.Second} + seen := map[string]struct{}{} + var seenOrder []string + + for ctx.Err() == nil { + select { + case <-ctx.Done(): + return + case <-time.After(3 * time.Second): + } + + pin := st.getPin() + if pin.Market == "" { + continue + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://api.elections.kalshi.com/v1/social/trades", nil) + if err != nil { + continue + } + req.Header.Set("User-Agent", userAgent) + + resp, err := client.Do(req) + if err != nil { + continue + } + + var body struct { + Trades []struct { + TradeID string `json:"trade_id"` + Ticker string `json:"ticker"` + Price int `json:"price"` + CreateDate string `json:"create_date"` + } `json:"trades"` + Cursor string `json:"cursor"` + } + if json.NewDecoder(resp.Body).Decode(&body) != nil { + resp.Body.Close() + continue + } + resp.Body.Close() + + for _, t := range body.Trades { + if _, dup := seen[t.TradeID]; dup { + continue + } + seen[t.TradeID] = struct{}{} + seenOrder = append(seenOrder, t.TradeID) + if len(seenOrder) > 4096 { + delete(seen, seenOrder[0]) + seenOrder = seenOrder[1:] + } + + if !strings.EqualFold(t.Ticker, pin.Market) { + continue + } + createDate, err := time.Parse(time.RFC3339Nano, t.CreateDate) + if err != nil { + continue + } + recvAt, ok := findKalshiWSMatch(t.Ticker, t.Price, createDate) + if !ok { + continue + } + delta := recvAt.Sub(createDate).Seconds() + if delta >= 0 && delta <= 30 { + wsTradePublishLag.WithLabelValues("kalshi", currentRegion, sourceDirect).Observe(delta) + } + } + } +} diff --git a/harnesses/pm-rate-limits/cmd/script/ws.go b/harnesses/pm-rate-limits/cmd/script/ws.go index 6ed3746d..2dca8864 100644 --- a/harnesses/pm-rate-limits/cmd/script/ws.go +++ b/harnesses/pm-rate-limits/cmd/script/ws.go @@ -113,6 +113,32 @@ func wsSession(ctx context.Context, st *venueState, token string) bool { wsInterarrival.WithLabelValues("polymarket", currentRegion, sourceDirect).Observe(now.Sub(lastMsg).Seconds()) } lastMsg = now + var frames []json.RawMessage + if data[0] == '[' { + _ = json.Unmarshal(data, &frames) + } else { + frames = []json.RawMessage{data} + } + for _, frame := range frames { + var ev struct { + EventType string `json:"event_type"` + Timestamp string `json:"timestamp"` + } + if err := json.Unmarshal(frame, &ev); err != nil { + continue + } + if ev.EventType != "last_trade_price" || ev.Timestamp == "" { + continue + } + tsMs, err := strconv.ParseFloat(ev.Timestamp, 64) + if err != nil { + continue + } + delta := now.Sub(time.UnixMilli(int64(tsMs))).Seconds() + if delta >= 0 && delta <= 30 { + wsTradePublishLag.WithLabelValues("polymarket", currentRegion, sourceDirect).Observe(delta) + } + } } } @@ -261,5 +287,21 @@ func kalshiWSSession(ctx context.Context, st *venueState, ticker, keyID, private wsInterarrival.WithLabelValues("kalshi", currentRegion, sourceDirect).Observe(now.Sub(lastMsg).Seconds()) } lastMsg = now + var tick struct { + Msg struct { + LastPrice int `json:"last_price"` + YesBid int `json:"yes_bid"` + YesAsk int `json:"yes_ask"` + } `json:"msg"` + } + if err := json.Unmarshal(data, &tick); err == nil { + price := tick.Msg.LastPrice + if price <= 0 && tick.Msg.YesBid > 0 && tick.Msg.YesAsk > 0 { + price = (tick.Msg.YesBid + tick.Msg.YesAsk) / 2 + } + if price > 0 { + recordKalshiWSTick(ticker, price, now) + } + } } }