T-Invest

pip install tinvest

Данный проект представляет собой инструментарий на языке Python для работы с OpenAPI Тинькофф Инвестиции, который можно использовать для создания торговых роботов.

Клиент предоставляет синхронный и асинхронный API для взаимодействия с Тинькофф Инвестиции.

Начало работы

Где взять токен аутентификации?

В разделе инвестиций вашего личного кабинета tinkoff . Далее:

• Перейдите в настройки
• Проверьте, что функция "Подтверждение сделок кодом" отключена
• Выпустите токен для торговли на бирже и режима "песочницы" (sandbox)
• Скопируйте токен и сохраните, токен отображается только один раз, просмотреть его позже не получится, тем не менее вы можете выпускать неограниченное количество токенов

Документация

tinvest

invest-openapi

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

Для непосредственного взаимодействия с OpenAPI нужно создать клиента. Клиенты разделены на streaming и rest.

Примеры использования SDK находятся ниже.

У меня есть вопрос

Основной репозиторий с документацией — в нем вы можете задать вопрос в Issues и получать информацию о релизах в Releases. Если возникают вопросы по данному SDK, нашёлся баг или есть предложения по улучшению, то можно задать его в Issues.

Также можно выполнить команду в консоли.

tinvest issues
tinvest issues openapi

Примеры

Для работы с данным пакетом вам нужно изучить OpenAPI Тинькофф Инвестиции

Streaming

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

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

Если вам требуется время сервера, то можете указать принимаемый аргумент в хендлере под именем server_time. В хендлер будет передано серверное время в формате datetime.

При сетевых сбоях будет произведена попытка переподключения.

import asyncio
from datetime import datetime

import tinvest

TOKEN = "<TOKEN>"

events = tinvest.StreamingEvents()


@events.reconnect()
def handle_reconnect():
    print('Reconnecting')


@events.candle()
async def handle_candle(
    api: tinvest.StreamingApi,
    payload: tinvest.CandleStreaming,
    server_time: datetime  # [optional] if you want
):
    print(payload)


@events.orderbook()
async def handle_orderbook(
    api: tinvest.StreamingApi, payload: tinvest.OrderbookStreaming
):
    print(payload)


@events.instrument_info()
async def handle_instrument_info(
    api: tinvest.StreamingApi, payload: tinvest.InstrumentInfoStreaming
):
    print(payload)


@events.error()
async def handle_error(
    api: tinvest.StreamingApi, payload: tinvest.ErrorStreaming
):
    print(payload)


@events.startup()
async def startup(api: tinvest.StreamingApi):
    await api.candle.subscribe("BBG0013HGFT4", tinvest.CandleResolution.min1)
    await api.orderbook.subscribe("BBG0013HGFT4", 5, "123ASD1123")
    await api.instrument_info.subscribe("BBG0013HGFT4")


@events.cleanup()
async def cleanup(api: tinvest.StreamingApi):
    await api.candle.unsubscribe("BBG0013HGFT4", "1min")
    await api.orderbook.unsubscribe("BBG0013HGFT4", 5)
    await api.instrument_info.unsubscribe("BBG0013HGFT4")


async def main():
    await tinvest.Streaming(TOKEN, state={"postgres": ...}).add_handlers(events).run()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Синхронный REST API Client

Для выполнения синхронных http запросов используется библиотека requests. С описанием клиентов можно ознакомиться по этой ссылке. Возвращаемый аргумент это типичный объект класса requests.Response, который имеет методы parse_json и parse_error. Эти методы возвращают объекты, наследованные от класса pydantic.BaseModel.

import tinvest

TOKEN = "<TOKEN>"

client = tinvest.SyncClient(TOKEN)
api = tinvest.PortfolioApi(client)

response = api.portfolio_get()  # requests.Response
if response.status_code == 200:
    print(response.parse_json())  # tinvest.PortfolioResponse

# Handle error
...
api = tinvest.OperationsApi(client)

response = api.operations_get("", "")
if response.status_code != 200:
    print(response.parse_error())  # tinvest.Error

Асинхронный REST API Client

Для выполнения асинхронных http запросов используется библиотека aiohttp. Клиенты имеют такой же интерфейс как в синхронной реализации, за исключением того, что функции возвращают объект корутина.

import asyncio
import tinvest

TOKEN = "<TOKEN>"

client = tinvest.AsyncClient(TOKEN)
api = tinvest.PortfolioApi(client)


async def request():
    async with api.portfolio_get() as response:  # aiohttp.ClientResponse
        if response.status == 200:
            print(await response.parse_json())  # tinvest.PortfolioResponse


loop = asyncio.get_event_loop()
loop.run_until_complete(request())

Contributing

Предлагайте свои пулл реквесты, проект с открытым исходным кодом.