Tinkoff Invest
Project description
T-Invest
pip install tinvest
Данный проект представляет собой инструментарий на языке Python для работы с OpenAPI Тинькофф Инвестиции, который можно использовать для создания торговых роботов.
Клиент предоставляет синхронный и асинхронный API для взаимодействия с Тинькофф Инвестиции.
Начало работы
Где взять токен аутентификации?
В разделе инвестиций вашего личного кабинета tinkoff . Далее:
- Перейдите в настройки
- Проверьте, что функция "Подтверждение сделок кодом" отключена
- Выпустите токен для торговли на бирже и режима "песочницы" (sandbox)
- Скопируйте токен и сохраните, токен отображается только один раз, просмотреть его позже не получится, тем не менее вы можете выпускать неограниченное количество токенов
Документация
Быстрый старт
Для непосредственного взаимодействия с 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())
Sandbox
Sandbox позволяет вам попробовать свои торговые стратегии, при этом не тратя реальные средства. Протокол взаимодействия полностью совпадает с Production окружением.
client = tinvest.AsyncClient(SANDBOX_TOKEN, use_sandbox=True)
# client = tinvest.SyncClient(SANDBOX_TOKEN, use_sandbox=True)
api = tinvest.SandboxApi(client)
Contributing
Предлагайте свои пулл реквесты, проект с открытым исходным кодом.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.