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.
https://github.com/Algorum/AlgorumStrategySamplesPython3

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.

📘Golden Crossover Strategy Python 3+ Sample on GitHub

Project Setup

Strategy Class Setup

Strategy Initialization

Receiving Symbol Data

Placing orders

Handling Order Events

Handling unsolicited errors

Logging important and debugging messages in your strategy

Publishing your strategy Metrics and Progress to Algorum Cloud

Managing strategy state

Loading state

Saving state

📘
Golden Crossover Strategy Python 3+ Sample on GitHub