Super Algorithm

Appearance

Writing Your First Trading Strategy ​

This guide will walk you through creating a simplified trading strategy using Super Algorithm, demonstrating the core concepts and implementation details. We recommend you read this guide, and the rest of the documentation to get a full understanding of the framework and it's capabilities.

Prerequisites ​

We will be using talipp for computing indicators:

bash
pip install talipp

Creating Your Strategy ​

Create a new file named sma_sample_strategy.py and implement your strategy following these steps:

1. Set Up the Basic Structure ​

First, import the necessary modules and create your strategy class:

python
from superalgorithm.strategy import BaseStrategy
from superalgorithm.types import Bar, OrderType, PositionType
from talipp.indicators import SMA

class SMAStrategy(BaseStrategy):
    pass

2. Initialize Your Strategy ​

Implement the init method to:

  • Set up event handlers for data aggregates
  • Initialize your indicators
python
def init(self):
    # Listen for 5-minute interval events
    self.on("5m", self.on_5)

    # Initialize Simple Moving Average indicators
    self.sma = SMA(14)        # Fast SMA with 14 periods
    self.sma_slow = SMA(200)  # Slow SMA with 200 periods

3. Handle Time Intervals ​

There are two ways to receive data in your strategy:

  • The on_tick(bar: Bar) method receives updates from all configured DataSources, including partial updates for Bars.
  • The on(<timeframe>, bar: Bar) method receives updates from all configured DataSources for each completed bar.

For this example implement the on_5 method to process 5-minute bars.

python
async def on_5(self, bar: Bar):
    # Update indicators with new data
    self.sma.add(bar.close)
    self.sma_slow.add(bar.close)

    # Execute trading logic
    await self.trade_logic()

TIP

Since we are using only one DataSource, we know that the bar will come from that source. If you are using multiple DataSources, you could check bar.source_id to identify which bar triggered the event.

4. Implement Trading Logic ​

Create the trade_logic method to:

  • Check for sufficient data
  • Generate buy/sell signals
  • Execute trades

The BaseStrategy comes with two convenience functions to access your data at any point:

  • The current bars' ohlcv: self.get("BTC/USDT", "5m") -> OHLCV
  • List of ohlcv values including the current bars' value: self.data("BTC/USDT", "5m") -> List[OHLCV]
python
async def trade_logic(self):
    # Ensure we have enough data and our indicators are populated
    if len(self.sma_slow) < 2 or self.sma_slow[-2] is None:
        return

    close = self.get("BTC/USDT", "5m").close

    # Generate signals
    buy = self.sma[-2] < self.sma_slow[-2] and self.sma[-1] > self.sma_slow[-1]
    sell = self.sma[-2] > self.sma_slow[-2] and self.sma[-1] < self.sma_slow[-1]

    # Execute trades based on signals
    if buy:
        await self.open("BTC/USDT", PositionType.LONG, 0.1, OrderType.LIMIT, close)
    if sell:
        await self.close("BTC/USDT", PositionType.LONG, 0.1, OrderType.LIMIT, close)

TIP

When submitting trades, we refer to either opening or closing a position. In the example above, we open a long position and then close a long position. In the same context you could open and close a short position. You can also open a long and short position at the same time, given your exchange supports this.

5. Implement Required Methods ​

Add the required on_tick method, even if unused:

python
async def on_tick(self, bar: Bar):
    pass

Complete Implementation ​

Here's the complete strategy implementation:

python
from superalgorithm.strategy import BaseStrategy
from superalgorithm.types import Bar, OrderType, PositionType
from talipp.indicators import SMA

class SMAStrategy(BaseStrategy):

    def init(self):
        self.on("5m", self.on_5)
        self.sma = SMA(14)
        self.sma_slow = SMA(200)

    async def on_5(self, bar: Bar):
        self.sma.add(bar.close)
        self.sma_slow.add(bar.close)
        await self.trade_logic()

    async def on_tick(self, bar: Bar):
        pass

    async def trade_logic(self):
        if len(self.sma_slow) < 2 or self.sma_slow[-2] is None:
            return

        close = self.get("BTC/USDT", "5m").close

        buy = self.sma[-2] < self.sma_slow[-2] and self.sma[-1] > self.sma_slow[-1]
        sell = self.sma[-2] > self.sma_slow[-2] and self.sma[-1] < self.sma_slow[-1]

        if buy:
            await self.open("BTC/USDT", PositionType.LONG, 0.1, OrderType.LIMIT, close)
        if sell:
            await self.close("BTC/USDT", PositionType.LONG, 0.1, OrderType.LIMIT, close)