Python Tinkoff API client for asyncio and humans
Project description
tinkoff-api
Python Tinkoff API client for asyncio and humans.
Table of contens
Covered APIs
- Tinkoff Investments (official docs)
Features
- Clients for both REST and Streaming protocols in Tinkoff Investments
- Presence of data classes for all interaction with API
- Automatic reconnection and keep-alive connections
- Internal exclusive rate limiter for every resource in REST protocol
- Friendly exceptions for API errors
Installation
Use pip to install:
$ pip install tinkoff-api
Usage examples
REST API client:
import asyncio from datetime import datetime from tinkoff.investments import ( TinkoffInvestmentsRESTClient, Environment, CandleResolution, ) from tinkoff.investments.client.exceptions import TinkoffInvestmentsError async def show_apple_year_candles(): try: async with TinkoffInvestmentsRESTClient( token="TOKEN", environment=Environment.SANDBOX ) as client: candles = await client.market.candles.get( figi="BBG000B9XRY4", dt_from=datetime(2019, 1, 1), dt_to=datetime(2019, 12, 31), interval=CandleResolution.DAY, ) for candle in candles: print(f"{candle.time}: {candle.h}") except TinkoffInvestmentsError as e: print(e) async def jackpot(): try: async with TinkoffInvestmentsRESTClient( token="TOKEN", environment=Environment.SANDBOX ) as client: instruments = await client.market.instruments.search("AAPL") apple = instruments[0] account = await client.sandbox.accounts.register() await client.sandbox.accounts.positions.set_balance( figi=apple.figi, balance=100, broker_account_id=account.brokerAccountId, ) print("We created the following portfolio:") positions = await client.portfolio.get_positions() for position in positions: print(f"{position.name}: {position.lots} lots") except TinkoffInvestmentsError as e: print(e) asyncio.run(jackpot())
Streaming Client:
import asyncio from datetime import datetime from tinkoff.investments import ( CandleEvent, CandleResolution, TinkoffInvestmentsStreamingClient, ) client = TinkoffInvestmentsStreamingClient(token="TOKEN") @client.events.candles("BBG009S39JX6", CandleResolution.MIN_1) @client.events.candles("BBG000B9XRY4", CandleResolution.MIN_1) async def on_candle(candle: CandleEvent, server_time: datetime): print(candle, server_time) asyncio.run(client.run())
Dynamic subscriptions in runtime:
import asyncio from datetime import datetime from tinkoff.investments import ( CandleEvent, CandleResolution, TinkoffInvestmentsStreamingClient, ) client = TinkoffInvestmentsStreamingClient(token="TOKEN") @client.events.candles("BBG000B9XRY4", CandleResolution.HOUR) async def on_candle(candle: CandleEvent, server_time: datetime): if candle.h > 1000: await client.events.candles.subscribe( callback=on_candle, figi=candle.figi, interval=CandleResolution.MIN_1, ) elif candle.h < 1000: await client.events.candles.unsubscribe( candle.figi, CandleResolution.MIN_1 ) asyncio.run(client.run())
Complete simple bot:
import asyncio from datetime import datetime from tinkoff.investments import ( CandleEvent, CandleResolution, OperationType, TinkoffInvestmentsRESTClient, TinkoffInvestmentsStreamingClient, ) streaming = TinkoffInvestmentsStreamingClient("TOKEN") rest = TinkoffInvestmentsRESTClient("TOKEN") @streaming.events.candles("BBG000B9XRY4", CandleResolution.MIN_1) async def buy_apple(candle: CandleEvent, server_time: datetime): if candle.c > 350: await rest.orders.create_market_order( figi="BBG000B9XRY4", lots=1, operation=OperationType.BUY, broker_account_id=123, ) asyncio.run(streaming.run())
Historical data:
import asyncio from datetime import datetime from tinkoff.investments import ( CandleResolution, Environment, TinkoffInvestmentsRESTClient, ) from tinkoff.investments.utils.historical_data import HistoricalData async def get_minute_candles(): # show 1 minute candles for AAPL in 1 year period of time async with TinkoffInvestmentsRESTClient( token="TOKEN", environment=Environment.SANDBOX ) as client: historical_data = HistoricalData(client) async for candle in historical_data.iter_candles( figi="BBG000B9XRY4", dt_from=datetime(2019, 1, 1), dt_to=datetime(2020, 1, 1), interval=CandleResolution.MIN_1, ): print(candle) asyncio.run(get_minute_candles())
TODO
- allow to provide str constants along with specific enum objects
- add ability to unsubscribe by pattern
- rename some fields
- make some fields in snake case
- generate documentation
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.
Source Distribution
tinkoff-api-1.19.tar.gz
(20.8 kB
view hashes)
Built Distribution
tinkoff_api-1.19-py3-none-any.whl
(28.6 kB
view hashes)
Close
Hashes for tinkoff_api-1.19-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | cce72f0afe27a28c9520b7c347982b4a7e6847ac8ba9dcc0f7055b4dd8bb943d |
|
MD5 | 11ef31e3210a3eb7fd5b013ddbae80ea |
|
BLAKE2-256 | 213dabb5c1ca89463c8584986f5abb89a91205eac5a80b97c52705ef40c35ac6 |