Tardis WebSocket API: คู่มือการกำหนดค่าการสมัครรับข้อมูลแบบเรียลไทม์

บทนำและภาพรวมของ Tardis WebSocket

สถาปัตยกรรมและหลักการทำงานเบื้องหลัง

การเชื่อมต่อและการยืนยันตัวตน

const WebSocket = require('ws');

class TardisWebSocketClient {
    constructor(apiKey, options = {}) {
        this.apiKey = apiKey;
        this.ws = null;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = options.maxReconnectAttempts || 10;
        this.reconnectDelay = options.reconnectDelay || 1000;
        this.subscriptions = new Map();
        this.messageHandlers = new Map();
    }

    connect() {
        return new Promise((resolve, reject) => {
            // Tardis WebSocket endpoint
            const url = 'wss://api.tardis.dev/v1/ws';
            
            this.ws = new WebSocket(url, {
                headers: {
                    'X-API-Key': this.apiKey
                }
            });

            this.ws.on('open', () => {
                console.log('[Tardis] WebSocket connected');
                this.reconnectAttempts = 0;
                // Resubscribe to previous subscriptions after reconnect
                this.resubscribeAll();
                resolve();
            });

            this.ws.on('message', (data) => this.handleMessage(data));

            this.ws.on('close', (code, reason) => {
                console.log([Tardis] Connection closed: ${code} - ${reason});
                this.scheduleReconnect();
            });

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

    scheduleReconnect() {
        if (this.reconnectAttempts >= this.maxReconnectAttempts) {
            console.error('[Tardis] Max reconnection attempts reached');
            return;
        }

        const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts);
        console.log([Tardis] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts + 1}));
        
        setTimeout(() => {
            this.reconnectAttempts++;
            this.connect().catch(err => {
                console.error('[Tardis] Reconnection failed:', err.message);
            });
        }, delay);
    }

    handleMessage(data) {
        try {
            const message = JSON.parse(data);
            
            // Route message to appropriate handler
            const handler = this.messageHandlers.get(message.type);
            if (handler) {
                handler(message);
            }
        } catch (error) {
            console.error('[Tardis] Failed to parse message:', error);
        }
    }

    subscribe(channel, symbol, handler) {
        const subscriptionKey = ${channel}:${symbol};
        
        const subscription = {
            channel,
            symbol,
            exchange: this.getExchangeFromSymbol(symbol)
        };

        const subscribeMessage = {
            type: 'subscribe',
            ...subscription
        };

        this.ws.send(JSON.stringify(subscribeMessage));
        this.subscriptions.set(subscriptionKey, subscription);
        this.messageHandlers.set(message.type, handler);
        
        console.log([Tardis] Subscribed to ${subscriptionKey});
    }

    unsubscribe(channel, symbol) {
        const subscriptionKey = ${channel}:${symbol};
        
        const unsubscribeMessage = {
            type: 'unsubscribe',
            channel,
            symbol
        };

        this.ws.send(JSON.stringify(unsubscribeMessage));
        this.subscriptions.delete(subscriptionKey);
        
        console.log([Tardis] Unsubscribed from ${subscriptionKey});
    }

    resubscribeAll() {
        for (const [key, subscription] of this.subscriptions) {
            const subscribeMessage = {
                type: 'subscribe',
                ...subscription
            };
            this.ws.send(JSON.stringify(subscribeMessage));
            console.log([Tardis] Resubscribed to ${key});
        }
    }

    getExchangeFromSymbol(symbol) {
        // Parse exchange from symbol format (e.g., "binance:btc-usdt")
        const parts = symbol.split(':');
        return parts.length > 1 ? parts[0] : 'unknown';
    }

    disconnect() {
        if (this.ws) {
            this.ws.close(1000, 'Client initiated close');
            this.ws = null;
        }
    }
}

module.exports = TardisWebSocketClient;

