Momentum-робот на Tinkoff Invest API: ATR-каналы, исполнение и риск

Что собираем

Маленький, но честный momentum-робот на Tinkoff Invest API:

• Подписка на 5-минутные свечи через gRPC stream.
• Сигнал: цена пробивает верх ATR-канала (Donchian + ATR-фильтр).
• Размер позиции: фиксированный риск 0.5% капитала на сделку.
• Исполнение: лимитка с peg-to-mid + 1 tick, кэнселл и
  переставление, если не исполнилась за 3 секунды.
• Стоп: ATR(14) × 2 от цены входа, переходит в б/у при +1R.

Цель — не «лучшая стратегия», а дисциплинированный
каркас: stream → сигнал → ордер → позиция → exit. На него
потом можно навесить любую логику.

Подготовка

python -m venv venv && source venv/bin/activate
pip install tinkoff-investments==0.2.* numpy

Токен с правом торговли создаётся в личном кабинете
(Settings → API). Для боевого аккаунта — отдельный токен,
никогда не коммитьте в git.

import os
TOKEN = os.environ["TINKOFF_TOKEN"]
ACCOUNT_ID = os.environ["TINKOFF_ACCOUNT_ID"]

Скелет приложения

import asyncio
from collections import deque
from datetime import timedelta
from decimal import Decimal

from tinkoff.invest import (
    AsyncClient, CandleInstrument, MarketDataRequest,
    SubscribeCandlesRequest, SubscriptionAction,
    SubscriptionInterval, OrderDirection, OrderType,
)

FIGI = "BBG004730N88"  # SBER, для примера
CANDLE_LIMIT = 200

async def main():
    async with AsyncClient(TOKEN) as client:
        candles = await load_history(client, FIGI, CANDLE_LIMIT)
        state = BotState(candles=candles)

        async def request_iter():
            yield MarketDataRequest(
                subscribe_candles_request=SubscribeCandlesRequest(
                    subscription_action=SubscriptionAction.SUBSCRIPTION_ACTION_SUBSCRIBE,
                    instruments=[CandleInstrument(
                        figi=FIGI,
                        interval=SubscriptionInterval.SUBSCRIPTION_INTERVAL_FIVE_MINUTES,
                    )],
                )
            )
            # держим стрим живым
            while True:
                await asyncio.sleep(60)

        async for msg in client.market_data_stream.market_data_stream(request_iter()):
            if msg.candle is None:
                continue
            await state.on_candle(client, msg.candle)

Ключевое: загружаем историю до старта стрима, чтобы
индикаторы были тёплыми с первой минуты.

ATR-канал и сигнал

import numpy as np

def atr(highs, lows, closes, period=14):
    tr = np.maximum.reduce([
        highs[1:] - lows[1:],
        np.abs(highs[1:] - closes[:-1]),
        np.abs(lows[1:]  - closes[:-1]),
    ])
    return np.mean(tr[-period:])

def signal(candles, lookback=20, atr_mult=0.5):
    highs  = np.array([c.high  for c in candles])
    lows   = np.array([c.low   for c in candles])
    closes = np.array([c.close for c in candles])

    upper = highs[-lookback-1:-1].max() + atr_mult * atr(highs, lows, closes)
    last_close = closes[-1]

    if last_close > upper:
        return "long"
    return None

atr_mult=0.5 — это фильтр: робот не входит на «бумажный»
пробой, движение должно быть значимым по сравнению с
текущей волатильностью.

Контроль риска и размер позиции

Не магические лоты, а математика:

