Template for an asynchronous trading strategy
I want to share a useful experience for those who develop trading robots – a code template for asynchronous trading strategy (live And offline), this code can be easily rebuilt for any instrument data provider API.
The code itself was published on Github by Tinkoff: https://github.com/Tinkoff/invest-python/blob/main/examples/wiseplat_live_strategy_print_ohlcv.py
I will immediately give a link to the same revised code by me, which uses data from open access provided by the provider – the MOEX exchange (you don’t even need to receive a token) for trading instruments (though with a delay of 15 minutes) https://github.com/WISEPLAT/Hackathon-Finam-NN-Trade-Robot/blob/master/7_live_strategy.py
Structure of the trading strategy code
Typically, the code for a trading strategy consists of the following blocks:
Code Run Block
API connection block to the data provider, in the cycle of which a separate instance of the trading strategy is launched for each instrument
Each running strategy instance has:
Entry block (once execution of code), usually used to initialize variables, etc.
Block of an infinite loop, in which the logic of the trading strategy is already executed
e.g. checking that the market is open – or waiting for it to open
obtaining historical and live data on the instrument
processing of received data by trading logic
Now let’s take a closer look at this code.
Code Run Block
if __name__ == "__main__":
sber_figi = "BBG004730N88"
vtbr_figi = "BBG004730ZJ9"
portfolio = {sber_figi, vtbr_figi}
timeframe = CandleInterval.CANDLE_INTERVAL_1_MIN
days_back = 1
check_interval = 10 # seconds to check interval for new completed candle
loop = asyncio.get_event_loop()
task = loop.create_task(
run_strategy(
portfolio=portfolio,
timeframe=timeframe,
days_back=days_back,
check_interval=check_interval,
)
)
loop.run_until_complete(task)
API connection block to the data provider
async def run_strategy(portfolio, timeframe, days_back, check_interval):
"""From this function we are starting
strategy for every ticker from portfolio.
"""
async with AsyncClient(token=TOKEN, app_name="TinkoffApp") as client:
strategy_tasks = []
for instrument in portfolio:
strategy = LogOnlyCandlesStrategy(
figi=instrument,
timeframe=timeframe,
days_back=days_back,
check_interval=check_interval,
client=client,
)
strategy_tasks.append(asyncio.create_task(strategy.start()))
await asyncio.gather(*strategy_tasks)
Entry Block (Single Code Execution)
async def start(self):
"""Strategy starts from this function."""
if self.account_id is None:
try:
self.account_id = (
(await self.client.users.get_accounts()).accounts.pop().id
)
except AioRequestError as are:
logger.error("Error taking account id. Stopping strategy. %s", are)
return
await self.main_cycle()
Infinite loop block (execution of trading strategy logic)
async def main_cycle(self):
"""Main cycle for live strategy."""
while True:
try:
await self.ensure_market_open()
await self.get_historical_data()
# put your strategy code here for live
# to generate signals for buying or selling tickers
logger.debug(
"- live mode: run some strategy code to buy or sell - figi=%s",
self.figi,
)
except AioRequestError as are:
logger.error("Client error %s", are)
await asyncio.sleep(self.check_interval)
I tried to document the code examples for an asynchronous trading strategy available at the links above.
I hope you find this useful +1 and help you save your time!