รูปแบบข้อมูลและการจัดการ Message

class MarketDataProcessor {
    constructor() {
        this.orderBooks = new Map();
        this.tradeBuffer = [];
        this.tickerData = new Map();
    }

    processOrderBookUpdate(message) {
        const { exchange, symbol, bids, asks, timestamp, localTimestamp } = message.data;
        const bookKey = ${exchange}:${symbol};

        // Initialize or update order book
        if (!this.orderBooks.has(bookKey)) {
            this.orderBooks.set(bookKey, { bids: new Map(), asks: new Map() });
        }

        const book = this.orderBooks.get(bookKey);

        // Apply incremental updates
        // bids and asks are arrays of [price, quantity] pairs
        for (const [price, quantity] of message.data.bids) {
            if (quantity === 0) {
                book.bids.delete(price);
            } else {
                book.bids.set(parseFloat(price), parseFloat(quantity));
            }
        }

        for (const [price, quantity] of message.data.asks) {
            if (quantity === 0) {
                book.asks.delete(price);
            } else {
                book.asks.set(parseFloat(price), parseFloat(quantity));
            }
        }

        // Calculate spread
        const bestBid = Math.max(...book.bids.keys());
        const bestAsk = Math.min(...book.asks.keys());
        const spread = bestAsk - bestBid;
        const spreadPercent = (spread / bestAsk) * 100;

        // Measure latency
        const latency = localTimestamp - timestamp;

        return {
            symbol: bookKey,
            bestBid,
            bestAsk,
            spread,
            spreadPercent,
            latencyMs: latency,
            timestamp
        };
    }

    processTrade(message) {
        const { id, exchange, symbol, side, price, quantity, timestamp, localTimestamp } = message.data;
        
        const trade = {
            id,
            exchange,
            symbol,
            side, // 'buy' or 'sell'
            price: parseFloat(price),
            quantity: parseFloat(quantity),
            value: parseFloat(price) * parseFloat(quantity),
            timestamp,
            localTimestamp,
            latencyMs: localTimestamp - timestamp
        };

        // Add to buffer for batch processing
        this.tradeBuffer.push(trade);

        return trade;
    }

    processTicker(message) {
        const { exchange, symbol, last, high, low, volume, timestamp, localTimestamp } = message.data;
        
        const ticker = {
            exchange,
            symbol,
            last: parseFloat(last),
            high: parseFloat(high),
            low: parseFloat(low),
            volume: parseFloat(volume),
            change: ((parseFloat(last) - parseFloat(high)) / parseFloat(high)) * 100,
            timestamp,
            latencyMs: localTimestamp - timestamp
        };

        this.tickerData.set(${exchange}:${symbol}, ticker);

        return ticker;
    }

    getOrderBook(exchange, symbol) {
        const bookKey = ${exchange}:${symbol};
        return this.orderBooks.get(bookKey);
    }

    getTopOfBook(exchange, symbol) {
        const book = this.getOrderBook(exchange, symbol);
        if (!book) return null;

        const sortedBids = Array.from(book.bids.entries()).sort((a, b) => b[0] - a[0]);
        const sortedAsks = Array.from(book.asks.entries()).sort((a, b) => a[0] - b[0]);

        return {
            bestBid: { price: sortedBids[0]?.[0], quantity: sortedBids[0]?.[1] },
            bestAsk: { price: sortedAsks[0]?.[0], quantity: sortedAsks[0]?.[1] },
            spread: sortedAsks[0]?.[0] - sortedBids[0]?.[0]
        };
    }

    flushTradeBuffer() {
        const trades = [...this.tradeBuffer];
        this.tradeBuffer = [];
        return trades;
    }

    getStatistics() {
        let totalLatency = 0;
        let count = 0;

        for (const trade of this.tradeBuffer) {
            totalLatency += trade.latencyMs;
            count++;
        }

        return {
            bufferedTrades: count,
            averageLatencyMs: count > 0 ? totalLatency / count : 0,
            orderBooks: this.orderBooks.size
        };
    }
}

