Розробка системи нормалізації даних з бірж

Розробка системи нормалізації даних з бірж

Кожна крипто-біржа — окремий всесвіт з власними угодами про найменування, форматами чисел, одиницями часу та семантикою полів. BTC/USDT на Binance називається BTCUSDT, на Kraken — XBT/USDT, на Bitfinex — tBTCUST. Нормалізація — це шар, який приховує цю несумісність за єдиним інтерфейсом.

Що потрібно нормалізувати

Символи та пари. Кожна біржа має власні угоди. Нормалізований формат — BASE/QUOTE у верхньому регістрі: BTC/USDT, ETH/BTC. Біржові символи зберігаються в маппінгу з можливістю зворотного перетворення.

Timestamps. Binance повертає мілісекунди, деякі біржи — секунди, OKX — наносекунди. Нормалізований формат — мілісекунди UTC, що зберігаються як int64.

Числа. REST API часто повертає числа як рядки ("43250.50"), деякі біржи втрачають trailing zeros. Нормалізований формат — Decimal з явною точністю, залежно від інструменту.

Сторони ордера. BUY/SELL, buy/sell, b/s, 1/-1 — все це зустрічається. Нормалізований формат — enum BUY | SELL.

Статуси ордерів. У кожної біржі власні статуси. Нормалізований маппінг:

Архітектура нормалізатора

Нормалізатор реалізується як набір адаптерів специфічних для бірж з загальним інтерфейсом:

from abc import ABC, abstractmethod
from decimal import Decimal

class ExchangeNormalizer(ABC):
    @abstractmethod
    def normalize_symbol(self, raw_symbol: str) -> str:
        """Перетворює біржевий символ у нормалізований формат BASE/QUOTE"""

    @abstractmethod
    def normalize_ticker(self, raw_data: dict) -> NormalizedTicker:
        """Нормалізує ticker дані"""

    @abstractmethod
    def normalize_order(self, raw_data: dict) -> NormalizedOrder:
        """Нормалізує дані ордера"""


class BinanceNormalizer(ExchangeNormalizer):
    SYMBOL_MAP = {
        "BTCUSDT": "BTC/USDT",
        "ETHUSDT": "ETH/USDT",
        # ... з API /api/v3/exchangeInfo
    }

    def normalize_ticker(self, raw: dict) -> NormalizedTicker:
        return NormalizedTicker(
            exchange="binance",
            symbol=self.normalize_symbol(raw["s"]),
            timestamp=int(raw["T"]),
            price=Decimal(raw["c"]),
            volume_24h=Decimal(raw["v"]),
        )

Динамічне завантаження маппінгу символів

Жорсткий маппінг символів у коді — погана ідея: біржи додають нові пари щодня. Правильний підхід — завантажити маппінг з Exchange Info API при запуску та періодично оновлювати:

async def load_symbol_map(self):
    exchange_info = await self.rest_client.get("/api/v3/exchangeInfo")
    self.symbol_map = {
        s["symbol"]: f"{s['baseAsset']}/{s['quoteAsset']}"
        for s in exchange_info["symbols"]
        if s["status"] == "TRADING"
    }
    # Інвертований маппінг для зворотного перетворення
    self.reverse_map = {v: k for k, v in self.symbol_map.items()}

Валідація нормалізованих даних

Після нормалізації важливо валідувати результат. Негативні ціни, нульові обсяги, timestamp у майбутньому — все це ознаки проблем з джерелом даних:

def validate_ticker(ticker: NormalizedTicker) -> list[str]:
    errors = []
    if ticker.price <= 0:
        errors.append(f"Invalid price: {ticker.price}")
    if ticker.timestamp > now_ms() + 5000:
        errors.append(f"Future timestamp: {ticker.timestamp}")
    if ticker.bid and ticker.ask and ticker.bid >= ticker.ask:
        errors.append(f"Crossed book: bid={ticker.bid} ask={ticker.ask}")
    return errors

Невалідні дані логуються та відкидаються, не потрапляючи в downstream-системи.

Тестування нормалізатора

Unit-тести з реальними прикладами raw-даних від кожної біржі — обов'язкові. Біржи іноді змінюють формат API без попередження. Набір фіксованих fixtures з очікуваними нормалізованими результатами дозволяє швидко виявити регресію:

def test_binance_normalizer():
    raw = {"s": "BTCUSDT", "c": "43250.50", "v": "28450.12", "T": 1704067200000}
    result = BinanceNormalizer().normalize_ticker(raw)
    assert result.symbol == "BTC/USDT"
    assert result.price == Decimal("43250.50")
    assert result.exchange == "binance"

Крім того — integration тести з live API біржей у sandbox-режимі, запускаються щодня в CI для раннього виявлення змін у API.