From f9d1a0f467603e5fdb8ff32348ef717d02443704 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Thu, 23 Jul 2026 20:14:27 +0200 Subject: [PATCH] token-trade-coverage: quota-respecting sub-sampling (Bitquery 6x sparser, page caps, api_calls_total counter for observability) --- harnesses/token-trade-coverage/.env.example | 39 +++++++++++-- .../cmd/scanner/bitquery.go | 20 ++++--- .../token-trade-coverage/cmd/scanner/codex.go | 13 +++-- .../cmd/scanner/config.go | 43 ++++++++++++++- .../token-trade-coverage/cmd/scanner/main.go | 55 ++++++++++++++++--- .../cmd/scanner/metrics.go | 12 ++++ .../cmd/scanner/mobula.go | 13 +++-- 7 files changed, 161 insertions(+), 34 deletions(-) diff --git a/harnesses/token-trade-coverage/.env.example b/harnesses/token-trade-coverage/.env.example index 26697984..a613cf9a 100644 --- a/harnesses/token-trade-coverage/.env.example +++ b/harnesses/token-trade-coverage/.env.example @@ -1,14 +1,41 @@ # token-trade-coverage harness. Fill in and rename to `.env`. -# Providers (all required for a full bench cycle; the harness -# tolerates missing keys per-provider and reports probe_ok=0 for that -# provider but keeps the others going, so partial config is fine for -# local debugging). +# Provider keys (harness tolerates missing keys per-provider; missing +# key sets probe_ok=0 for that provider only, others keep going). MOBULA_API_KEY= BITQUERY_API_KEY= CODEX_API_KEY= -# Cadence + runtime knobs. -SWEEP_SEC=1800 +# ─── Cadence ────────────────────────────────────────────────────────── +# Base sweep cadence. 3600 = 1 hour. Every SWEEP_SEC the harness iterates +# every (provider, chain, token) tuple, respecting the per-provider +# EveryN sub-sampling below. +SWEEP_SEC=3600 + +# ─── Per-provider sub-sampling (respects free-tier quotas) ──────────── +# A provider runs only when `iteration % EveryN == 0`. Iteration is the +# 0-indexed sweep counter (boot = 0, first tick = 1, ...). +# +# Default free-tier budget math (at SWEEP_SEC=3600, 8 EVM+Solana tokens, +# 2 Stellar tokens Mobula-only): +# Mobula MOBULA_EVERY_N=1 → 240 calls / day = free (own API) +# Bitquery BITQUERY_EVERY_N=6 → 8 tokens × 4 sweeps/day × 30 = 960 +# calls / month at ~1 pt each = ~960 pts (Free plan cap 1000) +# Codex CODEX_EVERY_N=1 → 192 calls / day = fits Codex rate limits +# +# On a paid Bitquery plan (Developer $99/mo = 500k pts), set BITQUERY_EVERY_N=1. +MOBULA_EVERY_N=1 +BITQUERY_EVERY_N=6 +CODEX_EVERY_N=1 + +# ─── Query caps (prevent runaway pagination on heavy tokens) ────────── +# Mobula: cursor pages, 5000 rows each → MAX_PAGES=20 caps at 100k trades / token / sweep. +# Bitquery: single-shot query, no cursor → MAX_ROWS is the `limit` in the GraphQL. +# Codex: cursor pages, 200 rows each → MAX_PAGES=10 caps at 2000 events / token / sweep. +MOBULA_MAX_PAGES=20 +BITQUERY_MAX_ROWS=10000 +CODEX_MAX_PAGES=10 + +# ─── Runtime knobs ──────────────────────────────────────────────────── METRICS_PORT=2112 HTTP_TIMEOUT_SEC=30 diff --git a/harnesses/token-trade-coverage/cmd/scanner/bitquery.go b/harnesses/token-trade-coverage/cmd/scanner/bitquery.go index babbbca1..2d2005cf 100644 --- a/harnesses/token-trade-coverage/cmd/scanner/bitquery.go +++ b/harnesses/token-trade-coverage/cmd/scanner/bitquery.go @@ -60,23 +60,27 @@ func fetchBitquery( apiKey string, tok Token, windowStart, windowEnd int64, + maxRows int, ) (int, int, error) { if apiKey == "" { return 0, 0, fmt.Errorf("BITQUERY_API_KEY not set") } + if maxRows <= 0 { + maxRows = 10000 + } sinceISO := time.Unix(windowStart/1000, 0).UTC().Format(time.RFC3339) tillISO := time.Unix(windowEnd/1000, 0).UTC().Format(time.RFC3339) var query string if tok.Chain == "solana" { - query = solanaTradesGQL(tok.Address, sinceISO, tillISO) + query = solanaTradesGQL(tok.Address, sinceISO, tillISO, maxRows) } else { network := bitqueryEVMNetwork(tok.Chain) if network == "" { return 0, 0, fmt.Errorf("bitquery: unsupported chain %s", tok.Chain) } - query = evmTradesGQL(network, tok.Address, sinceISO, tillISO) + query = evmTradesGQL(network, tok.Address, sinceISO, tillISO, maxRows) } body, _ := json.Marshal(map[string]string{"query": query}) @@ -152,11 +156,11 @@ func bitqueryEVMNetwork(chain string) string { return "" } -func solanaTradesGQL(mintAddress, since, till string) string { +func solanaTradesGQL(mintAddress, since, till string, maxRows int) string { return fmt.Sprintf(`{ Solana(dataset: realtime) { DEXTradeByTokens( - limit: {count: 10000} + limit: {count: %d} where: { Block: {Time: {since: "%s", till: "%s"}} Trade: {Currency: {MintAddress: {is: "%s"}}} @@ -166,14 +170,14 @@ func solanaTradesGQL(mintAddress, since, till string) string { Trade { Dex { ProtocolName } } } } -}`, since, till, mintAddress) +}`, maxRows, since, till, mintAddress) } -func evmTradesGQL(network, tokenAddress, since, till string) string { +func evmTradesGQL(network, tokenAddress, since, till string, maxRows int) string { return fmt.Sprintf(`{ EVM(dataset: realtime, network: %s) { DEXTrades( - limit: {count: 10000} + limit: {count: %d} where: { Block: {Time: {since: "%s", till: "%s"}} Trade: {Buy: {Currency: {SmartContract: {is: "%s"}}}} @@ -183,5 +187,5 @@ func evmTradesGQL(network, tokenAddress, since, till string) string { Trade { Dex { ProtocolName } } } } -}`, network, since, till, tokenAddress) +}`, network, maxRows, since, till, tokenAddress) } diff --git a/harnesses/token-trade-coverage/cmd/scanner/codex.go b/harnesses/token-trade-coverage/cmd/scanner/codex.go index 85cbce04..9b19b224 100644 --- a/harnesses/token-trade-coverage/cmd/scanner/codex.go +++ b/harnesses/token-trade-coverage/cmd/scanner/codex.go @@ -62,10 +62,14 @@ func fetchCodex( apiKey string, tok Token, windowStart, windowEnd int64, + maxPages int, ) (int, int, error) { if apiKey == "" { return 0, 0, fmt.Errorf("CODEX_API_KEY not set") } + if maxPages <= 0 { + maxPages = 10 + } networkID := codexNetworkID(tok.Chain) if networkID == 0 { return 0, 0, fmt.Errorf("codex: unsupported chain %s", tok.Chain) @@ -75,11 +79,10 @@ func fetchCodex( toSec := windowEnd / 1000 var ( - total int - cursor string - exchSet = map[string]struct{}{} - hashSet = map[string]struct{}{} - maxPages = 50 + total int + cursor string + exchSet = map[string]struct{}{} + hashSet = map[string]struct{}{} ) for page := 0; page < maxPages; page++ { diff --git a/harnesses/token-trade-coverage/cmd/scanner/config.go b/harnesses/token-trade-coverage/cmd/scanner/config.go index 71f91d3f..68717aca 100644 --- a/harnesses/token-trade-coverage/cmd/scanner/config.go +++ b/harnesses/token-trade-coverage/cmd/scanner/config.go @@ -30,6 +30,23 @@ type Config struct { Tokens []Token Capabilities map[string]ProviderCapability // provider -> {chain -> supported} MeasurementWinMs int64 // rolling window per measurement + + // Per-provider sub-sampling (respects free-tier quotas). + // A provider only runs when `iteration % everyN == 0`. Default 1 + // means every sweep. Higher = sparser samples but preserves free- + // tier point budget on providers with tight limits (Bitquery Free + // gives 1000 points / month, so on the default 60-min cadence with + // EveryN=6 we spend ≤ 960 points / month across 8 supported tokens). + MobulaEveryN int + BitqueryEveryN int + CodexEveryN int + + // Page caps: bound the max pagination pages per provider per token + // per sweep. Prevents runaway cursor loops on a single very-heavy + // token from burning a whole sweep's quota. + MobulaMaxPages int + BitqueryMaxRows int // Bitquery has no cursor — this is the per-query row cap + CodexMaxPages int } // LoadConfig reads env, defaults + hardcoded reference token list. @@ -38,13 +55,37 @@ type Config struct { // goes illiquid. func LoadConfig() *Config { return &Config{ - SweepSec: envInt("SWEEP_SEC", 1800), + // Base cadence: 60 min. Doubled from the initial 30-min value + // after quota math on Bitquery's 1000-pt free plan (see + // BitqueryEveryN below). 24 sweeps / day gives >= 12 samples per + // (chain, token) per day for providers that run every sweep — + // plenty for a stable p50 over 24h. + SweepSec: envInt("SWEEP_SEC", 3600), MetricsPort: envStr("METRICS_PORT", "2112"), HTTPTimeoutSec: envInt("HTTP_TIMEOUT_SEC", 30), MeasurementWinMs: int64(60*60) * 1000, // 60 min rolling window MobulaKey: os.Getenv("MOBULA_API_KEY"), BitqueryKey: os.Getenv("BITQUERY_API_KEY"), CodexKey: os.Getenv("CODEX_API_KEY"), + + // Mobula: unlimited (own infra). Run every sweep. + MobulaEveryN: envInt("MOBULA_EVERY_N", 1), + // Bitquery Free: 1000 pts / month. At ~1-2 pts per + // DEXTradeByTokens call and 8 supported tokens per sweep, + // running every 6th sweep = 4 sweeps / day × 8 tokens × 30 + // days = 960 calls / month, well inside the free budget. + // Bump to 1 if on the Developer plan ($99/mo, 500k pts). + BitqueryEveryN: envInt("BITQUERY_EVERY_N", 6), + // Codex Free: generous rate-based limits (no monthly point + // cap on the current tier). Run every sweep. + CodexEveryN: envInt("CODEX_EVERY_N", 1), + + // Page caps. A high-volume token that fires cursor-paginated + // requests indefinitely can drain a monthly budget in one + // sweep. These bound the worst case per token per sweep. + MobulaMaxPages: envInt("MOBULA_MAX_PAGES", 20), + BitqueryMaxRows: envInt("BITQUERY_MAX_ROWS", 10000), + CodexMaxPages: envInt("CODEX_MAX_PAGES", 10), Tokens: []Token{ // Reference tokens selected 2026-07-23 via a Mobula // `/api/2/trades/filters` sweep over the last hour, picking diff --git a/harnesses/token-trade-coverage/cmd/scanner/main.go b/harnesses/token-trade-coverage/cmd/scanner/main.go index eb454ccc..017d4560 100644 --- a/harnesses/token-trade-coverage/cmd/scanner/main.go +++ b/harnesses/token-trade-coverage/cmd/scanner/main.go @@ -51,37 +51,72 @@ func main() { // First sweep runs immediately at boot so Prom sees data within // SWEEP_SEC + measurement time rather than waiting a full cycle. - sweep(ctx, cfg, client) + iteration := 0 + sweep(ctx, cfg, client, iteration) for { select { case <-ctx.Done(): log.Printf("[shutdown] signal received") return case <-ticker.C: - sweep(ctx, cfg, client) + iteration++ + sweep(ctx, cfg, client, iteration) } } } +// providerEnabled reports whether the provider should run on this +// iteration given its sub-sampling cadence (see Config.*EveryN). +func providerEnabled(cfg *Config, provider string, iteration int) bool { + var everyN int + switch provider { + case "mobula": + everyN = cfg.MobulaEveryN + case "bitquery": + everyN = cfg.BitqueryEveryN + case "codex": + everyN = cfg.CodexEveryN + default: + return false + } + if everyN <= 1 { + return true + } + return iteration%everyN == 0 +} + // sweep runs one full measurement cycle. For each token, we fan out // across every supported provider in parallel, wait for all to finish // (or fail), compute the union baseline as max(counts) and emit. -func sweep(ctx context.Context, cfg *Config, client *http.Client) { +// +// `iteration` is the 0-indexed sweep counter — used for per-provider +// sub-sampling (see providerEnabled). Skipped providers keep their +// previous capture_pct value in Prom, which `avg_over_time([24h])` +// then averages across the sparse sample. +func sweep(ctx context.Context, cfg *Config, client *http.Client, iteration int) { sweepStart := time.Now() windowEnd := sweepStart.UnixMilli() windowStart := windowEnd - cfg.MeasurementWinMs + activeProviders := make([]string, 0, len(providers)) + for _, p := range providers { + if providerEnabled(cfg, p, iteration) { + activeProviders = append(activeProviders, p) + } + } + log.Printf("[sweep] iter=%d active=%v", iteration, activeProviders) + for _, tok := range cfg.Tokens { if tok.Address == "" { // Placeholder row (Stellar addresses TBD). Skip silently // so the container doesn't spam warnings until they land. continue } - results := make([]Result, 0, len(providers)) + results := make([]Result, 0, len(activeProviders)) var mu sync.Mutex var wg sync.WaitGroup - for _, p := range providers { + for _, p := range activeProviders { if !cfg.Supports(p, tok.Chain) { continue } @@ -127,7 +162,9 @@ func sweep(ctx context.Context, cfg *Config, client *http.Client) { log.Printf("[sweep] done in %.1fs", time.Since(sweepStart).Seconds()) } -// fetchOne dispatches to the provider-specific fetcher. +// fetchOne dispatches to the provider-specific fetcher, threading each +// provider's page/row cap so a runaway pagination loop can never drain +// a monthly quota in a single sweep. func fetchOne( ctx context.Context, client *http.Client, @@ -138,11 +175,11 @@ func fetchOne( ) (int, int, error) { switch provider { case "mobula": - return fetchMobula(ctx, client, cfg.MobulaKey, tok, windowStart, windowEnd) + return fetchMobula(ctx, client, cfg.MobulaKey, tok, windowStart, windowEnd, cfg.MobulaMaxPages) case "bitquery": - return fetchBitquery(ctx, client, cfg.BitqueryKey, tok, windowStart, windowEnd) + return fetchBitquery(ctx, client, cfg.BitqueryKey, tok, windowStart, windowEnd, cfg.BitqueryMaxRows) case "codex": - return fetchCodex(ctx, client, cfg.CodexKey, tok, windowStart, windowEnd) + return fetchCodex(ctx, client, cfg.CodexKey, tok, windowStart, windowEnd, cfg.CodexMaxPages) } return 0, 0, fmt.Errorf("unknown provider %s", provider) } diff --git a/harnesses/token-trade-coverage/cmd/scanner/metrics.go b/harnesses/token-trade-coverage/cmd/scanner/metrics.go index b0cf151a..3481c072 100644 --- a/harnesses/token-trade-coverage/cmd/scanner/metrics.go +++ b/harnesses/token-trade-coverage/cmd/scanner/metrics.go @@ -33,6 +33,17 @@ var ( Name: "ocb_token_trade_probe_ok", Help: "1 on successful fetch, 0 on error or timeout. Consumed by the bench spec's `success` query.", }, []string{"provider", "chain", "token"}) + + // Cumulative API-call counter for quota observability. Counted at + // the fetchOne granularity — one increment per (provider, token) + // tuple in a sweep, regardless of how many paginated sub-requests + // fired underneath — because that's what maps 1:1 to the provider's + // monthly point/credit budget as billed. `increase(...[30d])` per + // provider gives the running monthly consumption. + apiCalls = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "ocb_token_trade_api_calls_total", + Help: "Total fetchOne invocations per provider since worker start. Query with increase()[30d] for monthly consumption vs free-tier budget.", + }, []string{"provider"}) ) // Result is the per-call outcome of one provider fetch. @@ -69,5 +80,6 @@ func emitCycle(results []Result, unionMax int) { } else { capturePct.With(lbl).Set(0) } + apiCalls.WithLabelValues(r.Provider).Inc() } } diff --git a/harnesses/token-trade-coverage/cmd/scanner/mobula.go b/harnesses/token-trade-coverage/cmd/scanner/mobula.go index 2dc140c4..ff32cd68 100644 --- a/harnesses/token-trade-coverage/cmd/scanner/mobula.go +++ b/harnesses/token-trade-coverage/cmd/scanner/mobula.go @@ -46,16 +46,19 @@ func fetchMobula( apiKey string, tok Token, windowStart, windowEnd int64, + maxPages int, ) (int, int, error) { if apiKey == "" { return 0, 0, fmt.Errorf("MOBULA_API_KEY not set") } + if maxPages <= 0 { + maxPages = 20 + } var ( - total int - cursor string - dexSet = map[string]struct{}{} - hashSet = map[string]struct{}{} - maxPages = 50 + total int + cursor string + dexSet = map[string]struct{}{} + hashSet = map[string]struct{}{} ) for page := 0; page < maxPages; page++ { u, _ := url.Parse(mobulaBase)