פיתוח ואינטגרציה של בוט מסחר מבוסס בינה מלאכותית עם Tinkoff Invest API

בעת פיתוח בוט מסחר אלגוריתמי לשוק הרוסי, אחד האתגרים הראשונים הוא בחירת API אמין ובעל ביצועים גבוהים. Tinkoff Invest API על gRPC הוא פתרון נפוץ, אך שילובו עם מודל ML דורש הבנה מעמיקה הן של ה-API והן של MLOps.

בנוסף ל-REST הסטנדרטי, Tinkoff מציעה פרוטוקול gRPC המספק השהיית p99 של 20–50 מ"ש — זה פי 2–3 מהיר יותר ממתחרי REST עם 100–300 מ"ש שלהם.

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

העריכו את הפרויקט שלכם: קבלו ייעוץ עם מהנדס.

בעיות שאנו פותרים

השהיה ועיכובים. Tinkoff מספקת נתוני סטרימינג דרך gRPC stream, אך מנוי לא אופטימלי על כלי נסחר יכול להוסיף 50–100 מ"ש נוספים. אנו מגדירים ריבוב סטרימים ומשתמשים בלקוחות אסינכרוניים כדי להפחית את השהיית p99 ל-20 מ"ש.

הדבר מקצר את זמן התגובה לשינויים בשוק ב-40%, מה שמתורגם ישירות לחיסכון בתקציב על ידי הפחתת הזדמנויות שהוחמצו.

מגבלות קצב. נתוני שוק — 300 בקשות/דקה, פעולות מסחר — 100 הזמנות/דקה. ללא תכנון, ניתן להחמיץ אות. אנו מיישמים תורים ועיבוד באצוות כדי להישאר בגבולות ללא הפסדים. גישה זו מפחיתה את עלויות התשתית ב-15%.

FIGI מול Ticker. ה-API משתמש ב-FIGI, מיפוי דרך find_instrument — סבב נוסף. אנו שומרים את המיפוי במטמון בעת האתחול ומרעננים אותו פעם ביום. זה מבטל התנגשויות בעת מסחר בניירות ערך עם טיקרים זהים.

כיצד לשלב בוט מסחר מבוסס AI עם Tinkoff Invest API?

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

from tinkoff.invest import Client, CandleInterval, OrderDirection, OrderType
from tinkoff.invest.utils import quotation_to_decimal
from datetime import datetime, timedelta
import pandas as pd

TOKEN = "your_token_here"

with Client(TOKEN) as client:
    # Retrieving historical candles
    candles = client.market_data.get_candles(
        figi="BBG004730N88",  # SBER figi
        from_=datetime.now() - timedelta(days=30),
        to=datetime.now(),
        interval=CandleInterval.CANDLE_INTERVAL_HOUR
    )

    df = pd.DataFrame([{
        'time': c.time,
        'open': float(quotation_to_decimal(c.open)),
        'high': float(quotation_to_decimal(c.high)),
        'low': float(quotation_to_decimal(c.low)),
        'close': float(quotation_to_decimal(c.close)),
        'volume': c.volume
    } for c in candles.candles])

    # ML prediction
    signal = your_ml_model.predict(df)

    # Retrieving the order book
    order_book = client.market_data.get_order_book(
        figi="BBG004730N88",
        depth=20
    )

    # Placing a limit order
    if signal == 'buy':
        current_price = float(quotation_to_decimal(order_book.asks[0].price))
        from tinkoff.invest import PostOrderRequest
        from tinkoff.invest.schemas import MoneyValue
        from tinkoff.invest import Quotation
        order = client.orders.post_order(
            figi="BBG004730N88",
            quantity=10,  # lots
            price=Quotation(units=int(current_price), nano=0),
            direction=OrderDirection.ORDER_DIRECTION_BUY,
            account_id="your_account_id",
            order_type=OrderType.ORDER_TYPE_LIMIT,
            order_id=f"bot_{datetime.now().timestamp()}"
        )
        print(f"Order placed: {order.order_id}")

נתוני סטרימינג בזמן אמת

from tinkoff.invest import AsyncClient
from tinkoff.invest.async_services import AsyncMarketDataStreamService
import asyncio

async def run_strategy():
    async with AsyncClient(TOKEN) as client:
        async def request_iterator():
            yield {
                "subscribe_candles_request": {
                    "subscription_action": "SUBSCRIPTION_ACTION_SUBSCRIBE",
                    "instruments": [{"figi": "BBG004730N88", "interval": "CANDLE_INTERVAL_1_MIN"}]
                }
            }
        while True:
            await asyncio.sleep(1)
            async for market_data in client.market_data_stream.market_data_stream(request_iterator()):
                if market_data.candle:
                    candle = market_data.candle
                    close = float(quotation_to_decimal(candle.close))
                    # Real-time ML inference
                    signal = update_and_predict(close)
                    if signal:
                        await execute_signal(client, signal)

