Tardis.dev와 OKX API 이중 데이터소스 실시간 시장 데이터 집계 튜토리얼

개요

암호화폐 거래소에서 실시간 시장 데이터를 안정적으로 수집하는 것은 고빈도 거래 시스템, 봇 개발, 데이터 분석에 필수적이다. 이 튜토리얼에서는 Tardis.dev(캡처 및 리플레이 서비스)와 OKX API를 활용한 **이중 데이터소스 실시간 시장 데이터 집계 아키텍처**를 구축한다.

┌─────────────────────────────────────────────────────────────┐
│                   실시간 시장 데이터 집계 아키텍처              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌──────────────┐         ┌──────────────┐               │
│   │   OKX API    │         │ Tardis.dev   │               │
│   │  (WebSocket) │         │  (Captured)  │               │
│   └──────┬───────┘         └──────┬───────┘               │
│          │                        │                        │
│          ▼                        ▼                        │
│   ┌──────────────┐         ┌──────────────┐               │
│   │ Primary Feed │         │  Backup Feed │               │
│   │  (실시간)     │         │   (리플레이)  │               │
│   └──────┬───────┘         └──────┬───────┘               │
│          │         ┌─────────────┘                        │
│          ▼         ▼                                       │
│   ┌─────────────────────────────────────┐                 │
│   │        Data Aggregator              │                 │
│   │    (중복 제거 + 정렬 + 병합)          │                 │
│   └──────────────────┬──────────────────┘                 │
│                      ▼                                     │
│   ┌─────────────────────────────────────┐                 │
│   │     Unified Market Data Stream      │                 │
│   └─────────────────────────────────────┘                 │
└─────────────────────────────────────────────────────────────┘

필수 사전 준비

Node.js 프로젝트 초기화
npm init -y

필요한 패키지 설치
npm install @tardis-dev/client
npm install @okx-api/api
npm install ws          # WebSocket 클라이언트
npm install axios       # HTTP 요청
npm install dotenv      # 환경변수 관리

.env 파일 설정
OKX_API_KEY=your_okx_api_key
OKX_SECRET_KEY=your_okx_secret_key
OKX_PASSPHRASE=your_okx_passphrase
OKX_FLAG=0  # 실거래: 0, 테스트넷: 1
TARDIS_API_KEY=your_tardis_api_key

Tardis.dev 기본 설정

Tardis.dev는 주요 거래소의 시장 데이터를 캡처하여 API와 SDK로 제공하는 서비스이다.

// tardis-client.js
const { createClient } = require('@tardis-dev/client');

class TardisDataSource {
  constructor(apiKey) {
    this.client = createClient({ apiKey });
    this.subscriptions = new Map();
  }

  // 특정 거래소 데이터订阅
  async subscribeExchange(exchange, symbol, messageHandler) {
    const subscription = this.client.subscribe({
      exchange,
      channel: 'trades',
      symbol
    });

    subscription.on('message', (message) => {
      const normalizedData = this.normalizeTardisTrade(message);
      messageHandler(normalizedData);
    });

    subscription.on('error', (error) => {
      console.error([Tardis] ${exchange} 오류:, error.message);
    });

    this.subscriptions.set(${exchange}:${symbol}, subscription);
    console.log([Tardis] ${exchange} ${symbol} 구독 시작);
    return subscription;
  }

  // Tardis 메시지를 표준 형식으로 변환
  normalizeTardisTrade(message) {
    return {
      source: 'tardis',
      exchange: message.exchange,
      symbol: message.symbol,
      price: parseFloat(message.price),
      quantity: parseFloat(message.side === 'buy' ? message.amount : -message.amount),
      side: message.side,
      timestamp: new Date(message.timestamp).getTime(),
      tradeId: message.id,
      localTimestamp: Date.now()
    };
  }

