Python 3.9+
This page takes you through the steps of coding your first strategy in Python 3.9+
In this section we will create a simple EMA based Golden Crossover strategy. Let's start!
Golden Crossover Strategy Python 3+ Sample on GitHub
If you can't wait to see the full code, go straight to the Algorum GitHub and pull the samples repo, which contains the sample strategy explained here. The py file to look into in the repo is golden_crossover_quant_strategy.py.
Project Setup
Download latest version of PyCharm IDE community edition for your OS (Windows or Linux). PyCharm is a popular IDE (Integrated Development Environment) for Python, with support for syntax checking, debugging, etc., After PyCharm is installed, create a new Python 3.9+ Project in PyCharm. Create a new sub folder named src in your project root directory.
Install the following dependencies for your project from the terminal of the PyCharm in the project root directory.
Library Name | Install Command |
---|---|
websocket-client | pip install websocket-client |
jsonpickle | pip install jsonpickle |
algorum-quant-client | pip install algorum-quant-client-py3 |
Strategy Class Setup
Add a new Python class named GoldenCrossoverQuantStrategy in the src sub folder. Derive the GoldenCrossoverQuantStrategy class from AlgorumQuantClient.quant_client.QuantEngineClient base class. QuantEngineClient class provides connectivity and API communication with your Algorum Quant Engine running in Algorum Cloud.
import AlgorumQuantClient.quant_client
import AlgorumQuantClient.algorum_types
class GoldenCrossoverQuantStrategy(AlgorumQuantClient.quant_client.QuantEngineClient):
...
Strategy Initialization
Write Initialization code for your strategy where you will subscribe for the symbols you want to trade in this strategy, create indicator evaluators for calculating EMA 50 and 200 periods. Ans also define some state that you would be using in the strategy (to track active orders, order direction, etc.,).
import datetime
import threading
import traceback
import uuid
import AlgorumQuantClient.quant_client
import AlgorumQuantClient.algorum_types
class GoldenCrossoverQuantStrategy(AlgorumQuantClient.quant_client.QuantEngineClient):
Capital = 100000
Leverage = 3 # 3x Leverage on Capital
class State(object):
def __init__(self):
self.Bought = False
self.LastTick = None
self.CurrentTick = None
self.Orders = []
self.CurrentOrderId = None
self.CurrentOrder = None
self.CrossAboveObj = None
def __init__(self, url, apikey, launchmode, sid, user_id, trace_ws=False):
try:
# Pass constructor arguments to base class
super(GoldenCrossoverQuantStrategy, self).__init__(url, apikey, launchmode, sid, user_id, trace_ws)
# Load any saved state
state_json_str = self.get_data("state")
if state_json_str is not None:
self.State = jsonpickle.decode(state_json_str)
if self.State is None or launchmode == AlgorumQuantClient.algorum_types.StrategyLaunchMode.Backtesting:
self.State = GoldenCrossoverQuantStrategy.State()
self.State.CrossAboveObj = AlgorumQuantClient.algorum_types.CrossAbove()
self.StateLock = threading.RLock()
# Subscribe for our symbol data
# For India users
self.symbol = AlgorumQuantClient.algorum_types.TradeSymbol(
AlgorumQuantClient.algorum_types.SymbolType.Stock,
'TATAMOTORS')
# For USA users
# self.symbol = AlgorumQuantClient.algorum_types.TradeSymbol(
# AlgorumQuantClient.algorum_types.SymbolType.Stock,
# 'AAPL')
symbols = [self.symbol]
self.subscribe_symbols(symbols)
# Create indicator evaluator, which will be automatically synchronized with the real time or backtesting
# data that is streaming into this algo
self.evaluator = self.create_indicator_evaluator(
AlgorumQuantClient.algorum_types.CreateIndicatorRequest(
self.symbol,
AlgorumQuantClient.algorum_types.CandlePeriod.Minute,
5))
except Exception:
print(traceback.format_exc())
self.log(AlgorumQuantClient.algorum_types.LogLevel.Error, traceback.format_exc())
The url parameter is the URL of your Algorum Quant Engine and of the form wss://india-quant-engine-api.algorum.net/quant/engine/api/v1. For local testing and debugging, you will have to point this to the endpoint of the Algorum Quant Engine that was described in the First, understand your Algorum Quant Engine section of this documentation. When you deploy your strategy to Algorum Cloud, this parameter will be automatically filled in with your Algorum Quant Engine endpoint.
The apikey parameter is your Algorum API Key. For local testing and debugging, you will have to pass this yourself. When you deploy your strategy to Algorum Cloud, this parameter will be automatically filled in with your Algorum API Key.
The launchmode parameter can be any strategy Launch Mode (Backtesting, Paper Trading, Live Trading) that you desire during your local testing and debugging. When you deploy your strategy to Algorum Cloud, this parameter will be automatically filled in with the Launch Mode as specified when running the strategy using Algorum CLI.
The sid parameter is the unique identifier of your strategy. You can pass anything for this during your local testing and debugging. When you deploy your strategy to Algorum Cloud, this parameter will be automatically filled in with the strategy's unique identifier.
The user_id parameter is the unique identifier of your Algorum user account. You have to pass your Algorum user account id during your local testing and debugging. You can retrieve this using the user-profile CLI command. When you deploy your strategy to Algorum Cloud, this parameter will be automatically filled in by Algorum Cloud.
Receiving Symbol Data
You have to implement the method on_tick in your class, where you will receive the tick wise data of the symbols that you have registered to, during the strategy Initialization phase. This method is called both in real time mode (Paper Trading or Live Trading) and also Backtesting mode, so you can use same login to deal with both real time and historical backtesting scenarios.
# This method is called on each tick for the subscribed symbols
def on_tick(self, tick_data):
Placing orders
Within the on_tick, if the stock is not in bought state, you will have your Algo logic that retrieves the EMA 50 and 200 candles (each of 5 minutes) and will place orders when EMA 50 crosses above EMA 200. If the stock is bought, then we check if the price of the stock has gone up by 0.25% (take profit condition) or down by 0.5% (stop loss condition), and then place a sell order. For details on the place_order API you can refer to the API Reference section. Once the order is placed, you will get back an order id that you can use to track the order. At this point the order is NOT completed, it is just accepted by the brokerage or backtesting engine. You will receive order update events as the order passes through different phases until completion, rejection or cancellation.
# This method is called on each tick for the subscribed symbols
def on_tick(self, tick_data):
try:
self.State.CurrentTick = tick_data
ema50 = self.Evaluator.ema(50)
ema200 = self.Evaluator.ema(200)
if self.State.LastTick is not None and (
datetime.datetime.strptime(tick_data.Timestamp,
AlgorumQuantClient.quant_client.QuantEngineClient.get_date_format(tick_data.Timestamp)) -
datetime.datetime.strptime(self.State.LastTick.Timestamp,
AlgorumQuantClient.quant_client.QuantEngineClient.get_date_format(self.State.LastTick.Timestamp))).total_seconds() < 60:
pass
else:
msg = str(tick_data.Timestamp) + ',' + str(tick_data.LTP) + ', ema50 ' \
+ str(ema50) + ', ema200 ' + str(ema200)
print(msg)
self.log(AlgorumQuantClient.algorum_types.LogLevel.Information, msg)
self.State.LastTick = tick_data
if ema50 > 0 and ema200 > 0 and \
self.State.CrossAboveObj.evaluate(ema50, ema200) and \
not self.State.Bought and \
self.State.CurrentOrderId is None:
self.State.CurrentOrderId = uuid.uuid4().hex
place_order_request = AlgorumQuantClient.algorum_types.PlaceOrderRequest()
place_order_request.OrderType = AlgorumQuantClient.algorum_types.OrderType.Market
place_order_request.Price = tick_data.LTP
place_order_request.Quantity = \
(GoldenCrossoverQuantStrategy.Capital / tick_data.LTP) * GoldenCrossoverQuantStrategy.Leverage
place_order_request.Symbol = self.symbol
if self.LaunchMode == AlgorumQuantClient.algorum_types.StrategyLaunchMode.Backtesting:
place_order_request.TradeExchange = AlgorumQuantClient.algorum_types.TradeExchange.PAPER
else:
place_order_request.TradeExchange = AlgorumQuantClient.algorum_types.TradeExchange.NSE
place_order_request.OrderDirection = AlgorumQuantClient.algorum_types.OrderDirection.Buy
place_order_request.Tag = self.State.CurrentOrderId
self.place_order(place_order_request)
self.set_data("state", self.State)
msg = 'Placed buy order for ' + str(place_order_request.Quantity) + ' units of ' + self.symbol.Ticker + \
' at price (approx) ' + str(tick_data.LTP) + ', ' + str(tick_data.Timestamp)
print(msg)
self.log(AlgorumQuantClient.algorum_types.LogLevel.Information, msg)
else:
if self.State.CurrentOrder is not None and \
((tick_data.LTP - self.State.CurrentOrder.AveragePrice >= (
self.State.CurrentOrder.AveragePrice * (0.25 / 100))) or
(self.State.CurrentOrder.AveragePrice - tick_data.LTP >= (
self.State.CurrentOrder.AveragePrice * (0.5 / 100)))) and self.State.Bought:
qty = self.State.CurrentOrder.FilledQuantity
self.State.CurrentOrderId = uuid.uuid4().hex
place_order_request = AlgorumQuantClient.algorum_types.PlaceOrderRequest()
place_order_request.OrderType = AlgorumQuantClient.algorum_types.OrderType.Limit
place_order_request.Price = tick_data.LTP
place_order_request.Quantity = qty
place_order_request.Symbol = self.symbol
if self.LaunchMode == AlgorumQuantClient.algorum_types.StrategyLaunchMode.Backtesting:
place_order_request.TradeExchange = AlgorumQuantClient.algorum_types.TradeExchange.PAPER
else:
place_order_request.TradeExchange = AlgorumQuantClient.algorum_types.TradeExchange.NSE
place_order_request.TriggerPrice = tick_data.LTP
place_order_request.OrderDirection = AlgorumQuantClient.algorum_types.OrderDirection.Sell
place_order_request.Tag = self.State.CurrentOrderId
self.place_order(place_order_request)
self.set_data("state", self.State)
msg = 'Placed sell order for ' + str(qty) + ' units of ' + self.symbol.Ticker + \
' at price (approx) ' + str(tick_data.LTP) + ', ' + str(tick_data.Timestamp)
print(msg)
self.log(AlgorumQuantClient.algorum_types.LogLevel.Information, msg)
if self.LaunchMode == AlgorumQuantClient.algorum_types.StrategyLaunchMode.Backtesting:
self.send_progress_async(tick_data)
except Exception:
self.log(AlgorumQuantClient.algorum_types.LogLevel.Error, traceback.format_exc())
Handling Order Events
Once the order is placed, Algorum Quant Engine will send order events to your strategy, which can be handled in on_order_update method of your strategy class. This method gets called once for each state change in the Order. This method gets the Order object with the current order status, which you can examine and take appropriate action (mostly setting your state, calculating metrics, etc.,)
# This method is called on order updates, once the place_order method is called
def on_order_update(self, order: AlgorumQuantClient.algorum_types.Order):
try:
if order.Status == AlgorumQuantClient.algorum_types.OrderStatus.Completed:
self.StateLock.acquire()
self.State.Orders.append(order)
self.StateLock.release()
if order.OrderDirection == AlgorumQuantClient.algorum_types.OrderDirection.Buy:
self.State.Bought = True
self.State.CurrentOrder = order
msg = 'Order Id ' + order.OrderId + ' Bought ' + \
str(order.FilledQuantity) + ' units of ' + order.Symbol.Ticker + ' at price ' + \
str(order.AveragePrice)
print(msg)
self.log(AlgorumQuantClient.algorum_types.LogLevel.Information, msg)
else:
self.State.Bought = False
self.State.CurrentOrder = None
msg = 'Order Id ' + order.OrderId + ' Sold ' + \
str(order.FilledQuantity) + ' units of ' + order.Symbol.Ticker + ' at price ' + \
str(order.AveragePrice)
print(msg)
self.log(AlgorumQuantClient.algorum_types.LogLevel.Information, msg)
self.State.CurrentOrderId = None
stats = self.get_stats(self.State.CurrentTick)
self.publish_stats(stats)
for k, v in stats.items():
print('Key: ' + str(k) + ', Value: ' + str(v))
self.set_data("state", self.State)
except Exception:
self.log(AlgorumQuantClient.algorum_types.LogLevel.Error, traceback.format_exc())
Handling unsolicited errors
Your quant engine may send any errors that it encounters during order processing, tick processing, etc., to your strategy. You can override the QuantEngineClient method on_error and custom handle those errors. By default this method in the QuantEngineClient class will simply write the error message to the console.
def on_error(self, msg: str):
print(msg)
Logging important and debugging messages in your strategy
The log method of the QuantEngineClient base class will allow you to log important and debugging messages in your strategy. These logs are stored in Algorum Cloud and you can retrieve them by strategy and by date any time you need them. Algorum CLI also lets you see the log stream in real time when your strategy is running live or in backtesting mode, so you can keep an eye on what's going on. The log stream is explained in View real time Log Stream of your strategy section.
msg = 'Order Id ' + order.OrderId + ' Bought ' + \
str(order.FilledQuantity) + ' units of ' + order.Symbol.Ticker + ' at price ' + \
str(order.AveragePrice)
self.log(AlgorumQuantClient.algorum_types.LogLevel.Information, msg)
Publishing your strategy Metrics and Progress to Algorum Cloud
You can publish your strategy stats/metrics and progress (applicable in backtesting scenario) to Algorum Cloud. This will enable Algorum CLI show this information to you remotely. In future Algorum Web Apps will use this information to show visual progress and stats/metrics of your strategy.
Below method from our strategy will return the stats, which will then be used by QuantEngineClient base class to send these stats to the Algorum Cloud. As shown in the get_stats method, we are primarily computing P&L and overall portfolio value, and returning this information in a map object. You can add any stats/metrics you want to this map and these will be sent to Algorum Cloud and stored in a cloud storage, and sent to the listeners of your strategy (like Algorum CLI).
def get_stats(self, tick_date: AlgorumQuantClient.algorum_types.TickData):
stats_map = None
try:
stats_map = {"Capital": GoldenCrossoverQuantStrategy.Capital, "Order Count": len(self.State.Orders)}
buy_val = 0.0
sell_val = 0.0
buy_qty = 0.0
sell_qty = 0.0
for order in self.State.Orders:
if (order.Status == AlgorumQuantClient.algorum_types.OrderStatus.Completed) and \
(order.OrderDirection == AlgorumQuantClient.algorum_types.OrderDirection.Buy) and \
order.Symbol.Ticker == tick_date.Symbol.Ticker:
buy_val += order.FilledQuantity * order.AveragePrice
buy_qty += order.FilledQuantity
if (order.Status == AlgorumQuantClient.algorum_types.OrderStatus.Completed) and \
(order.OrderDirection == AlgorumQuantClient.algorum_types.OrderDirection.Sell) and \
order.Symbol.Ticker == tick_date.Symbol.Ticker:
sell_val += order.FilledQuantity * order.AveragePrice
sell_qty += order.FilledQuantity
if sell_qty < buy_qty:
sell_val += (buy_qty - sell_qty) * tick_date.LTP
pl = sell_val - buy_val
stats_map['PL'] = pl
stats_map['Portfolio Value'] = GoldenCrossoverQuantStrategy.Capital + pl
except Exception:
self.log(AlgorumQuantClient.algorum_types.LogLevel.Error, traceback.format_exc())
return stats_map
Managing strategy state
You can write your strategy state to the Algorum Cloud using the set_data method. This allows you to save the state in case you have to stop and start your strategy for any reasons like restarting your Algorum Quant Engine for upgrades, Algorum Cloud automatically stopping your strategy at the end of the market hours to optimize the infrastructure cost and efficiently using the resources, etc.,. You can get back your state during the initialization of your strategy using the get_data method.
Loading state
def __init__(self, url, apikey, launchmode, sid):
try:
...
# Load any saved state
state_json_str = self.get_data("state")
if state_json_str is not None:
self.State = jsonpickle.decode(state_json_str)
...
except Exception:
print(traceback.format_exc())
self.log(AlgorumQuantClient.algorum_types.LogLevel.Error, traceback.format_exc())
Saving state
# This method is called on each tick for the subscribed symbols
def on_tick(self, tick_data):
try:
self.State.CurrentTick = tick_data
...
self.set_data("state", self.State)
...
except Exception:
self.log(AlgorumQuantClient.algorum_types.LogLevel.Error, traceback.format_exc())
Proceed to next section to learn how to use our strategy class to backtest locally, and for further steps.
Updated almost 3 years ago