Finam Trade API — Python SDK

Python SDK Finam для gRPC Trade API. Оборачивает сгенерированные заглушки:

• одна точка входа FinamClient / AsyncFinamClient;
• автоматический выпуск JWT и фоновое обновление (через AuthService.SubscribeJwtRenewal);
• помощник from_rpc_error() для типизированных ошибок;
• повторы с экспоненциальной задержкой для UNAVAILABLE и одобренные сервером повторы при ограничении частоты запросов.

Методы сервисов вызываются напрямую на сгенерированных заглушках, без дополнительного слоя преобразования запросов и ответов.

Установка

Нужен Python 3.10 или новее.

Текущий checkout соответствует версии 2.19.1. Создайте и активируйте виртуальное окружение, затем установите пакет:

python3 -m venv .venv
source .venv/bin/activate
python -m pip install finam-sdk==2.19.1

Дистрибутив в PyPI называется finam-sdk; имя импорта в Python — finam_trade_api. (Имя finam-trade-api в PyPI занято сторонним REST-клиентом, не связанным с этим проектом.)

Быстрый старт

Первая программа аутентифицируется и печатает ID счетов, видимых секрету. Она ограничена и не выставляет заявку.

Сохраните это как quickstart.py:

import os

from finam_trade_api import FinamClient
from finam_trade_api.auth_messages import TokenDetailsRequest


secret = os.environ["TRADE_API_SECRET"]

with FinamClient(secret=secret) as client:
    token = client.get_token()
    if token is None:
        raise RuntimeError("Authentication did not return a token")

    details = client.auth.TokenDetails(TokenDetailsRequest(token=token))
    print("Available account IDs:", list(details.account_ids))

Запустите внутри активированного виртуального окружения:

TRADE_API_SECRET=... python quickstart.py

Если аутентификация не проходит, убедитесь, что секрет активен. Пустой список счетов означает, что токен не открывает торговый счёт; он всё ещё может подходить для рыночных данных, если у него есть соответствующее право.

Чтобы получить один из обнаруженных счетов, расширьте программу так:

import os

from finam_trade_api import FinamClient
from finam_trade_api.accounts import GetAccountRequest
from finam_trade_api.auth_messages import TokenDetailsRequest


with FinamClient(secret=os.environ["TRADE_API_SECRET"]) as client:
    token = client.get_token()
    if token is None:
        raise RuntimeError("Authentication did not return a token")

    details = client.auth.TokenDetails(TokenDetailsRequest(token=token))
    if not details.account_ids:
        raise RuntimeError("This secret exposes no accounts")

    account_id = details.account_ids[0]
    account = client.accounts.GetAccount(GetAccountRequest(account_id=account_id))
    print(account)

Быстрый старт с asyncio

import asyncio
import os

from finam_trade_api import AsyncFinamClient
from finam_trade_api.auth_messages import TokenDetailsRequest


async def main() -> None:
    async with AsyncFinamClient(secret=os.environ["TRADE_API_SECRET"]) as client:
        token = client.get_token()
        if token is None:
            raise RuntimeError("Authentication did not return a token")

        details = await client.auth.TokenDetails(TokenDetailsRequest(token=token))
        print("Available account IDs:", list(details.account_ids))


asyncio.run(main())

Подписка на рыночные данные

Потоковые RPC возвращают итераторы. Этот синхронный пример работает до Ctrl-C; выход из контекстного менеджера закрывает канал и поток обновления токена:

import os

from finam_trade_api import FinamClient
from finam_trade_api.market_data import SubscribeQuoteRequest


with FinamClient(secret=os.environ["TRADE_API_SECRET"]) as client:
    for tick in client.market_data.SubscribeQuote(
        SubscribeQuoteRequest(symbols=["SBER@MISX"])
    ):
        print(tick)

В репозитории также есть точечные примеры для аутентификации и счетов, асинхронного потока котировок и настоящей заявки с отменой. Они живут в автономном проекте-потребителе, который по умолчанию устанавливает опубликованный пакет.

Доступные сервисы

Клиент открывает эти сервисы Trade API как подклиенты:

Справочник API