  // 특정 시간 범위 데이터 조회 (리플레이)
  async replayRange(exchange, symbol, startTime, endTime) {
    const messages = [];
    
    for await (const message of this.client.replay({
      exchange,
      channel: 'trades',
      symbol,
      from: startTime,
      to: endTime
    })) {
      messages.push(this.normalizeTardisTrade(message));
    }
    
    return messages;
  }

  disconnect() {
    for (const sub of this.subscriptions.values()) {
      sub.destroy();
    }
    this.subscriptions.clear();
    console.log('[Tardis] 모든 연결 해제됨');
  }
}

module.exports = TardisDataSource;

OKX API 실시간 데이터 수집

// okx-websocket-client.js
const WebSocket = require('ws');
const crypto = require('crypto');

class OKXWebSocketClient {
  constructor(apiKey, secretKey, passphrase, flag = '0') {
    this.apiKey = apiKey;
    this.secretKey = secretKey;
    this.passphrase = passphrase;
    this.flag = flag;
    this.ws = null;
    this.subscriptions = new Map();
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 10;
    this.pingInterval = null;
  }

  // WebSocket 연결 URL 생성 (서명 포함)
  generateWsUrl() {
    const timestamp = new Date().toISOString();
    const sign = crypto
      .createHmac('sha256', this.secretKey)
      .update(timestamp + 'GET' + '/users/self/verify')
      .digest('base64');

    const params = new URLSearchParams({
      apiKey: this.apiKey,
      timestamp,
      sign,
      passphrase: this.passphrase,
      flag: this.flag
    });

    return wss://ws.okx.com:8443/ws/v5/public?${params.toString()};
  }

  connect() {
    return new Promise((resolve, reject) => {
      const wsUrl = this.generateWsUrl();
      console.log('[OKX] WebSocket 연결 시도:', wsUrl.replace(this.apiKey, '***'));

      this.ws = new WebSocket(wsUrl);

      this.ws.on('open', () => {
        console.log('[OKX] WebSocket 연결 성공');
        this.reconnectAttempts = 0;
        this.startPing();
        resolve();
      });

      this.ws.on('message', (data) => {
        try {
          const message = JSON.parse(data.toString());
          this.handleMessage(message);
        } catch (e) {
          console.error('[OKX] 메시지 파싱 오류:', e.message);
        }
      });

      this.ws.on('error', (error) => {
        console.error('[OKX] WebSocket 오류:', error.message);
        reject(error);
      });

      this.ws.on('close', () => {
        console.log('[OKX] WebSocket 연결 종료');
        this.stopPing();
        this.scheduleReconnect();
      });
    });
  }

  // 거래 데이터 구독
  subscribeTrades(instId, handler) {
    const subscribeMsg = {
      op: 'subscribe',
      args: [{
        channel: 'trades',
        instId: instId  // 예: 'BTC-USDT'
      }]
    };

    this.ws.send(JSON.stringify(subscribeMsg));
    this.subscriptions.set(instId, handler);
    console.log([OKX] ${instId} 거래 구독 요청됨);
  }

  handleMessage(message) {
    // 구독 확인 메시지
    if (message.event === 'subscribe') {
      console.log([OKX] 구독 성공: ${message.arg?.instId});
      return;
    }

    // 데이터 메시지
    if (message.data && Array.isArray(message.data)) {
      for (const trade of message.data) {
        const normalized = this.normalizeTrade(trade);
        const handler = this.subscriptions.get(trade.instId);
        if (handler) {
          handler(normalized);
        }
      }
    }
  }

  normalizeTrade(trade) {
    return {
      source: 'okx',
      exchange: 'okx',
      symbol: trade.instId,
      price: parseFloat(trade.px),
      quantity: parseFloat(trade.sz) * (trade.side === 'buy' ? 1 : -1),
      side: trade.side,
      timestamp: parseInt(trade.ts),
      tradeId: trade.tradeId,
      localTimestamp: Date.now()
    };
  }

  startPing() {
    this.pingInterval = setInterval(() => {
      if (this.ws && this.ws.readyState === WebSocket.OPEN) {
        this.ws.ping();
      }
    }, 20000);
  }

