diff --git a/README.md b/README.md index bc7d231..aee69d3 100644 --- a/README.md +++ b/README.md @@ -131,9 +131,11 @@ Logs: `polydisplay.log` in the working directory. Rolled at local midnight to Candles and prices: Kraken, with Coinbase fallback. Kraken ticker prices are fetched in one request; OHLC requests are paced to its public API guidance. -Positions and activity: Polymarket data-api, polled every 30s. Account P/L: -Polymarket user-pnl-api, 720 hourly points over 30 days, polled every 2 min -and thinned to 120 points for the sparkline. Gamma market metadata supplies +Positions and activity: Polymarket data-api, independently polled every 30s +and 1 min without request bursts. Account P/L: Polymarket user-pnl-api, 720 +hourly points over 30 days, polled every 30 min and thinned to 120 points for +the sparkline. Each endpoint backs off independently on failures and honours +`Retry-After`. Gamma market metadata supplies exact end times, time to resolution in the device's timezone, and the price to beat for Up/Down positions. diff --git a/server.go b/server.go index bad7fa6..ac70740 100644 --- a/server.go +++ b/server.go @@ -20,6 +20,7 @@ import ( "fmt" "io" "log" + "math/rand/v2" "net/http" "net/url" "os" @@ -291,23 +292,34 @@ var ( /* --------------------- Polymarket pacing --------------------- */ // -// data-api sends `cache-control: max-age=15`, so polling faster than that only -// ever misses Cloudflare's cache and hits the origin rate limiter. We poll at -// 30s, and on HTTP 429 back off exponentially (honouring Retry-After) while -// continuing to serve the last positions we got. +// The APIs are rate limited per public IP by Cloudflare. Keep each endpoint on +// its own schedule, space calls to the shared Data API host, and add jitter so +// multiple clients behind the same egress do not synchronize their requests. // data-api host; a var so tests can point it at a stub var polyBase = "https://data-api.polymarket.com" const ( - polyInterval = 30 * time.Second - polyBackoffMin = 60 * time.Second - polyBackoffMax = 10 * time.Minute + positionsInterval = 30 * time.Second + activityInterval = time.Minute + pnlInterval = 30 * time.Minute + dataAPIMinGap = 5 * time.Second + activityStartWait = 10 * time.Second + pnlStartWait = 20 * time.Second + polyBackoffMin = time.Minute + polyBackoffMax = 30 * time.Minute ) +type pollSchedule struct { + nextAt time.Time + backoff time.Duration +} + var ( - polyNextAt time.Time // don't call data-api before this - polyBackoff time.Duration + positionsPoll pollSchedule + activityPoll pollSchedule + pnlPoll pollSchedule + dataAPILastAt time.Time polyWallet string // wallet the cached positions/activity belong to lastPositions []Position lastActivity []Act @@ -317,39 +329,43 @@ var ( // only moves as fast as prices do, so it gets a slower cadence of its own. var pnlBase = "https://user-pnl-api.polymarket.com" -const ( - pnlInterval = 2 * time.Minute - pnlRateLimited = 10 * time.Minute - pnlSeriesMax = 120 // points kept for the sparkline -) +const pnlSeriesMax = 120 // points kept for the sparkline -var ( - pnlNextAt time.Time - pnlSeries [][2]float64 -) +var pnlSeries [][2]float64 + +func jitter(d time.Duration) time.Duration { + if d <= 0 { + return d + } + // Positive-only jitter preserves minimum delays and Retry-After semantics. + return d + time.Duration(rand.Int64N(max(1, int64(d/10)))) +} + +func (p *pollSchedule) success(now time.Time, interval time.Duration) { + p.backoff = 0 + p.nextAt = now.Add(jitter(interval)) +} -// schedule the next data-api call after a rate-limit rejection -func polyRateLimited(retryAfter time.Duration) { - if polyBackoff == 0 { - polyBackoff = polyBackoffMin - } else if polyBackoff < polyBackoffMax { - polyBackoff *= 2 +func (p *pollSchedule) failed(now time.Time, retryAfter time.Duration) time.Duration { + if p.backoff == 0 { + p.backoff = polyBackoffMin + } else if p.backoff < polyBackoffMax { + p.backoff *= 2 } - if polyBackoff > polyBackoffMax { - polyBackoff = polyBackoffMax + if p.backoff > polyBackoffMax { + p.backoff = polyBackoffMax } - wait := polyBackoff + wait := jitter(p.backoff) if retryAfter > wait { wait = retryAfter } - polyNextAt = time.Now().Add(wait) - log.Printf("polymarket: rate limited, backing off %s", wait.Round(time.Second)) + p.nextAt = now.Add(wait) + return wait } -// banner text while we're sitting out a rate limit -func polyBackoffNote() string { +func polyBackoffNote(p pollSchedule) string { return fmt.Sprintf("polymarket: rate limited, retrying in %s", - time.Until(polyNextAt).Round(time.Second)) + time.Until(p.nextAt).Round(time.Second)) } /* ------------------------- HTTP helpers ------------------------- */ @@ -880,24 +896,22 @@ func buildPnl(series [][2]float64) *PnL { } // refresh the cached P/L series when it's due; failures keep the last one -func refreshPnl(wallet string) { - if wallet == "" || time.Now().Before(pnlNextAt) { +func refreshPnl(wallet string, now time.Time) { + if wallet == "" || now.Before(pnlPoll.nextAt) { return } s, err := fetchPnlSeries(wallet) if err == nil { pnlSeries = s - pnlNextAt = time.Now().Add(pnlInterval) + pnlPoll.success(now, pnlInterval) return } - wait := pnlInterval + var retryAfter time.Duration if he, ok := err.(*httpError); ok && he.Status == 429 { - wait = pnlRateLimited - if he.RetryAfter > wait { - wait = he.RetryAfter - } + retryAfter = he.RetryAfter } - pnlNextAt = time.Now().Add(wait) + wait := pnlPoll.failed(now, retryAfter) + log.Printf("polymarket pnl: request failed, backing off %s", wait.Round(time.Second)) } /* ------------------------- refresh loops ------------------------- */ @@ -907,45 +921,70 @@ func refreshFast() { mu.RLock() c := cfg mu.RUnlock() + now := time.Now() note := "" positions, activity := lastPositions, lastActivity wallet := strings.TrimSpace(c.Wallet) if wallet != polyWallet { // wallet changed -> refetch now, drop stale data - polyWallet, polyNextAt, polyBackoff = wallet, time.Time{}, 0 + polyWallet = wallet + positionsPoll = pollSchedule{} + activityPoll = pollSchedule{nextAt: now.Add(activityStartWait)} + pnlPoll = pollSchedule{nextAt: now.Add(pnlStartWait)} + dataAPILastAt = time.Time{} positions, activity = nil, nil - pnlSeries, pnlNextAt = nil, time.Time{} + pnlSeries = nil } if wallet == "" { positions, activity = nil, nil - } else if time.Now().Before(polyNextAt) { - if polyBackoff > 0 { // rate limited: say so rather than silently showing stale data - note = polyBackoffNote() + } else if now.Before(positionsPoll.nextAt) { + if positionsPoll.backoff > 0 { // say why positions are stale + note = polyBackoffNote(positionsPoll) } } else { p, err := fetchPositions(wallet) + finishedAt := time.Now() + dataAPILastAt = finishedAt he, isHTTP := err.(*httpError) switch { case err == nil: - positions, polyBackoff = p, 0 - polyNextAt = time.Now().Add(polyInterval) - if a, aerr := fetchActivity(wallet); aerr == nil { - activity = a - } + positions = p + positionsPoll.success(finishedAt, positionsInterval) case isHTTP && he.Status == 429: - polyRateLimited(he.RetryAfter) // keep serving the last positions we got - note = polyBackoffNote() + wait := positionsPoll.failed(finishedAt, he.RetryAfter) + log.Printf("polymarket positions: rate limited, backing off %s", wait.Round(time.Second)) + note = polyBackoffNote(positionsPoll) default: note = "polymarket: " + err.Error() - polyNextAt = time.Now().Add(polyInterval) + wait := positionsPoll.failed(finishedAt, 0) + log.Printf("polymarket positions: request failed, backing off %s", wait.Round(time.Second)) + } + } + // Never burst activity immediately after positions on the shared Data API. + if wallet != "" && !now.Before(activityPoll.nextAt) && + (dataAPILastAt.IsZero() || now.Sub(dataAPILastAt) >= dataAPIMinGap) { + a, err := fetchActivity(wallet) + finishedAt := time.Now() + dataAPILastAt = finishedAt + he, isHTTP := err.(*httpError) + switch { + case err == nil: + activity = a + activityPoll.success(finishedAt, activityInterval) + case isHTTP && he.Status == 429: + wait := activityPoll.failed(finishedAt, he.RetryAfter) + log.Printf("polymarket activity: rate limited, backing off %s", wait.Round(time.Second)) + default: + wait := activityPoll.failed(finishedAt, 0) + log.Printf("polymarket activity: request failed, backing off %s", wait.Round(time.Second)) } } lastPositions, lastActivity = positions, activity if wallet == "" { pnlSeries = nil } else { - refreshPnl(wallet) + refreshPnl(wallet, time.Now()) } // Kraken returns all requested tickers in one call. Only missing pairs fall diff --git a/server_test.go b/server_test.go index 1d741d5..a74d320 100644 --- a/server_test.go +++ b/server_test.go @@ -43,9 +43,9 @@ func TestParseRetryAfter(t *testing.T) { } } -func TestPolyRateLimitedBackoff(t *testing.T) { - polyNextAt, polyBackoff = time.Time{}, 0 - t.Cleanup(func() { polyNextAt, polyBackoff = time.Time{}, 0 }) +func TestPollScheduleBackoff(t *testing.T) { + var poll pollSchedule + now := time.Now() // doubles from the floor, then clamps at the ceiling want := []time.Duration{ @@ -53,34 +53,37 @@ func TestPolyRateLimitedBackoff(t *testing.T) { 2 * polyBackoffMin, 4 * polyBackoffMin, 8 * polyBackoffMin, + 16 * polyBackoffMin, polyBackoffMax, polyBackoffMax, } for i, w := range want { - polyRateLimited(0) - if polyBackoff != w { - t.Fatalf("after %d rate limits: backoff = %v, want %v", i+1, polyBackoff, w) + wait := poll.failed(now, 0) + if poll.backoff != w { + t.Fatalf("after %d failures: backoff = %v, want %v", i+1, poll.backoff, w) } - if d := time.Until(polyNextAt); d > w || d < w-time.Second { - t.Fatalf("after %d rate limits: next call in %v, want ~%v", i+1, d, w) + if wait < w || wait >= w+w/10 { + t.Fatalf("after %d failures: wait = %v, want [%v, %v)", i+1, wait, w, w+w/10) + } + if poll.nextAt != now.Add(wait) { + t.Fatalf("after %d failures: nextAt = %v, want %v", i+1, poll.nextAt, now.Add(wait)) } } } -func TestPolyRateLimitedHonoursRetryAfter(t *testing.T) { - polyNextAt, polyBackoff = time.Time{}, 0 - t.Cleanup(func() { polyNextAt, polyBackoff = time.Time{}, 0 }) +func TestPollScheduleHonoursRetryAfter(t *testing.T) { + var poll pollSchedule + now := time.Now() // Retry-After longer than our own backoff wins... - polyRateLimited(polyBackoffMin + time.Minute) - if d := time.Until(polyNextAt); d < polyBackoffMin+50*time.Second { - t.Errorf("next call in %v, want the longer Retry-After", d) + want := polyBackoffMin + time.Minute + if wait := poll.failed(now, want); wait != want { + t.Errorf("wait = %v, want Retry-After %v", wait, want) } // ...and a shorter one does not shorten the backoff. - polyNextAt, polyBackoff = time.Time{}, 0 - polyRateLimited(time.Second) - if d := time.Until(polyNextAt); d < polyBackoffMin-time.Second { - t.Errorf("next call in %v, want at least the %v floor", d, polyBackoffMin) + poll = pollSchedule{} + if wait := poll.failed(now, time.Second); wait < polyBackoffMin { + t.Errorf("wait = %v, want at least %v", wait, polyBackoffMin) } } @@ -107,12 +110,14 @@ func TestRefreshFastBacksOffAndKeepsLastPositions(t *testing.T) { origBase := polyBase polyBase = srv.URL cfg = Config{Wallet: "0xtest", CandleDays: 1, Sort: "az"} // no coins -> no market-data calls - polyNextAt, polyBackoff, polyWallet = time.Time{}, 0, "" + positionsPoll, activityPoll, pnlPoll = pollSchedule{}, pollSchedule{}, pollSchedule{} + dataAPILastAt, polyWallet = time.Time{}, "" lastPositions, lastActivity = nil, nil t.Cleanup(func() { polyBase = origBase cfg, state = Config{}, State{} - polyNextAt, polyBackoff, polyWallet = time.Time{}, 0, "" + positionsPoll, activityPoll, pnlPoll = pollSchedule{}, pollSchedule{}, pollSchedule{} + dataAPILastAt, polyWallet = time.Time{}, "" lastPositions, lastActivity = nil, nil }) @@ -125,7 +130,7 @@ func TestRefreshFastBacksOffAndKeepsLastPositions(t *testing.T) { // Now the API starts rate limiting. Force the next call to be due. atomic.StoreInt32(&rateLimited, 1) - polyNextAt = time.Time{} + positionsPoll.nextAt = time.Time{} refreshFast() if len(state.Positions) != 1 { t.Errorf("after 429: positions=%d, want the last good ones kept", len(state.Positions)) @@ -133,8 +138,8 @@ func TestRefreshFastBacksOffAndKeepsLastPositions(t *testing.T) { if !strings.Contains(state.Note, "rate limited") { t.Errorf("after 429: note=%q, want a rate-limit explanation", state.Note) } - if polyBackoff != polyBackoffMin { - t.Errorf("after 429: backoff=%v, want %v", polyBackoff, polyBackoffMin) + if positionsPoll.backoff != polyBackoffMin { + t.Errorf("after 429: backoff=%v, want %v", positionsPoll.backoff, polyBackoffMin) } // Subsequent cycles inside the backoff window must not touch the API. @@ -150,10 +155,114 @@ func TestRefreshFastBacksOffAndKeepsLastPositions(t *testing.T) { // Once the window passes and the API recovers, we resume and clear the note. atomic.StoreInt32(&rateLimited, 0) - polyNextAt = time.Now().Add(-time.Second) + positionsPoll.nextAt = time.Now().Add(-time.Second) + refreshFast() + if state.Note != "" || positionsPoll.backoff != 0 { + t.Errorf("after recovery: note=%q backoff=%v, want cleared", state.Note, positionsPoll.backoff) + } +} + +func TestRefreshFastPacesActivityIndependently(t *testing.T) { + var positionsHits, activityHits int32 + activityLimited := atomic.Bool{} + activityLimited.Store(true) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/positions": + atomic.AddInt32(&positionsHits, 1) + w.Write([]byte(`[]`)) + case "/activity": + atomic.AddInt32(&activityHits, 1) + if activityLimited.Load() { + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.Write([]byte(`[{"title":"filled"}]`)) + default: + http.NotFound(w, r) + } + })) + defer srv.Close() + + origBase := polyBase + polyBase = srv.URL + cfg = Config{Wallet: "0xtest", CandleDays: 1, Sort: "az"} + positionsPoll, activityPoll = pollSchedule{}, pollSchedule{} + pnlPoll = pollSchedule{nextAt: time.Now().Add(time.Hour)} + dataAPILastAt, polyWallet = time.Time{}, "0xtest" + lastPositions, lastActivity = nil, []Act{{Title: "cached"}} + t.Cleanup(func() { + polyBase = origBase + cfg, state = Config{}, State{} + positionsPoll, activityPoll, pnlPoll = pollSchedule{}, pollSchedule{}, pollSchedule{} + dataAPILastAt, polyWallet = time.Time{}, "" + lastPositions, lastActivity = nil, nil + }) + + // When both are due, positions runs first and activity is not burst behind it. refreshFast() - if state.Note != "" || polyBackoff != 0 { - t.Errorf("after recovery: note=%q backoff=%v, want cleared", state.Note, polyBackoff) + if positionsHits != 1 || activityHits != 0 { + t.Fatalf("startup hits: positions=%d activity=%d, want 1/0", positionsHits, activityHits) + } + + // A rate limit applies only to activity and retains its cached value. + positionsPoll.nextAt = time.Now().Add(time.Hour) + activityPoll.nextAt = time.Time{} + dataAPILastAt = time.Time{} + refreshFast() + if activityHits != 1 || activityPoll.backoff != polyBackoffMin { + t.Fatalf("after activity 429: hits=%d backoff=%v", activityHits, activityPoll.backoff) + } + if len(state.Activity) != 1 || state.Activity[0].Title != "cached" { + t.Fatalf("activity after 429 = %+v, want cached value", state.Activity) + } + if state.Note != "" { + t.Fatalf("activity failure should not mark positions stale: note=%q", state.Note) + } + + // It does not retry inside its own backoff, then resets after recovery. + refreshFast() + if activityHits != 1 { + t.Fatalf("activity calls during backoff=%d, want 1 total", activityHits) + } + activityLimited.Store(false) + activityPoll.nextAt = time.Time{} + dataAPILastAt = time.Time{} + refreshFast() + if activityHits != 2 || activityPoll.backoff != 0 { + t.Fatalf("after activity recovery: hits=%d backoff=%v", activityHits, activityPoll.backoff) + } + if len(state.Activity) != 1 || state.Activity[0].Title != "filled" { + t.Fatalf("activity after recovery = %+v", state.Activity) + } +} + +func TestRefreshPnlUsesThirtyMinuteCadence(t *testing.T) { + var hits int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&hits, 1) + w.Write([]byte(`[{"t":1,"p":2}]`)) + })) + defer srv.Close() + + origBase := pnlBase + pnlBase = srv.URL + pnlPoll = pollSchedule{} + pnlSeries = nil + t.Cleanup(func() { + pnlBase = origBase + pnlPoll = pollSchedule{} + pnlSeries = nil + }) + + now := time.Now() + refreshPnl("0xtest", now) + refreshPnl("0xtest", now.Add(time.Minute)) + if hits != 1 { + t.Fatalf("P/L hits=%d, want 1 inside 30-minute cadence", hits) + } + if wait := pnlPoll.nextAt.Sub(now); wait < pnlInterval || wait >= pnlInterval+pnlInterval/10 { + t.Fatalf("next P/L poll in %v, want [%v, %v)", wait, pnlInterval, pnlInterval+pnlInterval/10) } } @@ -658,13 +767,15 @@ func TestRefreshFastSkipsPolymarketWithoutWallet(t *testing.T) { origBase := polyBase polyBase = srv.URL cfg = Config{Wallet: " ", CandleDays: 1, Sort: "az"} - polyNextAt, polyBackoff, polyWallet = time.Time{}, 0, "stale" + positionsPoll, activityPoll, pnlPoll = pollSchedule{}, pollSchedule{}, pollSchedule{} + dataAPILastAt, polyWallet = time.Time{}, "stale" lastPositions = []Position{{Title: "leftover"}} lastActivity = []Act{{Title: "leftover"}} t.Cleanup(func() { polyBase = origBase cfg, state = Config{}, State{} - polyNextAt, polyBackoff, polyWallet = time.Time{}, 0, "" + positionsPoll, activityPoll, pnlPoll = pollSchedule{}, pollSchedule{}, pollSchedule{} + dataAPILastAt, polyWallet = time.Time{}, "" lastPositions, lastActivity = nil, nil })