def position_size(equity_rub, entry, stop, lot_size=10):
    risk_per_trade = equity_rub * 0.005  # 0.5% капитала
    risk_per_share = abs(entry - stop)
    if risk_per_share <= 0:
        return 0
    qty = risk_per_trade / risk_per_share
    lots = int(qty // lot_size)
    return max(lots, 0)

На капитале 500 000 ₽, ATR(14)=3.2 ₽ для SBER, стоп
на 6.4 ₽ ниже входа: размер позиции = 500 000 × 0.005 / 6.4
≈ 390 акций ≈ 39 лотов. Если ATR вырастет вдвое — размер
автоматически уполовинится. Это главное правило:
размер подстраивается под волатильность, а не наоборот.

Исполнение

Рыночные ордера на тонком стакане третьего эшелона
съедают 5–15 пунктов проскальзывания. Лимитка с
подкруткой даёт исполнение по mid + 1 тик в 80%
случаев:

async def smart_buy(client, figi, lots, max_attempts=3):
    for attempt in range(max_attempts):
        book = await client.market_data.get_order_book(figi=figi, depth=1)
        bid = quotation_to_decimal(book.bids[0].price)
        ask = quotation_to_decimal(book.asks[0].price)
        mid = (bid + ask) / 2
        tick = quotation_to_decimal(book.last_price) - bid  # упрощение
        price = mid + tick

        order = await client.orders.post_order(
            figi=figi,
            quantity=lots,
            price=decimal_to_quotation(price),
            direction=OrderDirection.ORDER_DIRECTION_BUY,
            account_id=ACCOUNT_ID,
            order_type=OrderType.ORDER_TYPE_LIMIT,
            order_id=str(uuid.uuid4()),
        )

        await asyncio.sleep(3)
        state = await client.orders.get_order_state(
            account_id=ACCOUNT_ID, order_id=order.order_id
        )
        if state.execution_report_status.name == "EXECUTION_REPORT_STATUS_FILL":
            return state
        await client.orders.cancel_order(
            account_id=ACCOUNT_ID, order_id=order.order_id,
        )
    # fallback — рынок
    return await client.orders.post_order(
        figi=figi, quantity=lots,
        direction=OrderDirection.ORDER_DIRECTION_BUY,
        account_id=ACCOUNT_ID,
        order_type=OrderType.ORDER_TYPE_MARKET,
        order_id=str(uuid.uuid4()),
    )

Три попытки лимиткой, потом fallback на рынок. Это
компромисс между slippage и риском пропустить движение.

Стоп и breakeven

class Position:
    def __init__(self, entry, stop, qty, atr_value):
        self.entry = entry
        self.stop = stop
        self.qty = qty
        self.r_unit = abs(entry - stop)  # 1R в рублях
        self.moved_to_be = False

    def update(self, last_price):
        if not self.moved_to_be and last_price - self.entry >= self.r_unit:
            self.stop = self.entry  # перевод в б/у
            self.moved_to_be = True

    def stopped_out(self, last_price):
        return last_price <= self.stop

После +1R стоп переезжает на цену входа — это убирает
риск убытка по позиции, ценой того, что часть «хороших»
сделок закроется в нуле. На длинной дистанции это
добавляет к Sharpe больше, чем съедает.

Чего ещё не хватает для прода

То, что разобрали — это MVP. Перед боевым запуском
нужно добавить:

1. Persistence состояния. Сейчас при рестарте робот
   забывает про открытые позиции. Минимум — SQLite с
   таблицей positions(figi, entry, stop, qty, opened_at).
2. Мониторинг стрима. Tinkoff API иногда роняет
   соединение. Нужен heartbeat-таймер: если 30 секунд нет
   сообщений — реконнект.
3. Sanity checks перед ордером. Проверка торгового
   статуса (get_trading_status), запрет торговли в
   первые 3 минуты сессии и в последние 5 (волатильность
   аукциона).
4. Ограничение по дневному убытку. «-2R за день — кладём
   робота до завтра». Без этого один плохой день режет
   капитал на 10–15%.
5. Логирование ВСЕГО. Каждый сигнал, каждый ордер,
   каждый ответ API — в structured log. Без этого
   невозможно дебажить расхождение между бэктестом и
   боем.

Что не делать

• Не запускайте на боевом счёте до 2 недель paper trading.
  На demo вы найдёте 5–10 багов исполнения, которые
  не видны в коде.
• Не используйте OrderType.MARKET как единственный
  вариант исполнения, кроме экстренного выхода.
• Не тестируйте «вживую» на третьем эшелоне.
  Низкая ликвидность убьёт стратегию через slippage даже
  при идеальном сигнале.

Каркас выше — около 200 строк рабочего кода. Дальше —
это уже не вопрос «можно ли написать робота», а вопрос
«есть ли у вас сигнал, который выживает walk-forward».
А это, как мы знаем, другая статья.