  stopPing() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }

  scheduleReconnect() {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error('[OKX] 최대 재연결 횟수 초과');
      return;
    }

    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
    this.reconnectAttempts++;
    console.log([OKX] ${delay}ms 후 재연결 시도 (${this.reconnectAttempts}/${this.maxReconnectAttempts}));

    setTimeout(async () => {
      try {
        await this.connect();
        // 기존 구독 재설정
        for (const instId of this.subscriptions.keys()) {
          this.subscribeTrades(instId, this.subscriptions.get(instId));
        }
      } catch (e) {
        console.error('[OKX] 재연결 실패:', e.message);
      }
    }, delay);
  }

  disconnect() {
    this.stopPing();
    if (this.ws) {
      this.ws.close();
      this.ws = null;
    }
    this.subscriptions.clear();
    console.log('[OKX] 연결 해제됨');
  }
}

module.exports = OKXWebSocketClient;

이중 데이터소스 집계기 구현

// market-data-aggregator.js
const TardisDataSource = require('./tardis-client');
const OKXWebSocketClient = require('./okx-websocket-client');

class MarketDataAggregator {
  constructor(config) {
    this.okxClient = new OKXWebSocketClient(
      config.okxApiKey,
      config.okxSecretKey,
      config.okxPassphrase,
      config.okxFlag
    );
    
    this.tardisClient = new TardisDataSource(config.tardisApiKey);
    
    this.primaryBuffer = new Map();     // OKX 실시간 데이터
    this.secondaryBuffer = new Map();   // Tardis 백업 데이터
    this.unifiedStream = new Map();     // 집계된 데이터
    this.messageHandlers = [];
    
    this.lagThreshold = 5000;           // 5초 이상 지연 시 알림
    this.lastOkxTimestamp = 0;
    this.lastTardisTimestamp = 0;
  }

  async start(symbols) {
    console.log('='.repeat(50));
    console.log('이중 데이터소스 집계기 시작');
    console.log('='.repeat(50));

    // 1단계: OKX 실시간 스트림 연결
    try {
      await this.okxClient.connect();
      for (const symbol of symbols) {
        const instId = symbol.replace('-', '-'); // OKX 형식
        this.okxClient.subscribeTrades(instId, (data) => {
          this.handlePrimaryData(data);
        });
      }
    } catch (error) {
      console.error('[ Aggregator ] OKX 연결 실패, Tardis 활성화:', error.message);
      await this.activateTardisBackup(symbols);
    }

    // 2단계: Tardis 백업/보조 스트림
    await this.activateTardisBackup(symbols);

    // 3단계: 지연 감시 시작
    this.startLagMonitoring();

    console.log('모든 데이터소스 활성화 완료');
  }

  async activateTardisBackup(symbols) {
    for (const symbol of symbols) {
      await this.tardisClient.subscribeExchange('okx', symbol, (data) => {
        this.handleSecondaryData(data);
      });
    }
  }

  handlePrimaryData(data) {
    const key = ${data.exchange}:${data.symbol};
    this.primaryBuffer.set(key, data);
    this.lastOkxTimestamp = data.timestamp;
    
    // 지연 감시
    const delay = Date.now() - data.timestamp;
    if (delay > this.lagThreshold) {
      console.warn([ Aggregator ] ${data.symbol} 지연 감지: ${delay}ms);
    }

    this.emitUnifiedData(key, data);
  }

  handleSecondaryData(data) {
    const key = ${data.exchange}:${data.symbol};
    this.secondaryBuffer.set(key, data);
    this.lastTardisTimestamp = data.timestamp;

    // primary가 없거나 오래된 경우 secondary 사용
    const primary = this.primaryBuffer.get(key);
    if (!primary || data.timestamp > primary.timestamp) {
      this.emitUnifiedData(key, data);
    }
  }