module.exports = MarketDataProcessor;

การควบคุม Concurrency และ Rate Limiting

class RateLimitedClient {
    constructor(client, options = {}) {
        this.client = client;
        this.messagesPerSecond = options.messagesPerSecond || 100;
        this.burstSize = options.burstSize || 20;
        this.messageQueue = [];
        this.isProcessing = false;
        this.lastResetTime = Date.now();
        this.messageCount = 0;
        this.tokens = this.burstSize;
        this.refillRate = this.messagesPerSecond / 1000; // tokens per ms
    }

    async subscribe(channel, symbol, handler) {
        await this.acquireToken();
        return this.client.subscribe(channel, symbol, handler);
    }

    async unsubscribe(channel, symbol) {
        await this.acquireToken();
        return this.client.unsubscribe(channel, symbol);
    }

    async acquireToken() {
        return new Promise((resolve) => {
            const tryAcquire = () => {
                this.refillTokens();
                
                if (this.tokens >= 1) {
                    this.tokens -= 1;
                    this.messageCount++;
                    resolve();
                } else {
                    // Calculate wait time
                    const waitTime = (1 - this.tokens) / this.refillRate;
                    setTimeout(tryAcquire, Math.max(waitTime, 1));
                }
            };

            tryAcquire();
        });
    }

    refillTokens() {
        const now = Date.now();
        const elapsed = now - this.lastResetTime;
        const newTokens = elapsed * this.refillRate;
        
        this.tokens = Math.min(this.burstSize, this.tokens + newTokens);
        this.lastResetTime = now;
    }

    getStats() {
        return {
            queuedMessages: this.messageQueue.length,
            totalMessages: this.messageCount,
            currentTokens: this.tokens,
            messagesPerSecond: this.messagesPerSecond
        };
    }
}

class MultiExchangeManager {
    constructor(options = {}) {
        this.exchanges = new Map();
        this.processors = new Map();
        this.aggregatedOrderBook = new Map();
        this.maxLatency = options.maxLatencyMs || 1000;
        this.latencyMonitor = new LatencyMonitor(this.maxLatency);
    }

    addExchange(exchange, apiKey, options = {}) {
        const client = new TardisWebSocketClient(apiKey, options);
        const processor = new MarketDataProcessor();
        const rateLimitedClient = new RateLimitedClient(client, options.rateLimit);

        this.exchanges.set(exchange, {
            client: rateLimitedClient,
            rawClient: client,
            processor
        });
        this.processors.set(exchange, processor);

        return rateLimitedClient;
    }

    async connectAll() {
        const connections = [];
        
        for (const [exchange, { client }] of this.exchanges) {
            try {
                await client.connect();
                connections.push({ exchange, success: true });
                console.log([Manager] Connected to ${exchange});
            } catch (error) {
                connections.push({ exchange, success: false, error: error.message });
                console.error([Manager] Failed to connect to ${exchange}:, error.message);
            }
        }

        return connections;
    }

    subscribeToAll(channel, symbols, handler) {
        const results = [];
        
        for (const [exchange, { processor }] of this.exchanges) {
            for (const symbol of symbols) {
                const fullSymbol = ${exchange}:${symbol};
                const wrappedHandler = (message) => {
                    // Wrap handler to add latency monitoring
                    const startTime = Date.now();
                    const result = handler(message, exchange, symbol);
                    const processingTime = Date.now() - startTime;
                    
                    this.latencyMonitor.record(processingTime, exchange);
                    
                    return result;
                };

                try {
                    this.exchanges.get(exchange).client.subscribe(channel, fullSymbol, wrappedHandler);
                    results.push({ exchange, symbol, success: true });
                } catch (error) {
                    results.push({ exchange, symbol, success: false, error: error.message });
                }
            }
        }

        return results;
    }

