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 details)
Built Distribution
File details
Details for the file tinkoff-api-1.19.tar.gz
.
File metadata
- Download URL: tinkoff-api-1.19.tar.gz
- Upload date:
- Size: 20.8 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/3.8.0 pkginfo/1.8.2 readme-renderer/32.0 requests/2.27.1 requests-toolbelt/0.9.1 urllib3/1.26.8 tqdm/4.62.3 importlib-metadata/4.11.0 keyring/23.5.0 rfc3986/2.0.0 colorama/0.4.4 CPython/3.10.1
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | b26f53c11a3ccdfbfba7f5f4859ac1ce93203fa48120adcf25b7b0502ed03c5c |
|
MD5 | f7b4f4ac7a3dde68346a6ffcdc138118 |
|
BLAKE2b-256 | e7be6a08053b3d24870892db4f15cd02f668cf43cab228ce32b74cd36261e23a |
File details
Details for the file tinkoff_api-1.19-py3-none-any.whl
.
File metadata
- Download URL: tinkoff_api-1.19-py3-none-any.whl
- Upload date:
- Size: 28.6 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/3.8.0 pkginfo/1.8.2 readme-renderer/32.0 requests/2.27.1 requests-toolbelt/0.9.1 urllib3/1.26.8 tqdm/4.62.3 importlib-metadata/4.11.0 keyring/23.5.0 rfc3986/2.0.0 colorama/0.4.4 CPython/3.10.1
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | cce72f0afe27a28c9520b7c347982b4a7e6847ac8ba9dcc0f7055b4dd8bb943d |
|
MD5 | 11ef31e3210a3eb7fd5b013ddbae80ea |
|
BLAKE2b-256 | 213dabb5c1ca89463c8584986f5abb89a91205eac5a80b97c52705ef40c35ac6 |