פיתוח מערכת נורמליזציה של נתוני מסחר

אחד הלקוחות שלנו, קרן גידור, ניהל תיק השקעות על פני 10 בורסות ובילה שלושה ימים בשבוע בתיאום ידני של טיקרים שונים. לאחר יישום מערכת נורמליזציה, הזמן הזה ירד לשעה אחת. אנו בונים מערכות כאלה כבר למעלה מעשור, תוך אינטגרציה עם 20+ בורסות — מ-Binance ועד פרוטוקולים מבוזרים. ללא נורמליזציה, אתה מקבל נתונים מפוזרים שחסרי תועלת למסחר, אנליטיקה או בדיקות חוזרות (backtesting). אנו פותרים זאת על ידי יצירת מודל נתונים אחיד שמסתיר את כל הניואנסים הספציפיים לבורסה מאחורי ממשק יחיד.

אבל הבעיה עמוקה יותר ממה שהיא נראית. גם לאחר נורמליזציה של סמלים וחותמות זמן, נותרות שאלות: כיצד לטפל בשגיאות API, לאמת נתונים ולהתרחב בעת הוספת בורסות חדשות? במאמר זה, אנו חולקים פתרונות קונקרטיים שבהם אנו משתמשים בפרויקטים מסחריים.

מה צריך לנרמל

סמלים וזוגות. לכל בורסה יש מוסכמות משלה. פורמט מנורמל: BASE/QUOTE באותיות גדולות — BTC/USDT, ETH/BTC. סמלי בורסה מאוחסנים במיפוי עם תמיכה בהמרה הפוכה.

חותמות זמן. Binance מחזירה מילישניות, חלק מהבורסות מחזירות שניות, OKX מחזירה ננו-שניות. פורמט מנורמל: UTC במילישניות המאוחסן כ-int64.

מספרים. ממשקי REST מחזירים לעיתים קרובות מספרים כמחרוזות (\"43250.50\"), חלק מהבורסות משמיטות אפסים סופיים. פורמט מנורמל: BUY עם דיוק מפורש בהתאם למכשיר.

צדי הזמנה. SELL/buy, sell/b, s/1, -1/BUY — כולם קיימים. פורמט מנורמל: enum SELL | NEW. סטטוסי הזמנה. לכל בורסה יש סט משלה. מיפוי מנורמל:

כיצד אנו ניגשים לנורמליזציה

אנו בונים את המנרמל כסט של מתאמים ספציפיים לבורסה החולקים ממשק משותף. זה מאפשר הוספת בורסות חדשות ללא שינוי קוד קיים. אנו משתמשים ב-Python אסינכרוני ו-pydantic לאימות סכמת קלט קפדנית.

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()}

אנו בודקים באופן קבוע עדכונים דרך תיעוד ה-API של Binance כדי לשמור על המיפוי עדכני.

אימות נתונים מנורמלים

לאחר הנורמליזציה, חיוני לאמת את הפלט. מחירים שליליים, נפחים אפסיים, חותמות זמן בעתיד — כל אלה הם סימנים לבעיות בנתוני המקור:

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

נתונים לא חוקיים מתועדים ונזרקים, ולעולם אינם מגיעים למערכות במורד הזרם. זה מבטיח שהאלגוריתמים שלך מקבלים תמיד נתונים נכונים.

מדוע נורמליזציה קריטית לפרויקט שלך

נורמליזציה לקויה מובילה לתוצאות בדיקות חוזרות שגויות, להזמנות שגויות ולאובדן כסף. הגישה שלנו מפחיתה שגיאות נתונים ב-80% בהשוואה לפתרונות אד-הוק. הארכיטקטורה האסינכרונית מעבדת עד 1000 טיקרים בשנייה על שרת יחיד — פי 3 מהר יותר מיישומי Python סינכרוניים טיפוסיים. החיסכון בתחזוקה מפורמט אחיד מגיע ל-50%.

כיצד אנו מבטיחים דיוק נורמליזציה

בדיקות יחידה עם דגימות נתונים גולמיים אמיתיים מכל בורסה הן חובה. בורסות לפעמים משנות את פורמט ה-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"

בנוסף, אנו מריצים בדיקות אינטגרציה מול סביבות sandbox חיות של בורסות מדי יום ב-CI כדי לתפוס שינויי API מוקדם.

רשימת צ'ק-ליסט לשלבי נורמליזציה

• ביקורת ממשקי API של בורסות: תיעוד, מגבלות קצב, פורמטים.
• עיצוב סכמת נתונים מנורמלת.
• יישום מתאמים לכל בורסה.
• כתיבת בדיקות יחידה ואינטגרציה.
• יצירת תיעוד אינטגרציה.
• תמיכה לחודש אחד לאחר המסירה: חידודים, ייעוץ.

מדריך שלב-אחר-שלב להוספת מתאם חדש

1. צור מחלקה היורשת מ-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()} .
2. יישם את 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 , 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" , ExchangeNormalizer.
3. כתוב בדיקות יחידה עם דגימות נתונים גולמיים.
4. רשום את המתאם במפעל המנרמל.
5. בדוק אינטגרציה על בורסות sandbox.
6. פרוס לייצור עם ניטור שגיאות.

תזמון ועלות

לוחות הזמנים נעים בין 2 ל-4 שבועות לכל בורסה; עבור פרויקט מורכב עם 5+ בורסות, 4 עד 8 שבועות. העלות מחושבת באופן אישי לאחר ניתוח הדרישות שלך. להערכה מדויקת, מלא טופס קצר — נשלח הצעה עם שלבים ולוחות זמנים.

הזמן פיתוח של מערכת נורמליזציה המותאמת לצרכים שלך. קבל ייעוץ מהמהנדס שלנו כבר עכשיו — נשיב תוך יום עסקים אחד.