  emitUnifiedData(key, data) {
    this.unifiedStream.set(key, data);
    
    for (const handler of this.messageHandlers) {
      handler({
        ...data,
        primary: this.primaryBuffer.has(key) ? 'okx' : 'tardis',
        timestamp: Date.now()
      });
    }
  }

  startLagMonitoring() {
    setInterval(() => {
      const now = Date.now();
      
      if (this.lastOkxTimestamp > 0) {
        const okxLag = now - this.lastOkxTimestamp;
        if (okxLag > this.lagThreshold) {
          console.warn([ Monitor ] OKX 데이터 지연: ${okxLag}ms);
        }
      }
    }, 10000);
  }

  onMessage(handler) {
    this.messageHandlers.push(handler);
  }

  getUnifiedData(symbol) {
    return this.unifiedStream.get(symbol) || null;
  }

  getStats() {
    return {
      primaryBufferSize: this.primaryBuffer.size,
      secondaryBufferSize: this.secondaryBuffer.size,
      unifiedStreamSize: this.unifiedStream.size,
      lastOkxTimestamp: this.lastOkxTimestamp,
      lastTardisTimestamp: this.lastTardisTimestamp,
      timestampDiff: this.lastOkxTimestamp - this.lastTardisTimestamp
    };
  }

  stop() {
    console.log('집계기 종료 중...');
    this.okxClient.disconnect();
    this.tardisClient.disconnect();
    this.primaryBuffer.clear();
    this.secondaryBuffer.clear();
    this.unifiedStream.clear();
    console.log('집계기 종료 완료');
  }
}

module.exports = MarketDataAggregator;

메인 실행 파일

// index.js
require('dotenv').config();
const MarketDataAggregator = require('./market-data-aggregator');