    getAggregatedDepth(exchange, symbol, levels = 10) {
        const processor = this.processors.get(exchange);
        if (!processor) return null;

        const book = processor.getOrderBook(exchange, symbol);
        if (!book) return null;

        const sortedBids = Array.from(book.bids.entries())
            .sort((a, b) => b[0] - a[0])
            .slice(0, levels);
        
        const sortedAsks = Array.from(book.asks.entries())
            .sort((a, b) => a[0] - b[0])
            .slice(0, levels);

        return {
            bids: sortedBids.map(([price, qty]) => ({ price, quantity: qty })),
            asks: sortedAsks.map(([price, qty]) => ({ price, quantity: qty })),
            timestamp: Date.now()
        };
    }

    async disconnectAll() {
        for (const [exchange, { rawClient }] of this.exchanges) {
            rawClient.disconnect();
            console.log([Manager] Disconnected from ${exchange});
        }
        this.exchanges.clear();
        this.processors.clear();
    }
}

class LatencyMonitor {
    constructor(maxLatency) {
        this.maxLatency = maxLatency;
        this.measurements = new Map();
    }

    record(latencyMs, exchange) {
        if (!this.measurements.has(exchange)) {
            this.measurements.set(exchange, []);
        }
        
        const measurements = this.measurements.get(exchange);
        measurements.push(latencyMs);
        
        // Keep only last 100 measurements
        if (measurements.length > 100) {
            measurements.shift();
        }

        // Alert if latency exceeds threshold
        if (latencyMs > this.maxLatency) {
            console.warn([LatencyMonitor] High latency on ${exchange}: ${latencyMs}ms);
        }
    }

    getStats(exchange) {
        const measurements = this.measurements.get(exchange) || [];
        if (measurements.length === 0) return null;

        const sorted = [...measurements].sort((a, b) => a - b);
        const sum = sorted.reduce((a, b) => a + b, 0);

        return {
            count: sorted.length,
            min: sorted[0],
            max: sorted[sorted.length - 1],
            avg: sum / sorted.length,
            p50: sorted[Math.floor(sorted.length * 0.5)],
            p95: sorted[Math.floor(sorted.length * 0.95)],
            p99: sorted[Math.floor(sorted.length * 0.99)]
        };
    }
}

module.exports = { RateLimitedClient, MultiExchangeManager, LatencyMonitor };

Best Practices สำหรับ Production Environment

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

// แก้ไข: เพิ่ม error handling ที่ละเอียด
this.ws.on('close', (code, reason) => {
    console.error([Tardis] Connection closed: code=${code}, reason=${reason});
    
    // Handle specific close codes
    if (code === 1006) {
        console.warn('[Tardis] Abnormal closure - possible rate limit or auth issue');
        // Wait longer before reconnect
        this.scheduleReconnect(30000);
    } else if (code === 1000) {
        console.log('[Tardis] Normal closure');
    } else {
        this.scheduleReconnect();
    }
});

// แก้ไข: เก็บ subscriptions ใน persistent storage
const subscriptions = new Map(); // in-memory

// เพิ่ม method สำหรับ persist subscriptions
async saveSubscription(channel, symbol) {
    const key = ${channel}:${symbol};
    subscriptions.set(key, {
        channel,
        symbol,
        subscribedAt: Date.now()
    });
    // Optional: persist to Redis or database for crash recovery
    await this.persistToStorage(key, { channel, symbol });
}

async loadAndResubscribe() {
    // Load from storage
    const saved = await this.loadFromStorage();
    
    for (const [key, data] of Object.entries(saved)) {
        // Check if subscription is still valid (not expired)
        const age = Date.now() - data.subscribedAt;
        if (age < this.maxSubscriptionAge) {
            await this.sendSubscription(data.channel, data.symbol);
            console.log([Tardis] Resubscribed to ${key});
        }
    }
}