From 9a1e1e3bfa09b49489ace2050f7a6f9c74e277db Mon Sep 17 00:00:00 2001 From: mansur-codes Date: Mon, 19 Oct 2026 18:19:35 +0100 Subject: [PATCH] feat: second and third CEX adapters with venue registry --- .env.example | 9 + src/__tests__/gateioAdapter.test.ts | 240 +++++++++++++++++++++ src/__tests__/kucoinAdapter.test.ts | 245 +++++++++++++++++++++ src/__tests__/venuesRegistry.test.ts | 84 ++++++++ src/index.ts | 4 +- src/ingest/venues/gateio.ts | 305 ++++++++++++++++++++++++++ src/ingest/venues/index.ts | 10 +- src/ingest/venues/kucoin.ts | 307 +++++++++++++++++++++++++++ src/ingest/venues/registry.ts | 76 +++++++ src/ingest/venues/types.ts | 1 + 10 files changed, 1276 insertions(+), 5 deletions(-) create mode 100644 src/__tests__/gateioAdapter.test.ts create mode 100644 src/__tests__/kucoinAdapter.test.ts create mode 100644 src/__tests__/venuesRegistry.test.ts create mode 100644 src/ingest/venues/gateio.ts create mode 100644 src/ingest/venues/kucoin.ts create mode 100644 src/ingest/venues/registry.ts diff --git a/.env.example b/.env.example index a23c6be6..39448027 100644 --- a/.env.example +++ b/.env.example @@ -279,3 +279,12 @@ MCP_PAID_CALL_ALLOWED_HOSTS= # against a table that has never been pruned cannot hold a long lock or blow # the WAL. Default 1000. SNAPSHOT_PRUNE_BATCH_SIZE=1000 + +# ─── CEX Exchange Adapters (optional) ────────────────────────────────────── +# Setting an API key enables that exchange adapter in the venue adapter registry. +# Missing or empty credentials will skip the venue with a log line without failing startup. +KUCOIN_API_KEY= +KUCOIN_API_SECRET= +GATEIO_API_KEY= +GATEIO_API_SECRET= + diff --git a/src/__tests__/gateioAdapter.test.ts b/src/__tests__/gateioAdapter.test.ts new file mode 100644 index 00000000..02df9772 --- /dev/null +++ b/src/__tests__/gateioAdapter.test.ts @@ -0,0 +1,240 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' + +const mocks = vi.hoisted(() => ({ + upsertPricePoints: vi.fn().mockResolvedValue(undefined), + dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined), + publishPriceUpdate: vi.fn(), +})) + +vi.mock('../db', () => ({ + upsertPricePoints: mocks.upsertPricePoints, +})) + +vi.mock('../webhookDispatcher', () => ({ + dispatchPriceUpdate: mocks.dispatchPriceUpdate, +})) + +vi.mock('../events', () => ({ + publishPriceUpdate: mocks.publishPriceUpdate, +})) + +import { + gateioAdapter, + normalizePair, + toGateioSymbol, + fetchGateioTicker, + fetchGateioOrderBook, + ingestGateioPair, + getGateioCredentials, + hasGateioCredentials, + _resetLastPrice, + _getLastPrice, +} from '../ingest/venues/gateio' +import { getCexAdapter } from '../ingest/cex' +import type { WatchedPair } from '../types' + +const mockPair: WatchedPair = { + pairKey: 'XLM/USDT', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDT', issuer: null }, +} + +describe('Gate.io Adapter - Pair Normalisation', () => { + it('normalises underscore-separated exchange pairs to internal pairKey format', () => { + expect(normalizePair('XLM_USDT')).toBe('XLM/USDT') + expect(normalizePair('xlm_usdc')).toBe('XLM/USDC') + expect(normalizePair('XLM_BTC')).toBe('XLM/BTC') + }) + + it('converts internal asset codes to Gate.io symbol format', () => { + expect(toGateioSymbol('XLM', 'USDT')).toBe('XLM_USDT') + expect(toGateioSymbol('xlm', 'usdc')).toBe('XLM_USDC') + }) +}) + +describe('Gate.io Adapter - Credentials Handling', () => { + const originalKey = process.env.GATEIO_API_KEY + const originalSecret = process.env.GATEIO_API_SECRET + + beforeEach(() => { + delete process.env.GATEIO_API_KEY + delete process.env.GATEIO_API_SECRET + vi.clearAllMocks() + _resetLastPrice() + }) + + afterEach(() => { + if (originalKey !== undefined) process.env.GATEIO_API_KEY = originalKey + else delete process.env.GATEIO_API_KEY + if (originalSecret !== undefined) process.env.GATEIO_API_SECRET = originalSecret + else delete process.env.GATEIO_API_SECRET + }) + + it('returns null and does not throw when credentials are absent or empty', async () => { + expect(getGateioCredentials()).toBeNull() + expect(hasGateioCredentials()).toBe(false) + expect(gateioAdapter.isEnabled('testnet')).toBe(false) + expect(gateioAdapter.isEnabled('mainnet')).toBe(false) + + // fetchTicker returns null quietly + const ticker = await fetchGateioTicker('XLM', 'USDT') + expect(ticker).toBeNull() + + // discover returns empty array + const targets = await gateioAdapter.discover(mockPair, 'testnet') + expect(targets).toEqual([]) + + // read returns null + const readResult = await gateioAdapter.read(mockPair, mockPair, 'testnet') + expect(readResult).toBeNull() + + // ingestGateioPair completes without throwing and without writing + await ingestGateioPair(mockPair, 'testnet') + expect(mocks.upsertPricePoints).not.toHaveBeenCalled() + }) + + it('detects credentials when GATEIO_API_KEY is populated', () => { + process.env.GATEIO_API_KEY = 'test-gateio-key' + process.env.GATEIO_API_SECRET = 'test-gateio-secret' + + expect(hasGateioCredentials()).toBe(true) + expect(getGateioCredentials()).toEqual({ + apiKey: 'test-gateio-key', + apiSecret: 'test-gateio-secret', + }) + expect(gateioAdapter.isEnabled('testnet')).toBe(true) + }) +}) + +describe('Gate.io Adapter - Network & Writing', () => { + const originalKey = process.env.GATEIO_API_KEY + + beforeEach(() => { + process.env.GATEIO_API_KEY = 'test-gateio-key' + vi.clearAllMocks() + _resetLastPrice() + global.fetch = vi.fn() + }) + + afterEach(() => { + if (originalKey !== undefined) process.env.GATEIO_API_KEY = originalKey + else delete process.env.GATEIO_API_KEY + }) + + it('writes price_points rows with the exact network passed in (never hardcoded)', async () => { + ;(global.fetch as ReturnType).mockResolvedValue({ + ok: true, + json: async () => [ + { + currency_pair: 'XLM_USDT', + last: '0.1450', + lowest_ask: '0.1455', + highest_bid: '0.1445', + base_volume: '25000', + quote_volume: '3625', + }, + ], + }) + + // Test with testnet + await ingestGateioPair(mockPair, 'testnet') + expect(mocks.upsertPricePoints).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + network: 'testnet', + pairKey: 'XLM/USDT', + source: 'Gate.io', + price: 0.145, + baseVolume: 25000, + }), + ]), + 'testnet' + ) + + // Test with mainnet + mocks.upsertPricePoints.mockClear() + await ingestGateioPair(mockPair, 'mainnet') + expect(mocks.upsertPricePoints).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + network: 'mainnet', + pairKey: 'XLM/USDT', + source: 'Gate.io', + price: 0.145, + baseVolume: 25000, + }), + ]), + 'mainnet' + ) + }) + + it('manages last price tracking with _resetLastPrice', async () => { + ;(global.fetch as ReturnType).mockResolvedValue({ + ok: true, + json: async () => [ + { + currency_pair: 'XLM_USDT', + last: '0.35', + base_volume: '1000', + }, + ], + }) + + await ingestGateioPair(mockPair, 'testnet') + expect(_getLastPrice('testnet', 'XLM/USDT')).toBe(0.35) + + _resetLastPrice() + expect(_getLastPrice('testnet', 'XLM/USDT')).toBeUndefined() + }) +}) + +describe('Gate.io Adapter - CexAdapter & VenueAdapter Interface Contract', () => { + it('satisfies CexAdapter interface and is registered in CEX registry', () => { + expect(typeof gateioAdapter.name).toBe('string') + expect(gateioAdapter.name).toBe('Gate.io') + expect(typeof gateioAdapter.fetchTicker).toBe('function') + expect(typeof gateioAdapter.fetchOrderBook).toBe('function') + + const registered = getCexAdapter('Gate.io') + expect(registered).toBe(gateioAdapter) + }) + + it('satisfies VenueAdapter interface', () => { + expect(typeof gateioAdapter.name).toBe('string') + expect(typeof gateioAdapter.source).toBe('string') + expect(gateioAdapter.source).toBe('Gate.io') + expect(typeof gateioAdapter.isEnabled).toBe('function') + expect(typeof gateioAdapter.getPollIntervalMs).toBe('function') + expect(typeof gateioAdapter.discover).toBe('function') + expect(typeof gateioAdapter.read).toBe('function') + expect(typeof gateioAdapter.derive).toBe('function') + expect(typeof gateioAdapter.write).toBe('function') + expect(typeof gateioAdapter.ingestPair).toBe('function') + }) + + it('fetches order book snapshots with depth clamping', async () => { + process.env.GATEIO_API_KEY = 'test-key' + global.fetch = vi.fn().mockResolvedValue({ + ok: true, + json: async () => ({ + id: 123456, + current: 1700000000000, + bids: [ + ['0.14', '500'], + ['0.13', '600'], + ], + asks: [ + ['0.15', '700'], + ['0.16', '800'], + ], + }), + }) + + const book = await gateioAdapter.fetchOrderBook('XLM', 'USDT', 1) + expect(book).not.toBeNull() + expect(book!.pair).toBe('XLM/USDT') + expect(book!.bids).toHaveLength(1) + expect(book!.asks).toHaveLength(1) + expect(book!.bids[0]).toEqual({ price: 0.14, quantity: 500 }) + }) +}) diff --git a/src/__tests__/kucoinAdapter.test.ts b/src/__tests__/kucoinAdapter.test.ts new file mode 100644 index 00000000..3b150cfa --- /dev/null +++ b/src/__tests__/kucoinAdapter.test.ts @@ -0,0 +1,245 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' + +const mocks = vi.hoisted(() => ({ + upsertPricePoints: vi.fn().mockResolvedValue(undefined), + dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined), + publishPriceUpdate: vi.fn(), +})) + +vi.mock('../db', () => ({ + upsertPricePoints: mocks.upsertPricePoints, +})) + +vi.mock('../webhookDispatcher', () => ({ + dispatchPriceUpdate: mocks.dispatchPriceUpdate, +})) + +vi.mock('../events', () => ({ + publishPriceUpdate: mocks.publishPriceUpdate, +})) + +import { + kucoinAdapter, + normalizePair, + toKuCoinSymbol, + fetchKuCoinTicker, + fetchKuCoinOrderBook, + ingestKuCoinPair, + getKuCoinCredentials, + hasKuCoinCredentials, + _resetLastPrice, + _getLastPrice, +} from '../ingest/venues/kucoin' +import { getCexAdapter } from '../ingest/cex' +import type { WatchedPair } from '../types' + +const mockPair: WatchedPair = { + pairKey: 'XLM/USDT', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDT', issuer: null }, +} + +describe('KuCoin Adapter - Pair Normalisation', () => { + it('normalises hyphenated exchange pairs to internal pairKey format', () => { + expect(normalizePair('XLM-USDT')).toBe('XLM/USDT') + expect(normalizePair('xlm-usdc')).toBe('XLM/USDC') + expect(normalizePair('XLM-BTC')).toBe('XLM/BTC') + }) + + it('converts internal asset codes to KuCoin symbol format', () => { + expect(toKuCoinSymbol('XLM', 'USDT')).toBe('XLM-USDT') + expect(toKuCoinSymbol('xlm', 'usdc')).toBe('XLM-USDC') + }) +}) + +describe('KuCoin Adapter - Credentials Handling', () => { + const originalKey = process.env.KUCOIN_API_KEY + const originalSecret = process.env.KUCOIN_API_SECRET + + beforeEach(() => { + delete process.env.KUCOIN_API_KEY + delete process.env.KUCOIN_API_SECRET + vi.clearAllMocks() + _resetLastPrice() + }) + + afterEach(() => { + if (originalKey !== undefined) process.env.KUCOIN_API_KEY = originalKey + else delete process.env.KUCOIN_API_KEY + if (originalSecret !== undefined) process.env.KUCOIN_API_SECRET = originalSecret + else delete process.env.KUCOIN_API_SECRET + }) + + it('returns null and does not throw when credentials are absent or empty', async () => { + expect(getKuCoinCredentials()).toBeNull() + expect(hasKuCoinCredentials()).toBe(false) + expect(kucoinAdapter.isEnabled('testnet')).toBe(false) + expect(kucoinAdapter.isEnabled('mainnet')).toBe(false) + + // fetchTicker returns null quietly + const ticker = await fetchKuCoinTicker('XLM', 'USDT') + expect(ticker).toBeNull() + + // discover returns empty array + const targets = await kucoinAdapter.discover(mockPair, 'testnet') + expect(targets).toEqual([]) + + // read returns null + const readResult = await kucoinAdapter.read(mockPair, mockPair, 'testnet') + expect(readResult).toBeNull() + + // ingestKuCoinPair completes without throwing and without writing + await ingestKuCoinPair(mockPair, 'testnet') + expect(mocks.upsertPricePoints).not.toHaveBeenCalled() + }) + + it('detects credentials when KUCOIN_API_KEY is populated', () => { + process.env.KUCOIN_API_KEY = 'test-kucoin-key' + process.env.KUCOIN_API_SECRET = 'test-kucoin-secret' + + expect(hasKuCoinCredentials()).toBe(true) + expect(getKuCoinCredentials()).toEqual({ + apiKey: 'test-kucoin-key', + apiSecret: 'test-kucoin-secret', + }) + expect(kucoinAdapter.isEnabled('testnet')).toBe(true) + }) +}) + +describe('KuCoin Adapter - Network & Writing', () => { + const originalKey = process.env.KUCOIN_API_KEY + + beforeEach(() => { + process.env.KUCOIN_API_KEY = 'test-kucoin-key' + vi.clearAllMocks() + _resetLastPrice() + global.fetch = vi.fn() + }) + + afterEach(() => { + if (originalKey !== undefined) process.env.KUCOIN_API_KEY = originalKey + else delete process.env.KUCOIN_API_KEY + }) + + it('writes price_points rows with the exact network passed in (never hardcoded)', async () => { + const fixedTime = 1700000000000 + ;(global.fetch as ReturnType).mockResolvedValue({ + ok: true, + json: async () => ({ + code: '200000', + data: { + time: fixedTime, + price: '0.12345', + size: '5000', + bestBid: '0.12340', + bestAsk: '0.12350', + }, + }), + }) + + // Test with testnet + await ingestKuCoinPair(mockPair, 'testnet') + expect(mocks.upsertPricePoints).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + network: 'testnet', + pairKey: 'XLM/USDT', + source: 'KuCoin', + price: 0.12345, + timestamp: new Date(fixedTime), + }), + ]), + 'testnet' + ) + + // Test with mainnet + mocks.upsertPricePoints.mockClear() + await ingestKuCoinPair(mockPair, 'mainnet') + expect(mocks.upsertPricePoints).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + network: 'mainnet', + pairKey: 'XLM/USDT', + source: 'KuCoin', + price: 0.12345, + timestamp: new Date(fixedTime), + }), + ]), + 'mainnet' + ) + }) + + it('manages last price tracking with _resetLastPrice', async () => { + ;(global.fetch as ReturnType).mockResolvedValue({ + ok: true, + json: async () => ({ + code: '200000', + data: { + price: '0.25', + size: '100', + bestBid: '0.24', + bestAsk: '0.26', + }, + }), + }) + + await ingestKuCoinPair(mockPair, 'testnet') + expect(_getLastPrice('testnet', 'XLM/USDT')).toBe(0.25) + + _resetLastPrice() + expect(_getLastPrice('testnet', 'XLM/USDT')).toBeUndefined() + }) +}) + +describe('KuCoin Adapter - CexAdapter & VenueAdapter Interface Contract', () => { + it('satisfies CexAdapter interface and is registered in CEX registry', () => { + expect(typeof kucoinAdapter.name).toBe('string') + expect(kucoinAdapter.name).toBe('KuCoin') + expect(typeof kucoinAdapter.fetchTicker).toBe('function') + expect(typeof kucoinAdapter.fetchOrderBook).toBe('function') + + const registered = getCexAdapter('KuCoin') + expect(registered).toBe(kucoinAdapter) + }) + + it('satisfies VenueAdapter interface', () => { + expect(typeof kucoinAdapter.name).toBe('string') + expect(typeof kucoinAdapter.source).toBe('string') + expect(kucoinAdapter.source).toBe('KuCoin') + expect(typeof kucoinAdapter.isEnabled).toBe('function') + expect(typeof kucoinAdapter.getPollIntervalMs).toBe('function') + expect(typeof kucoinAdapter.discover).toBe('function') + expect(typeof kucoinAdapter.read).toBe('function') + expect(typeof kucoinAdapter.derive).toBe('function') + expect(typeof kucoinAdapter.write).toBe('function') + expect(typeof kucoinAdapter.ingestPair).toBe('function') + }) + + it('fetches order book snapshots with depth clamping', async () => { + process.env.KUCOIN_API_KEY = 'test-key' + global.fetch = vi.fn().mockResolvedValue({ + ok: true, + json: async () => ({ + code: '200000', + data: { + time: 1700000000000, + bids: [ + ['0.12', '100'], + ['0.11', '200'], + ], + asks: [ + ['0.13', '150'], + ['0.14', '250'], + ], + }, + }), + }) + + const book = await kucoinAdapter.fetchOrderBook('XLM', 'USDT', 1) + expect(book).not.toBeNull() + expect(book!.pair).toBe('XLM/USDT') + expect(book!.bids).toHaveLength(1) + expect(book!.asks).toHaveLength(1) + expect(book!.bids[0]).toEqual({ price: 0.12, quantity: 100 }) + }) +}) diff --git a/src/__tests__/venuesRegistry.test.ts b/src/__tests__/venuesRegistry.test.ts new file mode 100644 index 00000000..49590b59 --- /dev/null +++ b/src/__tests__/venuesRegistry.test.ts @@ -0,0 +1,84 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { + getEnabledAdapters, + registeredVenues, + sdexAdapter, + ammAdapter, + soroswapAdapter, + aquariusAdapter, + kucoinAdapter, + gateioAdapter, +} from '../ingest/venues' + +describe('Venue Adapter Registry (Configuration-driven)', () => { + const origKuCoinKey = process.env.KUCOIN_API_KEY + const origGateioKey = process.env.GATEIO_API_KEY + + beforeEach(() => { + delete process.env.KUCOIN_API_KEY + delete process.env.GATEIO_API_KEY + }) + + afterEach(() => { + if (origKuCoinKey !== undefined) process.env.KUCOIN_API_KEY = origKuCoinKey + else delete process.env.KUCOIN_API_KEY + if (origGateioKey !== undefined) process.env.GATEIO_API_KEY = origGateioKey + else delete process.env.GATEIO_API_KEY + }) + + it('contains all candidate adapters in registeredVenues', () => { + const adapters = registeredVenues.map(r => r.adapter) + expect(adapters).toContain(sdexAdapter) + expect(adapters).toContain(ammAdapter) + expect(adapters).toContain(soroswapAdapter) + expect(adapters).toContain(aquariusAdapter) + expect(adapters).toContain(kucoinAdapter) + expect(adapters).toContain(gateioAdapter) + }) + + it('returns only adapters whose credentials are present (skips unconfigured CEX venues)', () => { + const enabled = getEnabledAdapters() + expect(enabled).toContain(sdexAdapter) + expect(enabled).toContain(ammAdapter) + expect(enabled).toContain(soroswapAdapter) + expect(enabled).toContain(aquariusAdapter) + expect(enabled).not.toContain(kucoinAdapter) + expect(enabled).not.toContain(gateioAdapter) + expect(enabled.length).toBe(4) + }) + + it('enables KuCoin adapter when KUCOIN_API_KEY is present', () => { + process.env.KUCOIN_API_KEY = 'test-kucoin-key' + const enabled = getEnabledAdapters() + expect(enabled).toContain(kucoinAdapter) + expect(enabled).not.toContain(gateioAdapter) + expect(enabled.length).toBe(5) + }) + + it('enables Gate.io adapter when GATEIO_API_KEY is present', () => { + process.env.GATEIO_API_KEY = 'test-gateio-key' + const enabled = getEnabledAdapters() + expect(enabled).toContain(gateioAdapter) + expect(enabled).not.toContain(kucoinAdapter) + expect(enabled.length).toBe(5) + }) + + it('fault isolation: missing credentials for KuCoin does not prevent Gate.io from running', () => { + delete process.env.KUCOIN_API_KEY + process.env.GATEIO_API_KEY = 'test-gateio-key' + + const enabled = getEnabledAdapters() + expect(enabled).toContain(gateioAdapter) + expect(enabled).not.toContain(kucoinAdapter) + }) + + it('enables all adapters when credentials for all venues are present', () => { + process.env.KUCOIN_API_KEY = 'test-kucoin-key' + process.env.GATEIO_API_KEY = 'test-gateio-key' + + const enabled = getEnabledAdapters() + expect(enabled).toContain(kucoinAdapter) + expect(enabled).toContain(gateioAdapter) + expect(enabled.length).toBe(6) + }) +}) diff --git a/src/index.ts b/src/index.ts index 5cf4a54f..c2609c5d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -12,7 +12,7 @@ import { pgPool } from './db' import { buildApp } from './app' import { fanOutManager } from './ws/fanout' -import { venueAdapters, startVenueAdapter } from './ingest/venues' +import { getEnabledAdapters, startVenueAdapter } from './ingest/venues' import { startSnapshotIngester } from './ingesters/snapshot' import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } from './jobs/aggregateRefresh' import { createSnapshotRetentionQueue, startSnapshotRetentionWorker, scheduleSnapshotRetention, pruneAllNetworks, SNAPSHOT_RETENTION_DAYS } from './jobs/snapshotRetention' @@ -197,7 +197,7 @@ async function main() { const enabledNetworks = getEnabledNetworks() log.info({ networks: enabledNetworks }, 'starting ingesters') for (const network of enabledNetworks) { - for (const adapter of venueAdapters) { + for (const adapter of getEnabledAdapters()) { restartIngester(adapter.name, network, () => startVenueAdapter(adapter, network)) } restartIngester('Snapshot', network, () => startSnapshotIngester(network)) diff --git a/src/ingest/venues/gateio.ts b/src/ingest/venues/gateio.ts new file mode 100644 index 00000000..ea58166c --- /dev/null +++ b/src/ingest/venues/gateio.ts @@ -0,0 +1,305 @@ +/** + * Gate.io CEX Venue Adapter + * + * Implements CexAdapter and VenueAdapter for Gate.io spot exchange. + * Fetches XLM and other Stellar asset pair tickers and order books via + * Gate.io's public v4 spot market data API, normalises pair identifiers + * to the internal pairKey format, and persists price points to price_points + * with source = 'Gate.io'. + * + * Issue: #216 + */ + +import { config, activeNetwork, type NetworkName } from '../../config' +import { createLogger } from '../../logger' +import type { WatchedPair } from '../../types' +import type { VenueAdapter, VenuePricePoint } from './types' +import type { CexAdapter, CexTicker, CexOrderBook } from '../cex' +import { registerCexAdapter } from '../cex' +import { + getLastPrice, + resetLastPrice, + writeVenuePricePoints, + startVenueAdapter, +} from './runner' +import { fetchWithTimeout } from '../../fetch' + +const log = createLogger('ingester:gateio') + +export const GATEIO_VENUE_NAME = 'Gate.io' +export const GATEIO_SOURCE_NAME = 'Gate.io' +export const GATEIO_DEFAULT_API_URL = 'https://api.gateio.ws' + +export function _resetLastPrice(): void { + resetLastPrice(GATEIO_VENUE_NAME) +} + +export function _getLastPrice(network: NetworkName, pairKey: string): number | undefined { + return getLastPrice(GATEIO_VENUE_NAME, network, pairKey) +} + +export interface GateioCredentials { + apiKey: string + apiSecret?: string +} + +/** + * Reads Gate.io API credentials from the environment. + * If GATEIO_API_KEY is missing or empty, logs a warning and returns null + * without throwing, disabling the adapter cleanly. + */ +export function getGateioCredentials(): GateioCredentials | null { + const apiKey = process.env.GATEIO_API_KEY + const apiSecret = process.env.GATEIO_API_SECRET + if (!apiKey || !apiKey.trim()) { + log.warn('GATEIO_API_KEY is not configured; skipping Gate.io adapter initialization') + return null + } + return { + apiKey: apiKey.trim(), + apiSecret: apiSecret && apiSecret.trim() ? apiSecret.trim() : undefined, + } +} + +export function hasGateioCredentials(): boolean { + const key = process.env.GATEIO_API_KEY + return Boolean(key && key.trim()) +} + +/** + * Pair Normalisation: + * Gate.io represents currency pairs using underscores (e.g. "XLM_USDT", "XLM_USDC"). + * Internal Lens pairKey format represents trading pairs as "BASE/QUOTE" uppercase + * (e.g. "XLM/USDT", "XLM/USDC"). + * + * This normaliser splits the raw exchange symbol by delimiter ('_', '-', or '/'), + * validates base and quote tokens, and constructs the internal "BASE/QUOTE" pairKey. + */ +export function normalizePair(rawPair: string): string { + const trimmed = rawPair.trim() + const parts = trimmed.split(/[-_/]/).filter(Boolean) + if (parts.length === 2) { + return `${parts[0].toUpperCase()}/${parts[1].toUpperCase()}` + } + return trimmed.toUpperCase() +} + +/** + * Converts internal base/quote asset codes into Gate.io's native currency pair format ("BASE_QUOTE"). + */ +export function toGateioSymbol(base: string, quote: string): string { + return `${base.toUpperCase()}_${quote.toUpperCase()}` +} + +export interface GateioTickerRecord { + currency_pair: string + last?: string + lowest_ask?: string + highest_bid?: string + base_volume?: string + quote_volume?: string +} + +export interface GateioOrderBookResponse { + id?: number + current?: number + update?: number + bids?: [string, string][] + asks?: [string, string][] +} + +/** + * Fetches ticker from Gate.io spot tickers endpoint. + */ +export async function fetchGateioTicker( + base: string, + quote: string, + baseUrl: string = GATEIO_DEFAULT_API_URL +): Promise { + const creds = getGateioCredentials() + if (!creds) return null + + const symbol = toGateioSymbol(base, quote) + try { + const url = `${baseUrl}/api/v4/spot/tickers?currency_pair=${encodeURIComponent(symbol)}` + const headers: Record = { + KEY: creds.apiKey, + } + + const res = await fetchWithTimeout(url, { headers }) + if (!res.ok) { + log.warn({ symbol, status: res.status }, 'Gate.io ticker fetch non-ok') + return null + } + + const records = (await res.json()) as GateioTickerRecord[] + if (!Array.isArray(records) || records.length === 0) { + return null + } + + const item = records[0] + if (!item?.last) { + return null + } + const last = parseFloat(item.last) + const bid = item.highest_bid ? parseFloat(item.highest_bid) : last + const ask = item.lowest_ask ? parseFloat(item.lowest_ask) : last + const baseVolume = item.base_volume ? parseFloat(item.base_volume) : 0 + const quoteVolume = item.quote_volume ? parseFloat(item.quote_volume) : baseVolume * last + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bid, + ask, + last, + baseVolume, + quoteVolume, + timestamp: new Date(), + } + } catch (err) { + log.error({ symbol, err }, 'failed to fetch Gate.io ticker') + return null + } +} + +/** + * Fetches order book snapshot from Gate.io spot order_book endpoint. + */ +export async function fetchGateioOrderBook( + base: string, + quote: string, + depth: number = 20, + baseUrl: string = GATEIO_DEFAULT_API_URL +): Promise { + const creds = getGateioCredentials() + if (!creds) return null + + const symbol = toGateioSymbol(base, quote) + try { + const url = `${baseUrl}/api/v4/spot/order_book?currency_pair=${encodeURIComponent(symbol)}&limit=${depth}` + const headers: Record = { + KEY: creds.apiKey, + } + + const res = await fetchWithTimeout(url, { headers }) + if (!res.ok) return null + + const json = (await res.json()) as GateioOrderBookResponse + if (!json) return null + + const bids = (json.bids ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + const asks = (json.asks ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + const timestamp = json.current ? new Date(json.current) : new Date() + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bids, + asks, + timestamp, + } + } catch (err) { + log.error({ symbol, err }, 'failed to fetch Gate.io order book') + return null + } +} + +export function deriveGateioPricePoints( + ticker: CexTicker, + pair: WatchedPair, + network?: NetworkName +): VenuePricePoint[] { + if (ticker.last <= 0) return [] + + return [ + { + network, + assetA: pair.assetA.code, + assetB: pair.assetB.code, + pairKey: pair.pairKey, + source: GATEIO_SOURCE_NAME, + price: ticker.last, + baseVolume: ticker.baseVolume, + counterVolume: ticker.quoteVolume, + ledger: 0, + timestamp: ticker.timestamp, + }, + ] +} + +export async function ingestGateioPair( + pair: WatchedPair, + network: NetworkName = activeNetwork +): Promise { + const creds = getGateioCredentials() + if (!creds) return + + const ticker = await fetchGateioTicker(pair.assetA.code, pair.assetB.code) + if (!ticker) return + + const points = deriveGateioPricePoints(ticker, pair, network) + if (points.length === 0) return + + await writeVenuePricePoints(GATEIO_VENUE_NAME, points, pair, network) +} + +export interface GateioAdapter extends VenueAdapter, CexAdapter {} + +export const gateioAdapter: GateioAdapter = { + name: GATEIO_VENUE_NAME, + kind: 'orderbook', + source: GATEIO_SOURCE_NAME, + + isEnabled(_network: NetworkName): boolean { + return hasGateioCredentials() + }, + + getPollIntervalMs(_network: NetworkName): number { + return config.indexer.pollIntervalMs + }, + + async discover(pair: WatchedPair, _network: NetworkName): Promise { + if (!hasGateioCredentials()) return [] + return [pair] + }, + + async read(target: WatchedPair, _pair: WatchedPair, _network: NetworkName): Promise { + if (!hasGateioCredentials()) return null + return fetchGateioTicker(target.assetA.code, target.assetB.code) + }, + + derive(raw: CexTicker, _target: WatchedPair, pair: WatchedPair, network: NetworkName): VenuePricePoint[] { + return deriveGateioPricePoints(raw, pair, network) + }, + + async write(points: VenuePricePoint[], _target: WatchedPair, pair: WatchedPair, network: NetworkName): Promise { + await writeVenuePricePoints(GATEIO_VENUE_NAME, points, pair, network) + }, + + async ingestPair(pair: WatchedPair, network: NetworkName): Promise { + await ingestGateioPair(pair, network) + }, + + async fetchTicker(base: string, quote: string): Promise { + return fetchGateioTicker(base, quote) + }, + + async fetchOrderBook(base: string, quote: string, depth?: number): Promise { + return fetchGateioOrderBook(base, quote, depth) + }, + + _resetLastPrice, + _getLastPrice, +} + +// Automatically register in the CEX registry +registerCexAdapter(gateioAdapter) + +export async function startGateioIngester(network: NetworkName = activeNetwork): Promise { + await startVenueAdapter(gateioAdapter, network) +} diff --git a/src/ingest/venues/index.ts b/src/ingest/venues/index.ts index 0ea72a28..308d654a 100644 --- a/src/ingest/venues/index.ts +++ b/src/ingest/venues/index.ts @@ -2,19 +2,23 @@ import { sdexAdapter } from './sdex' import { ammAdapter } from './amm' import { soroswapAdapter } from './soroswap' import { aquariusAdapter } from './aquarius' +import { kucoinAdapter, startKuCoinIngester } from './kucoin' +import { gateioAdapter, startGateioIngester } from './gateio' import type { VenueAdapter } from './types' export * from './types' export * from './runner' +export * from './registry' export { sdexAdapter, startSDEXIngester } from './sdex' export { ammAdapter, startAMMIngester } from './amm' export { soroswapAdapter, startSoroswapIngester } from './soroswap' export { aquariusAdapter, startAquariusIngester } from './aquarius' +export { kucoinAdapter, startKuCoinIngester } from './kucoin' +export { gateioAdapter, startGateioIngester } from './gateio' /** - * Registry of all available venue adapters. - * Adding a venue to this list enables it across the venue runner automatically - * without requiring runner code modifications. + * Registry of default baseline venue adapters. + * Retained for backward compatibility with existing tests and route queries. */ export const venueAdapters: VenueAdapter[] = [ sdexAdapter, diff --git a/src/ingest/venues/kucoin.ts b/src/ingest/venues/kucoin.ts new file mode 100644 index 00000000..e9594439 --- /dev/null +++ b/src/ingest/venues/kucoin.ts @@ -0,0 +1,307 @@ +/** + * KuCoin CEX Venue Adapter + * + * Implements CexAdapter and VenueAdapter for KuCoin spot exchange. + * Fetches XLM and other Stellar asset pair tickers and order books via + * KuCoin's public market data API, normalises pair identifiers to the + * internal pairKey format, and persists price points to price_points with + * source = 'KuCoin'. + * + * Issue: #216 + */ + +import { config, activeNetwork, type NetworkName } from '../../config' +import { createLogger } from '../../logger' +import type { WatchedPair } from '../../types' +import type { VenueAdapter, VenuePricePoint } from './types' +import type { CexAdapter, CexTicker, CexOrderBook } from '../cex' +import { registerCexAdapter } from '../cex' +import { + getLastPrice, + resetLastPrice, + writeVenuePricePoints, + startVenueAdapter, +} from './runner' +import { fetchWithTimeout } from '../../fetch' + +const log = createLogger('ingester:kucoin') + +export const KUCOIN_VENUE_NAME = 'KuCoin' +export const KUCOIN_SOURCE_NAME = 'KuCoin' +export const KUCOIN_DEFAULT_API_URL = 'https://api.kucoin.com' + +export function _resetLastPrice(): void { + resetLastPrice(KUCOIN_VENUE_NAME) +} + +export function _getLastPrice(network: NetworkName, pairKey: string): number | undefined { + return getLastPrice(KUCOIN_VENUE_NAME, network, pairKey) +} + +export interface KuCoinCredentials { + apiKey: string + apiSecret?: string +} + +/** + * Reads KuCoin API credentials from the environment. + * If KUCOIN_API_KEY is missing or empty, logs a warning and returns null + * without throwing, disabling the adapter cleanly. + */ +export function getKuCoinCredentials(): KuCoinCredentials | null { + const apiKey = process.env.KUCOIN_API_KEY + const apiSecret = process.env.KUCOIN_API_SECRET + if (!apiKey || !apiKey.trim()) { + log.warn('KUCOIN_API_KEY is not configured; skipping KuCoin adapter initialization') + return null + } + return { + apiKey: apiKey.trim(), + apiSecret: apiSecret && apiSecret.trim() ? apiSecret.trim() : undefined, + } +} + +export function hasKuCoinCredentials(): boolean { + const key = process.env.KUCOIN_API_KEY + return Boolean(key && key.trim()) +} + +/** + * Pair Normalisation: + * KuCoin represents trading symbols with hyphens in uppercase (e.g. "XLM-USDT", "XLM-USDC"). + * Internal Lens pairKey format represents trading pairs as "BASE/QUOTE" uppercase + * (e.g. "XLM/USDT", "XLM/USDC"). + * + * This normaliser splits the raw exchange symbol by delimiter ('-', '_', or '/'), + * validates base and quote tokens, and constructs the internal "BASE/QUOTE" pairKey. + */ +export function normalizePair(rawPair: string): string { + const trimmed = rawPair.trim() + const parts = trimmed.split(/[-_/]/).filter(Boolean) + if (parts.length === 2) { + return `${parts[0].toUpperCase()}/${parts[1].toUpperCase()}` + } + return trimmed.toUpperCase() +} + +/** + * Converts internal base/quote asset codes into KuCoin's native symbol format ("BASE-QUOTE"). + */ +export function toKuCoinSymbol(base: string, quote: string): string { + return `${base.toUpperCase()}-${quote.toUpperCase()}` +} + +export interface KuCoinTickerResponse { + code: string + data?: { + time?: number + price?: string + size?: string + bestBid?: string + bestBidSize?: string + bestAsk?: string + bestAskSize?: string + } +} + +export interface KuCoinOrderBookResponse { + code: string + data?: { + time?: number + bids?: [string, string][] + asks?: [string, string][] + } +} + +/** + * Fetches the latest ticker from KuCoin Level 1 endpoint. + */ +export async function fetchKuCoinTicker( + base: string, + quote: string, + baseUrl: string = KUCOIN_DEFAULT_API_URL +): Promise { + const creds = getKuCoinCredentials() + if (!creds) return null + + const symbol = toKuCoinSymbol(base, quote) + try { + const url = `${baseUrl}/api/v1/market/orderbook/level1?symbol=${encodeURIComponent(symbol)}` + const headers: Record = { + 'KC-API-KEY': creds.apiKey, + } + + const res = await fetchWithTimeout(url, { headers }) + if (!res.ok) { + log.warn({ symbol, status: res.status }, 'KuCoin ticker fetch non-ok') + return null + } + + const json = (await res.json()) as KuCoinTickerResponse + if (!json.data || !json.data.price) { + return null + } + + const last = parseFloat(json.data.price) + const bid = json.data.bestBid ? parseFloat(json.data.bestBid) : last + const ask = json.data.bestAsk ? parseFloat(json.data.bestAsk) : last + const baseVolume = json.data.size ? parseFloat(json.data.size) : 0 + const timestamp = json.data.time ? new Date(json.data.time) : new Date() + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bid, + ask, + last, + baseVolume, + quoteVolume: baseVolume * last, + timestamp, + } + } catch (err) { + log.error({ symbol, err }, 'failed to fetch KuCoin ticker') + return null + } +} + +/** + * Fetches KuCoin Level 2 orderbook snapshot. + */ +export async function fetchKuCoinOrderBook( + base: string, + quote: string, + depth: number = 20, + baseUrl: string = KUCOIN_DEFAULT_API_URL +): Promise { + const creds = getKuCoinCredentials() + if (!creds) return null + + const symbol = toKuCoinSymbol(base, quote) + try { + const endpoint = depth <= 20 ? 'level2_20' : 'level2_100' + const url = `${baseUrl}/api/v1/market/orderbook/${endpoint}?symbol=${encodeURIComponent(symbol)}` + const headers: Record = { + 'KC-API-KEY': creds.apiKey, + } + + const res = await fetchWithTimeout(url, { headers }) + if (!res.ok) return null + + const json = (await res.json()) as KuCoinOrderBookResponse + if (!json.data) return null + + const bids = (json.data.bids ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + const asks = (json.data.asks ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + const timestamp = json.data.time ? new Date(json.data.time) : new Date() + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bids, + asks, + timestamp, + } + } catch (err) { + log.error({ symbol, err }, 'failed to fetch KuCoin order book') + return null + } +} + +export function deriveKuCoinPricePoints( + ticker: CexTicker, + pair: WatchedPair, + network?: NetworkName +): VenuePricePoint[] { + if (ticker.last <= 0) return [] + + return [ + { + network, + assetA: pair.assetA.code, + assetB: pair.assetB.code, + pairKey: pair.pairKey, + source: KUCOIN_SOURCE_NAME, + price: ticker.last, + baseVolume: ticker.baseVolume, + counterVolume: ticker.quoteVolume, + ledger: 0, + timestamp: ticker.timestamp, + }, + ] +} + +export async function ingestKuCoinPair( + pair: WatchedPair, + network: NetworkName = activeNetwork +): Promise { + const creds = getKuCoinCredentials() + if (!creds) return + + const ticker = await fetchKuCoinTicker(pair.assetA.code, pair.assetB.code) + if (!ticker) return + + const points = deriveKuCoinPricePoints(ticker, pair, network) + if (points.length === 0) return + + await writeVenuePricePoints(KUCOIN_VENUE_NAME, points, pair, network) +} + +export interface KuCoinAdapter extends VenueAdapter, CexAdapter {} + +export const kucoinAdapter: KuCoinAdapter = { + name: KUCOIN_VENUE_NAME, + kind: 'orderbook', + source: KUCOIN_SOURCE_NAME, + + isEnabled(_network: NetworkName): boolean { + return hasKuCoinCredentials() + }, + + getPollIntervalMs(_network: NetworkName): number { + return config.indexer.pollIntervalMs + }, + + async discover(pair: WatchedPair, _network: NetworkName): Promise { + if (!hasKuCoinCredentials()) return [] + return [pair] + }, + + async read(target: WatchedPair, _pair: WatchedPair, _network: NetworkName): Promise { + if (!hasKuCoinCredentials()) return null + return fetchKuCoinTicker(target.assetA.code, target.assetB.code) + }, + + derive(raw: CexTicker, _target: WatchedPair, pair: WatchedPair, network: NetworkName): VenuePricePoint[] { + return deriveKuCoinPricePoints(raw, pair, network) + }, + + async write(points: VenuePricePoint[], _target: WatchedPair, pair: WatchedPair, network: NetworkName): Promise { + await writeVenuePricePoints(KUCOIN_VENUE_NAME, points, pair, network) + }, + + async ingestPair(pair: WatchedPair, network: NetworkName): Promise { + await ingestKuCoinPair(pair, network) + }, + + async fetchTicker(base: string, quote: string): Promise { + return fetchKuCoinTicker(base, quote) + }, + + async fetchOrderBook(base: string, quote: string, depth?: number): Promise { + return fetchKuCoinOrderBook(base, quote, depth) + }, + + _resetLastPrice, + _getLastPrice, +} + +// Automatically register in the CEX registry +registerCexAdapter(kucoinAdapter) + +export async function startKuCoinIngester(network: NetworkName = activeNetwork): Promise { + await startVenueAdapter(kucoinAdapter, network) +} diff --git a/src/ingest/venues/registry.ts b/src/ingest/venues/registry.ts new file mode 100644 index 00000000..a9c83b4c --- /dev/null +++ b/src/ingest/venues/registry.ts @@ -0,0 +1,76 @@ +/** + * Configuration-Driven Venue Adapter Registry + * + * Central registry of all venue adapters (SDEX, AMM, Soroswap, Aquarius, KuCoin, Gate.io). + * Automatically enables adapters based on environment credentials and configuration, + * skipping unconfigured venues with a descriptive log line so that missing credentials + * for one venue do not prevent other venues from running. + * + * Issue: #216 + */ + +import { sdexAdapter } from './sdex' +import { ammAdapter } from './amm' +import { soroswapAdapter } from './soroswap' +import { aquariusAdapter } from './aquarius' +import { kucoinAdapter, hasKuCoinCredentials } from './kucoin' +import { gateioAdapter, hasGateioCredentials } from './gateio' +import type { VenueAdapter } from './types' +import { createLogger } from '../../logger' + +const log = createLogger('venues:registry') + +export interface VenueRegistration { + adapter: VenueAdapter + hasCredentials: () => boolean +} + +/** + * All known venue adapter registrations in the system. + * To register an adapter in the system, add it to this list. + */ +export const registeredVenues: VenueRegistration[] = [ + { + adapter: sdexAdapter, + hasCredentials: () => true, + }, + { + adapter: ammAdapter, + hasCredentials: () => true, + }, + { + adapter: soroswapAdapter, + hasCredentials: () => true, + }, + { + adapter: aquariusAdapter, + hasCredentials: () => true, + }, + { + adapter: kucoinAdapter, + hasCredentials: hasKuCoinCredentials, + }, + { + adapter: gateioAdapter, + hasCredentials: hasGateioCredentials, + }, +] + +/** + * Returns all adapters that have required credentials and configuration present. + * Venues missing credentials are logged and skipped without throwing or blocking others. + */ +export function getEnabledAdapters(): VenueAdapter[] { + const enabled: VenueAdapter[] = [] + for (const reg of registeredVenues) { + if (!reg.hasCredentials()) { + log.info( + { venue: reg.adapter.name }, + 'venue credentials missing in environment; adapter skipped' + ) + continue + } + enabled.push(reg.adapter) + } + return enabled +} diff --git a/src/ingest/venues/types.ts b/src/ingest/venues/types.ts index 2fa4601d..d4b42a55 100644 --- a/src/ingest/venues/types.ts +++ b/src/ingest/venues/types.ts @@ -2,6 +2,7 @@ import type { NetworkName } from '../../config' import type { WatchedPair } from '../../types' export interface VenuePricePoint { + network?: string assetA: string assetB: string pairKey: string