fix(core): sync Gemini Titan order and websocket handling (#957)

Author: realfishsam

core/src/exchanges/gemini-titan/fetcher.ts

@@ -7,6 +7,7 @@ import {
     GeminiRawEventsResponse,
     GeminiRawOrder,
     GeminiRawActiveOrdersResponse,
+    GeminiRawOrderHistoryResponse,
     GeminiRawPosition,
     GeminiRawPositionsResponse,
     GeminiRawOrderBook,
@@ -135,30 +136,50 @@ export class GeminiFetcher implements IExchangeFetcher<GeminiRawEvent, GeminiRaw
         );
     }
 
-    async cancelRawOrder(orderId: number): Promise<{ result: string; message: string }> {
-        return this.postAuthenticated<{ result: string; message: string }>(
+    async cancelRawOrder(orderId: number): Promise<GeminiRawOrder> {
+        return this.postAuthenticated<GeminiRawOrder>(
             '/v1/prediction-markets/order/cancel',
             { orderId },
         );
     }
 
     async fetchRawActiveOrders(symbol?: string): Promise<GeminiRawOrder[]> {
-        const extra: Record<string, unknown> = {};
-        if (symbol) extra.symbol = symbol;
-
-        const response = await this.postAuthenticated<GeminiRawActiveOrdersResponse>(
-            '/v1/prediction-markets/orders/active',
-            extra,
-        );
-        return response.orders;
+        return this.fetchPaginatedOrders('/v1/prediction-markets/orders/active', symbol ? { symbol } : {});
     }
 
     async fetchRawOrderHistory(): Promise<GeminiRawOrder[]> {
-        const response = await this.postAuthenticated<GeminiRawOrder[]>(
-            '/v1/prediction-markets/orders/history',
-            {},
-        );
-        return Array.isArray(response) ? response : [];
+        return this.fetchPaginatedOrders('/v1/prediction-markets/orders/history', {});
+    }
+
+    private async fetchPaginatedOrders(
+        path: '/v1/prediction-markets/orders/active' | '/v1/prediction-markets/orders/history',
+        extra: Record<string, unknown>,
+    ): Promise<GeminiRawOrder[]> {
+        const allOrders: GeminiRawOrder[] = [];
+        const limit = 100;
+        let offset = 0;
+
+        while (true) {
+            const response = await this.postAuthenticated<GeminiRawActiveOrdersResponse | GeminiRawOrderHistoryResponse>(
+                path,
+                { ...extra, limit, offset },
+            );
+
+            const orders = response.orders ?? [];
+            allOrders.push(...orders);
+
+            const pagination = response.pagination;
+            const count = pagination?.count ?? orders.length;
+            const pageOffset = pagination?.offset ?? offset;
+
+            if (orders.length === 0 || pageOffset + orders.length >= count) {
+                break;
+            }
+
+            offset = pageOffset + orders.length;
+        }
+
+        return allOrders;
     }
 
     async fetchRawPositions(): Promise<GeminiRawPosition[]> {

core/src/exchanges/gemini-titan/index.ts

@@ -263,19 +263,8 @@ export class GeminiTitanExchange extends PredictionMarketExchange {
         this.requireAuth();
 
         try {
-            await this.fetcher.cancelRawOrder(parseInt(orderId, 10));
-            return {
-                id: orderId,
-                marketId: '',
-                outcomeId: '',
-                side: 'buy',
-                type: 'limit',
-                amount: 0,
-                status: 'canceled',
-                filled: 0,
-                remaining: 0,
-                timestamp: Date.now(),
-            };
+            const rawOrder = await this.fetcher.cancelRawOrder(parseInt(orderId, 10));
+            return this.normalizer.normalizeOrder(rawOrder);
         } catch (error: any) {
             throw geminiErrorMapper.mapError(error);
         }

core/src/exchanges/gemini-titan/types.ts

@@ -158,6 +158,8 @@ export interface GeminiRawActiveOrdersResponse {
     };
 }
 
+export type GeminiRawOrderHistoryResponse = GeminiRawActiveOrdersResponse;
+
 export interface GeminiRawPositionsResponse {
     positions: GeminiRawPosition[];
     total?: number;

core/src/exchanges/gemini-titan/websocket.ts

@@ -19,7 +19,7 @@ export interface GeminiWebSocketConfig {
  * Gemini Titan WebSocket for real-time order book and trade streaming.
  *
  * Subscribes to:
- *   - {symbol}@depth20  (L2 partial depth snapshots at 1s intervals)
+ *   - {symbol}@depth@100ms  (full depth snapshots/deltas at 100ms intervals)
  *   - {symbol}@trade    (executed trades)
  *
  * Auth headers are sent during the handshake if credentials are provided
@@ -75,7 +75,7 @@ export class GeminiWebSocket {
                     // Resubscribe on reconnect
                     const allStreams: string[] = [];
                     for (const sym of this.subscribedDepthSymbols) {
-                        allStreams.push(`${sym}@depth20`);
+                        allStreams.push(this.depthStream(sym));
                     }
                     for (const sym of this.subscribedTradeSymbols) {
                         allStreams.push(`${sym}@trade`);
@@ -152,7 +152,7 @@ export class GeminiWebSocket {
 
     private handleMessage(message: any): void {
         // Gemini sends flat objects, NOT wrapped in { stream, data }.
-        // Depth snapshots: { lastUpdateId, symbol, bids, asks }
+        // Depth snapshots: { lastUpdateId, s, bids, asks }
         // Depth deltas:    { e, E, s, U, u, b, a }
         // Trades:          { E, s, t, p, q, m }
         // Confirmations:   { id, status: 200 }
@@ -171,7 +171,12 @@ export class GeminiWebSocket {
     private handleDepthSnapshot(data: any): void {
         // symbol comes back lowercase from the API, but we subscribed with
         // uppercase. Normalize to uppercase for resolver lookup.
-        const symbol = (data.symbol as string).toUpperCase();
+        const rawSymbol = data.s ?? data.symbol;
+        if (typeof rawSymbol !== 'string' || rawSymbol.length === 0) {
+            logger.warn('[gemini-titan] depth snapshot missing symbol field');
+            return;
+        }
+        const symbol = rawSymbol.toUpperCase();
 
         const bids: OrderLevel[] = (data.bids ?? []).map((level: [string, string]) => ({
             price: parseFloat(level[0]),
@@ -277,7 +282,7 @@ export class GeminiWebSocket {
                 }
             });
         } else {
-            this.sendSubscribe([`${symbol}@depth20`]);
+            this.sendSubscribe([this.depthStream(symbol)]);
         }
 
         const dataPromise = new Promise<OrderBook>((resolve, reject) => {
@@ -338,6 +343,10 @@ export class GeminiWebSocket {
         );
     }
 
+    private depthStream(symbol: string): string {
+        return `${symbol}@depth@100ms`;
+    }
+
     async close(): Promise<void> {
         this.isTerminated = true;
 

core/test/exchanges/gemini-titan-fetcher.test.ts

@@ -0,0 +1,84 @@
+import { GeminiFetcher } from '../../src/exchanges/gemini-titan/fetcher';
+import { FetcherContext } from '../../src/exchanges/interfaces';
+
+function makeFetcher(responses: unknown[]) {
+    const post = jest.fn(async () => ({ data: responses.shift() }));
+    const buildHeaders = jest.fn(() => ({ 'X-GEMINI-APIKEY': 'test-key' }));
+    const auth = {
+        nonce: jest.fn(() => 12345),
+        buildHeaders,
+    } as any;
+    const ctx: FetcherContext = {
+        http: { post } as any,
+        callApi: jest.fn() as any,
+        getHeaders: jest.fn(() => ({})),
+    };
+
+    return {
+        fetcher: new GeminiFetcher(ctx, 'https://api.gemini.test', auth),
+        post,
+        buildHeaders,
+    };
+}
+
+describe('GeminiFetcher authenticated orders', () => {
+    it('reads paginated order history envelopes', async () => {
+        const { fetcher, buildHeaders } = makeFetcher([
+            {
+                orders: [{ orderId: 1, status: 'filled' }],
+                pagination: { limit: 100, offset: 0, count: 2 },
+            },
+            {
+                orders: [{ orderId: 2, status: 'cancelled' }],
+                pagination: { limit: 100, offset: 1, count: 2 },
+            },
+        ]);
+
+        await expect(fetcher.fetchRawOrderHistory()).resolves.toEqual([
+            { orderId: 1, status: 'filled' },
+            { orderId: 2, status: 'cancelled' },
+        ]);
+
+        expect(buildHeaders).toHaveBeenNthCalledWith(1, expect.objectContaining({
+            request: '/v1/prediction-markets/orders/history',
+            limit: 100,
+            offset: 0,
+        }));
+        expect(buildHeaders).toHaveBeenNthCalledWith(2, expect.objectContaining({
+            request: '/v1/prediction-markets/orders/history',
+            limit: 100,
+            offset: 1,
+        }));
+    });
+
+    it('passes limit and offset when fetching active orders', async () => {
+        const { fetcher, buildHeaders } = makeFetcher([
+            {
+                orders: [],
+                pagination: { limit: 100, offset: 0, count: 0 },
+            },
+        ]);
+
+        await expect(fetcher.fetchRawActiveOrders('BTCUSD-PERP')).resolves.toEqual([]);
+
+        expect(buildHeaders).toHaveBeenCalledWith(expect.objectContaining({
+            request: '/v1/prediction-markets/orders/active',
+            symbol: 'BTCUSD-PERP',
+            limit: 100,
+            offset: 0,
+        }));
+    });
+
+    it('returns the full raw cancel order response', async () => {
+        const rawOrder = {
+            orderId: 123,
+            symbol: 'BTCUSD-PERP',
+            side: 'buy',
+            outcome: 'yes',
+            status: 'cancelled',
+        };
+        const { fetcher } = makeFetcher([rawOrder]);
+
+        await expect(fetcher.cancelRawOrder(123)).resolves.toBe(rawOrder);
+    });
+});

core/test/exchanges/gemini-titan-websocket.test.ts

@@ -0,0 +1,30 @@
+import { GeminiWebSocket } from '../../src/exchanges/gemini-titan/websocket';
+
+describe('GeminiWebSocket depth snapshots', () => {
+    it('routes snapshots using s when symbol is absent', () => {
+        const ws = new GeminiWebSocket(undefined, { wsUrl: 'wss://example.test' }) as any;
+        const resolved: unknown[] = [];
+        ws.orderBookResolvers.set('BTCUSD-PERP', [{
+            resolve: (value: unknown) => resolved.push(value),
+            reject: jest.fn(),
+        }]);
+
+        ws.handleDepthSnapshot({
+            lastUpdateId: 1,
+            s: 'btcusd-perp',
+            bids: [['0.48', '10']],
+            asks: [['0.52', '12']],
+        });
+
+        expect(resolved).toEqual([{
+            bids: [{ price: 0.48, size: 10 }],
+            asks: [{ price: 0.52, size: 12 }],
+            timestamp: expect.any(Number),
+        }]);
+    });
+
+    it('uses the documented full-depth stream name', () => {
+        const ws = new GeminiWebSocket(undefined, { wsUrl: 'wss://example.test' }) as any;
+        expect(ws.depthStream('BTCUSD-PERP')).toBe('BTCUSD-PERP@depth@100ms');
+    });
+});