From a72305ab3c248b3e23b2317d7d3bc3a05669b6a4 Mon Sep 17 00:00:00 2001
From: Samuel Tinnerholm <samuel.tinnerholm@gmail.com>
Date: Fri, 5 Jun 2026 14:36:05 +0000
Subject: [PATCH] fix: cap Kalshi pagination and serialize status=all

---
 core/src/exchanges/kalshi/fetcher.ts         | 13 +++--
 core/test/pipeline/kalshi-pagination.test.ts | 54 ++++++++++++++++++++
 2 files changed, 60 insertions(+), 7 deletions(-)

diff --git a/core/src/exchanges/kalshi/fetcher.ts b/core/src/exchanges/kalshi/fetcher.ts
index 17ab5000..ee63a6c8 100644
--- a/core/src/exchanges/kalshi/fetcher.ts
+++ b/core/src/exchanges/kalshi/fetcher.ts
@@ -223,11 +223,9 @@ export class KalshiFetcher implements IExchangeFetcher<KalshiRawEvent, KalshiRaw
             const status = (params?.status as string | undefined) || 'active';
 
             if (status === 'all') {
-                const [openEvents, closedEvents, settledEvents] = await Promise.all([
-                    this.fetchAllWithStatus('open'),
-                    this.fetchAllWithStatus('closed'),
-                    this.fetchAllWithStatus('settled'),
-                ]);
+                const openEvents = await this.fetchAllWithStatus('open');
+                const closedEvents = await this.fetchAllWithStatus('closed');
+                const settledEvents = await this.fetchAllWithStatus('settled');
                 return this.enrichEventsWithSeriesList([...openEvents, ...closedEvents, ...settledEvents]);
             } else if (status === 'closed' || status === 'inactive') {
                 const [closedEvents, settledEvents] = await Promise.all([
@@ -472,7 +470,8 @@ export class KalshiFetcher implements IExchangeFetcher<KalshiRawEvent, KalshiRaw
         }
 
         const isSorted = params?.sort && (params.sort === 'volume' || params.sort === 'liquidity');
-        const fetchLimit = isSorted ? 1000 : limit;
+        const safetyCap = BATCH_SIZE * 10;
+        const fetchLimit = isSorted ? Math.min(1000, safetyCap) : Math.min(limit, safetyCap);
 
         const [allEvents, fetchedSeriesMap] = await Promise.all([
             this.fetchActiveEvents(fetchLimit, apiStatus),
@@ -555,7 +554,7 @@ export class KalshiFetcher implements IExchangeFetcher<KalshiRawEvent, KalshiRaw
                 cursor = data.cursor;
                 page++;
 
-                if (!targetMarketCount && page >= 10) break;
+                if (page >= 10) break;
             } catch (e: any) {
                 throw kalshiErrorMapper.mapError(e);
             }
diff --git a/core/test/pipeline/kalshi-pagination.test.ts b/core/test/pipeline/kalshi-pagination.test.ts
index 53d24ad5..89004ea2 100644
--- a/core/test/pipeline/kalshi-pagination.test.ts
+++ b/core/test/pipeline/kalshi-pagination.test.ts
@@ -78,6 +78,29 @@ describe('Kalshi cursor pagination', () => {
         ]);
     });
 
+    it('caps the default fetch limit at 10 pages', async () => {
+        const { fetcher, calls } = createFetcher([
+            { events: buildEvents(200), cursor: 'cursor-200' },
+            { events: buildEvents(200, 200), cursor: 'cursor-400' },
+            { events: buildEvents(200, 400), cursor: 'cursor-600' },
+            { events: buildEvents(200, 600), cursor: 'cursor-800' },
+            { events: buildEvents(200, 800), cursor: 'cursor-1000' },
+            { events: buildEvents(200, 1000), cursor: 'cursor-1200' },
+            { events: buildEvents(200, 1200), cursor: 'cursor-1400' },
+            { events: buildEvents(200, 1400), cursor: 'cursor-1600' },
+            { events: buildEvents(200, 1600), cursor: 'cursor-1800' },
+            { events: buildEvents(200, 1800), cursor: 'cursor-2000' },
+            { events: buildEvents(200, 2000), cursor: 'cursor-2200' },
+        ]);
+
+        const events = await fetcher.fetchRawMarkets();
+
+        expect(events).toHaveLength(2000);
+        expect(calls).toHaveLength(10);
+        expect(calls[0]).toEqual({ limit: 200, with_nested_markets: true, status: 'open' });
+        expect(calls[9]).toEqual({ limit: 200, with_nested_markets: true, status: 'open', cursor: 'cursor-1800' });
+    });
+
     it('starts from a supplied cursor', async () => {
         const { fetcher, calls } = createFetcher([
             { events: buildEvents(25, 500), cursor: 'cursor-525' },
@@ -92,6 +115,37 @@ describe('Kalshi cursor pagination', () => {
         ]);
     });
 
+    it('runs status=all fetches sequentially instead of concurrently', async () => {
+        const active = { count: 0, max: 0 };
+        const responses: Array<{ events: unknown[]; cursor?: string | null }> = [
+            { events: buildEvents(1), cursor: null },
+            { events: buildEvents(1, 100), cursor: null },
+            { events: buildEvents(1, 200), cursor: null },
+        ];
+        const calls: Array<Record<string, unknown>> = [];
+        const ctx: any = {
+            http: {},
+            getHeaders: () => ({}),
+            callApi: async (operation: string, params?: Record<string, unknown>) => {
+                if (operation === 'GetSeriesList') return { series: [] };
+                if (operation !== 'GetEvents') return {};
+                calls.push(params ?? {});
+                active.count += 1;
+                active.max = Math.max(active.max, active.count);
+                await new Promise((resolve) => setTimeout(resolve, 0));
+                active.count -= 1;
+                return responses.shift() ?? { events: [], cursor: null };
+            },
+        };
+        const fetcher = new KalshiFetcher(ctx);
+
+        const events = await fetcher.fetchRawEvents({ status: 'all' } as any);
+
+        expect(events).toHaveLength(3);
+        expect(active.max).toBe(1);
+        expect(calls.map(call => call.status)).toEqual(['open', 'closed', 'settled']);
+    });
+
     it('enriches paginated events with series title and tags', async () => {
         const { fetcher, calls } = createSeriesFetcher();