Каждый RPC перечисленных ниже сервисов доступен напрямую на своём подклиенте. Типы сообщений запросов и ответов реэкспортируются из коротких модулей по сервисам:

Исходные глубоко вложенные пути (finam_trade_api.proto.grpc.tradeapi.v1.<service>.<service>_service_pb2) по-прежнему работают и остаются источником истины.

Обозначения: ▶ унарный · ⇉ серверный поток · ⇄ двунаправленный поток

client.auth — AuthService

AuthService принимает секрет или токен в теле запроса и должен вызываться без заголовка Authorization: сервер отклоняет TokenDetails, если заголовок присутствует. SDK держит client.auth на отдельном канале без учётных данных вызова, поэтому вручную ничего делать не нужно.

client.accounts — AccountsService

client.assets — AssetsService

client.corporate_actions — CorporateActionsService

client.market_data — MarketDataService

client.orders — OrdersService

client.reports — ReportsService

client.metrics — UsageMetricsService

Жизненный цикл клиента

for_testing(...) открывает небезопасный (plaintext) канал к локальному фейковому серверу. Никогда не используйте его с api.finam.ru — JWT уйдёт открытым текстом.

Обработка ошибок

Вызовы SDK поднимают «сырые» grpc.RpcError. Используйте from_rpc_error(), когда приложению полезна типизированная иерархия FinamError:

import grpc
from finam_trade_api import FinamClient, RateLimitError, from_rpc_error

with FinamClient(secret="...") as client:
    try:
        client.accounts.GetAccount(GetAccountRequest(account_id="A12345"))
    except grpc.RpcError as raw:
        err = from_rpc_error(raw)
        if isinstance(err, RateLimitError):
            ...
        raise err

Классы исключений: AuthError (401), PermissionDeniedError (403), InvalidArgumentError (400), NotFoundError (404), RateLimitError (429), InternalError (500), ServiceUnavailableError (503), DeadlineExceededError (504). Все наследуются от FinamError.

Повторы

Унарные RPC автоматически повторяются при UNAVAILABLE с экспоненциальной задержкой и джиттером. RESOURCE_EXHAUSTED повторяется только когда сервер передаёт grpc-retry-pushback-ms; иначе SDK возвращает ошибку ограничения частоты, а не усиливает нагрузку. Потоковые RPC не повторяются — ожидается, что вызывающий код переподключится на осмысленной границе, например с последней полученной свечи.

Переопределить политику:

from finam_trade_api import FinamClient, RetryPolicy

policy = RetryPolicy(max_attempts=6, initial_backoff=0.5, max_backoff=10.0)
client = FinamClient(secret="...", retry_policy=policy)

Локальная сборка

Из корня репозитория:

cd sdk/python
uv sync --locked
uv run ./scripts/generate_proto.sh
uv run ruff check finam_trade_api tests
uv run mypy finam_trade_api
uv run pytest

scripts/generate_proto.sh компилирует .proto-файлы из ../../proto/ в finam_trade_api/proto/. Перезапускайте его при каждом изменении контрактов.

Структура

sdk/python/
├── pyproject.toml
├── uv.lock
├── README.md
├── LICENSE
├── scripts/
│   ├── generate_proto.sh       # вызов protoc (только для контрибьюторов)
│   └── verify_distribution.py  # проверка содержимого wheel и sdist в CI
└── finam_trade_api/
    ├── __init__.py
    ├── client.py               # FinamClient (sync)
    ├── aio.py                  # AsyncFinamClient
    ├── auth.py                 # жизненный цикл JWT
    ├── retry.py                # политика повторов + интерсепторы
    ├── exceptions.py           # типизированные ошибки
    ├── _insecure_auth.py       # аутентификация на plaintext-канале для тестов
    ├── _metadata.py            # передача заголовка Authorization
    ├── _services.py            # ленивый реестр сгенерированных сервисов
    ├── accounts.py             # реэкспорт сообщений (по сервисам)
    ├── assets.py
    ├── auth_messages.py
    ├── corporate_actions.py
    ├── market_data.py
    ├── orders.py
    ├── reports.py
    ├── metrics.py
    └── proto/                  # генерируется CI; входит в wheel и sdist