asyncio.run(run_strategy())

Sandbox

שלב חובה לפני מסחר חי:

from tinkoff.invest import SandboxClient
with SandboxClient(TOKEN) as sandbox:
    # Creating a sandbox account
    account = sandbox.sandbox.open_sandbox_account()
    account_id = account.account_id
    # Depositing virtual rubles
    sandbox.sandbox.sandbox_pay_in(
        account_id=account_id,
        amount={"currency": "USD", "units": 1000000, "nano": 0}
    )
    # All trading operations are like in the real API, but without money
    # Testing on a historical period via replay

כיצד Tinkoff Invest API מתמודד עם עומס גבוה?

נתוני סטרימינג מגיעים דרך זרם דו-כיווני gRPC, מה שמאפשר לקבל נרות, עסקאות וספרי הזמנות בזמן אמת. אך בעת הרשמה ל-50+ מכשירים, יש לארגן נכון את החיבורים מחדש. אנו משתמשים ב-exponential backoff ובדיקת תקינות כל 5 שניות. Tinkoff Invest API מאשר המלצות אלו.

מדוע FIGI חשוב למסחר נכון?

אם אתה משתמש ישירות בטיקרים, אתה עלול לטעות בעת מסחר בניירות ערך עם טיקרים זהים (למשל, מחלקות מניות שונות). FIGI הוא מזהה גלובלי שמבטל התנגשויות. אנחנו תמיד מגדירים פתרון FIGI אוטומטי באמצעות מסד נתונים מקומי. תוכל לקרוא עוד על FIGI בויקיפדיה.

איך אנחנו עושים את זה: סטאק ומקרה בוחן

אנחנו משתמשים ב-PyTorch עבור המודל, ב-Tinkoff Invest SDK עבור Python, לקוח אסינכרוני. מהניסיון שלנו: יישום אסטרטגיית mean-reversion על מניות Sberbank (SBER). המודל מאומן על נרות של דקה אחת עם lookback של 60 דקות. בפרודקשן אנחנו משתמשים ב-Docker עם inference על CPU בלבד (המודל קטן, 50 KB). Pipeline:

1. מנוי סטרימינג לנרות של SBER.
2. כל דקה — inference של המודל.
3. upon signal — שליחת פקודת limit עם בדיקת ספר הפקודות הנוכחי.
4. תיעוד כל העסקאות ב-PostgreSQL.

תהליך העבודה

• אנליטיקה — אנו דנים באסטרטגיה, בוחרים כלים ומעריכים דרישות השהיה.
• עיצוב — ארכיטקטורת מיקרו-שירותים, בחירת broker (Tinkoff), אב-טיפוס על sandbox.
• מימוש — כתיבת קוד ב-Python, שילוב מודל ML, הגדרת streaming.
• בדיקות — backtesting על נתונים היסטוריים (שנה אחת), paper trading על sandbox.
• פריסה — פריסה על VPS (או Kubernetes), ניטור (Prometheus + Grafana).

מה כוללת העבודה

• בחירה והגדרה של חשבון מסחר ב-Tinkoff.
• פיתוח עטיפת SDK מותאמת אישית עבור האסטרטגיה שלך.
• שילוב מודל ML/LLM (PyTorch, TensorFlow).
• טיפול בשגיאות, logging, reconnects.
• תיעוד ומדריך תפעול.
• תמיכה טכנית למשך חודש לאחר ההשקה.

טעויות נפוצות וכיצד להימנע מהן

ציר זמן

אינטגרציה בסיסית: 1–2 שבועות (כולל סביבת בדיקה). פרויקט מלא עם מודל ואופטימיזציה: 4–6 שבועות. העלות מחושבת באופן אישי לאחר הערכת הפרויקט.

השוואה של Tinkoff Invest API עם מתחרי REST

gRPC מספק השהיה נמוכה פי 2-3 בהשוואה ל-REST — זה קריטי לאסטרטגיות בתדירות גבוהה.

השוואה נוספת: Streaming לעומת Polling

עבור אסטרטגיות רגישות להשהיה, עדיף gRPC streaming. צרו קשר כדי להעריך את הפרויקט שלכם. הזמינו אינטגרציה turnkey — אנו מבטיחים איכות ועמידה בלוחות זמנים.