token-trade-coverage: quota-respecting sub-sampling

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

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)
 }

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++ {

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

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)
 }

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()
 	}
 }

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)