async function main() {
  const symbols = ['BTC-USDT', 'ETH-USDT', 'SOL-USDT'];
  
  const aggregator = new MarketDataAggregator({
    okxApiKey: process.env.OKX_API_KEY,
    okxSecretKey: process.env.OKX_SECRET_KEY,
    okxPassphrase: process.env.OKX_PASSPHRASE,
    okxFlag: process.env.OKX_FLAG || '0',
    tardisApiKey: process.env.TARDIS_API_KEY
  });

  // 실시간 데이터 처리 핸들러
  aggregator.onMessage((data) => {
    const time = new Date(data.timestamp).toISOString();
    console.log([${time}] ${data.source.padEnd(5)} | ${data.symbol.padEnd(10)} | ${data.price} | Qty: ${Math.abs(data.quantity)} | ${data.side.toUpperCase()});
  });

  // 통계 출력 (10초마다)
  setInterval(() => {
    const stats = aggregator.getStats();
    console.log('\n--- 현재 통계 ---');
    console.log(Primary 버퍼: ${stats.primaryBufferSize});
    console.log(Secondary 버퍼: ${stats.secondaryBufferSize});
    console.log(`타이밍 차이: ${stats.timestampDiff}ms');
    console.log('------------------\n');
  }, 10000);

  try {
    await aggregator.start(symbols);
    console.log('데이터 수집 중... (Ctrl+C로 종료)\n');
  } catch (error) {
    console.error('시작 실패:', error);
    process.exit(1);
  }

  // Graceful shutdown
  process.on('SIGINT', () => {
    console.log('\n\n종료 시그널 수신...');
    aggregator.stop();
    process.exit(0);
  });
}

main();

성능 최적화 팁

// 고성능 집계기 (배치 처리 최적화)
class OptimizedAggregator {
  constructor(options = {}) {
    this.batchSize = options.batchSize || 100;
    this.flushInterval = options.flushInterval || 100;
    this.batchBuffer = [];
    this.lastFlush = Date.now();
  }

  // 배치 버퍼링으로 처리량 향상
  addToBatch(data) {
    this.batchBuffer.push(data);
    
    if (this.batchBuffer.length >= this.batchSize || 
        Date.now() - this.lastFlush >= this.flushInterval) {
      this.flushBatch();
    }
  }

  flushBatch() {
    if (this.batchBuffer.length === 0) return;

    // 배치 처리 (예: DB 벌크 인서트, 분석 시스템 전달)
    const batch = this.batchBuffer.splice(0, this.batchBuffer.length);
    this.lastFlush = Date.now();
    
    return batch;
  }

  // 압축 처리 (고빈도 데이터용)
  compressTrades(trades) {
    const priceMap = new Map();
    
    for (const trade of trades) {
      const key = trade.symbol;
      const existing = priceMap.get(key);
      
      if (existing) {
        // 같은 심볼 내 마지막 거래만 유지 (가격만 업데이트)
        existing.price = trade.price;
        existing.quantity += Math.abs(trade.quantity);
        existing.timestamp = trade.timestamp;
      } else {
        priceMap.set(key, { ...trade });
      }
    }
    
    return Array.from(priceMap.values());
  }
}

자주 발생하는 오류와 해결책

오류 유형	원인	해결 방법
WebSocket connection failed	OKX 서버 연결 거부, 방화벽, 네트워크 문제	재연결 로직 구현, 프록시 사용, 연결 타임아웃 설정
Tardis API rate limit exceeded	API 요청 과다, 구독 제한 초과	요청 간 딜레이 추가, 프리미엄 플랜 업그레이드, 캐싱 활용
Data timestamp out of order	네트워크 지연으로 인한 순서 역전	타임스탬프 기반 정렬 버퍼 구현, 지연 데이터 필터링
Invalid signature for OKX API	비밀키 불일치, 타임스탬프 오차	서명 알고리즘 검증, 서버 시간 동기화 (NTP)
Memory leak in WebSocket	버퍼 미청소, 이벤트 리스너 누수	주기적 버퍼 플러시, 명시적 리스너 제거, maxBufferSize 설정

추가 최적화 및 모니터링

// 헬스체크 및 메트릭스 수집
class MetricsCollector {
  constructor() {
    this.metrics = {
      messagesReceived: 0,
      messagesBySource: { okx: 0, tardis: 0 },
      errors: [],
      latencies: [],
      lastMessageTime: null
    };
  }

  recordMessage(source, latency) {
    this.metrics.messagesReceived++;
    this.metrics.messagesBySource[source]++;
    this.metrics.latencies.push(latency);
    this.metrics.lastMessageTime = Date.now();

    // 1000개 샘플만 유지
    if (this.metrics.latencies.length > 1000) {
      this.metrics.latencies.shift();
    }
  }

  recordError(error) {
    this.metrics.errors.push({
      timestamp: Date.now(),
      message: error.message,
      stack: error.stack
    });

    // 최근 100개만 유지
    if (this.metrics.errors.length > 100) {
      this.metrics.errors.shift();
    }
  }

  getReport() {
    const avgLatency = this.metrics.latencies.reduce((a, b) => a + b, 0) / 
                       this.metrics.latencies.length || 0;
    
    return {
      totalMessages: this.metrics.messagesReceived,
      bySource: this.metrics.messagesBySource,
      avgLatency: avgLatency.toFixed(2) + 'ms',
      errorCount: this.metrics.errors.length,
      uptime: this.metrics.lastMessageTime ? 
              Date.now() - this.metrics.lastMessageTime : 'N/A'
    };
  }
}

결론

Tardis.dev와 OKX API를 결합한 이중 데이터소스 아키텍처는 단일 소스 의존도를 낮추고 데이터 가용성을 극대화한다. OKX를 primary 실시간 스트림으로 사용하고 Tardis를 backup 및 historical 데이터 소스로 활용하면 네트워크 단절이나 API 장애 시에도 안정적인 시장 데이터 수집이 가능하다. 실제 거래 시스템에서는 이 집계기에 **_FAILOVER 메커니즘_, 중복 데이터 제거, 배치 처리 최적화**를 반드시 구현해야 하며, 메모리 관리와 연결 상태 모니터링도 필수적이다.