Development and Integration of an AI Trading Bot with Tinkoff Invest API

When developing an algorithmic trading bot for the Russian market, one of the first challenges is choosing a reliable and performant API. Tinkoff Invest API on gRPC is a common solution, but integrating it with an ML model requires a deep understanding of both the API and MLOps.

In addition to the standard REST, Tinkoff offers a gRPC protocol that provides p99 latency of 20–50 ms — this is 2–3 times faster than REST competitors with their 100–300 ms.

Such a difference is critical for high-frequency strategies, where every millisecond affects profitability. We have experience implementing three projects with this API, including high-frequency strategies. Our guarantee is stable 24/7 operation of the bot with minimal latency.

Evaluate your project: get a consultation with an engineer.

Problems we solve

Latency and delays. Tinkoff provides streaming data via gRPC stream, but suboptimal instrument subscription can add an extra 50–100 ms. We configure stream multiplexing and use asynchronous clients to reduce p99 latency to 20 ms.

This cuts reaction time to market changes by 40%, which translates directly into budget savings by reducing missed opportunities.

Rate limits. Market data — 300 requests/min, trading operations — 100 orders/min. Without planning, a signal can be missed. We implement queuing and batch processing to stay within limits without losses. This approach reduces infrastructure costs by 15%.

FIGI vs Ticker. The API uses FIGI, mapping via find_instrument — an extra round-trip. We cache the mapping at startup and refresh it once a day. This eliminates collisions when trading securities with identical tickers.

How to integrate an AI trading bot with the Tinkoff Invest API?

Integration begins with setting up an asynchronous client and subscribing to streaming data. Below is typical code for retrieving candles and placing an order.

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

Real-time streaming data

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

A mandatory stage before live trading:

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

How does Tinkoff Invest API handle high load?

Streaming data arrives via a gRPC bidirectional stream, which makes it possible to receive candles, trades, and order books in real time. But when subscribing to 50+ instruments, reconnects need to be organized properly. We use exponential backoff and a health check every 5 seconds. Tinkoff Invest API confirms these recommendations.

Why is FIGI important for correct trading?

If you use tickers directly, you can make mistakes when trading securities with identical tickers (for example, different share classes). FIGI is a global identifier that eliminates collisions. We always set up automatic FIGI resolution via a local database. You can read more about FIGI on Wikipedia.

How we do it: stack and case study

We use PyTorch for the model, Tinkoff Invest SDK for Python, an asynchronous client. From our practice: implementing a mean-reversion strategy on Sberbank (SBER) stocks. The model is trained on 1-minute candles with a 60-minute lookback. In production we use Docker with CPU-only inference (the model is small, 50 KB). Pipeline:

1. Streaming subscription to SBER candles.
2. Every minute — model inference.
3. On a signal — placing a limit order with a check of the current order book.
4. Logging all trades to PostgreSQL.

Work Process

1. Analytics — we discuss the strategy, choose tools, and evaluate latency requirements.
2. Design — microservices architecture, broker selection (Tinkoff), prototype on sandbox.
3. Implementation — writing code in Python, integrating the ML model, setting up streaming.
4. Testing — backtesting on historical data (1 year), paper trading on sandbox.
5. Deployment — deployment on VPS (or Kubernetes), monitoring (Prometheus + Grafana).

What the work includes

• Selection and setup of a Tinkoff trading account.
• Development of a custom SDK wrapper for your strategy.
• Integration of an ML/LLM model (PyTorch, TensorFlow).
• Error handling, logging, reconnects.
• Documentation and an operations manual.
• Technical support for 1 month after launch.

Common mistakes and how to avoid them

Timeline

Basic integration: 1–2 weeks (including sandbox). Full project with model and optimization: 4–6 weeks. Cost is calculated individually after project assessment.

Comparison of Tinkoff Invest API with REST competitors

gRPC provides 2-3 times lower latency compared to REST — this is critical for high-frequency strategies.

Additional Comparison: Streaming vs Polling

For latency-sensitive strategies, gRPC streaming is preferable.

Contact us to evaluate your project. Order a turnkey integration — we guarantee quality and adherence to deadlines.