Tags: ema, trend-analysis, synthetic-data, dark-theme

Osmosis ATOM-OSMO mean reversion strategy#

This is an example notebook how to backtest trading strategies on Osmosis Cosmos DEX. It is based on work done in HackAtom Seoul 2022 hackathon.

Some highlights of this notebook:

Set up#

Set up strategy paramets that will decide its behavior

import sys

import datetime
import logging

import pandas as pd

from tradingstrategy.chain import ChainId
from tradingstrategy.timebucket import TimeBucket
from tradeexecutor.strategy.cycle import CycleDuration
from tradeexecutor.strategy.strategy_module import TradeRouting, ReserveCurrency

# Rebalance every 8h
trading_strategy_cycle = CycleDuration.cycle_8h

# How much of the cash to put on a single trade
position_size = 0.90

# Strategy thinking specific parameter

# 14 days
slow_ema_candle_count = 14*24

# 5 days
fast_ema_candle_count = 5*24

# How many candles to extract from the dataset once
batch_size = slow_ema_candle_count * 2

# Range of backtesting and synthetic data generation.
# Because we are using synthetic data actual dates do not really matter -
# only the duration

# Osmosis launched at 2021-12-25
start_at = datetime.datetime(2022, 1, 25)

# When our data ends
end_at = datetime.datetime(2023, 1, 25)

Strategy logic and trade decisions#

decide_trades function decide what trades to take. In this example, we calculate two exponential moving averages (EMAs) and make decisions based on those.

from typing import List, Dict

from pandas_ta.overlap import ema

from tradingstrategy.universe import Universe

from tradeexecutor.state.visualisation import PlotKind
from tradeexecutor.state.trade import TradeExecution
from tradeexecutor.strategy.pricing_model import PricingModel
from tradeexecutor.strategy.pandas_trader.position_manager import PositionManager
from tradeexecutor.state.state import State

def decide_trades(
        timestamp: pd.Timestamp,
        universe: Universe,
        state: State,
        pricing_model: PricingModel,
        cycle_debug_data: Dict) -> List[TradeExecution]:
    """The brain function to decide the trades on each trading strategy cycle."""

    # The pair we are trading
    pair = universe.pairs.get_single()

    assert pair.token0_symbol == "ATOM", f"Got pair {pair}"
    assert pair.token1_symbol == "OSMO", f"Got pair {pair}"

    # How much cash we have in the hand
    cash = state.portfolio.get_current_cash()

    # Get OHLCV candles for our trading pair as Pandas Dataframe.
    # We could have candles for multiple trading pairs in a different strategy,
    # but this strategy only operates on single pair candle.
    # We also limit our sample size to N latest candles to speed up calculations.
    candles: pd.DataFrame = universe.candles.get_single_pair_data(timestamp, sample_count=batch_size)

    # We have data for open, high, close, etc.
    # We only operate using candle close values in this strategy.
    close = candles["close"]

    # Calculate exponential moving averages based on slow and fast sample numbers.
    # https://github.com/twopirllc/pandas-ta
    # https://github.com/twopirllc/pandas-ta/blob/bc3b292bf1cc1d5f2aba50bb750a75209d655b37/pandas_ta/overlap/ema.py#L7
    slow_ema_series = ema(close, length=slow_ema_candle_count)
    fast_ema_series = ema(close, length=fast_ema_candle_count)

    if slow_ema_series is None or fast_ema_series is None:
        # Cannot calculate EMA, because
        # not enough samples in backtesting
        return []

    slow_ema = slow_ema_series.iloc[-1]
    fast_ema = fast_ema_series.iloc[-1]

    # Get the last close price from close time series
    # that's Pandas's Series object
    # https://pandas.pydata.org/docs/reference/api/pandas.Series.iat.html
    current_price = close.iloc[-1]

    # List of any trades we decide on this cycle.
    # Because the strategy is simple, there can be
    # only zero (do nothing) or 1 (open or close) trades
    # decides
    trades = []

    # Create a position manager helper class that allows us easily to create
    # opening/closing trades for different positions
    position_manager = PositionManager(timestamp, universe, state, pricing_model)

    if not position_manager.is_any_open():
        if current_price >= slow_ema:
            # Entry condition:
            # Close price is higher than the slow EMA
            buy_amount = cash * position_size
            trades += position_manager.open_1x_long(pair, buy_amount)
        if slow_ema >= fast_ema:
        # Exit condition:
        # Fast EMA crosses slow EMA
            trades += position_manager.close_all()

    # Visualize strategy
    # See available Plotly colours here
    # https://community.plotly.com/t/plotly-colours-list/11730/3?u=miohtama
    visualisation = state.visualisation
    visualisation.plot_indicator(timestamp, "Slow EMA", PlotKind.technical_indicator_on_price, slow_ema, colour="darkblue")
    visualisation.plot_indicator(timestamp, "Fast EMA", PlotKind.technical_indicator_on_price, fast_ema, colour="#003300")

    return trades

Defining trading universe#

We create a trading universe with a single blockchain, exchange and trading pair. For the sake of easier understanding the code, we name this “Uniswap v2” like exchange with a single ETH-USDC trading pair.

The trading pair contains generated noise-like OHLCV trading data.


from pathlib import Path
import os
import random
from tradeexecutor.state.identifier import AssetIdentifier, TradingPairIdentifier
from tradingstrategy.candle import GroupedCandleUniverse
from tradeexecutor.testing.synthetic_ethereum_data import generate_random_ethereum_address
from tradeexecutor.testing.synthetic_exchange_data import generate_exchange
from tradeexecutor.testing.synthetic_price_data import generate_ohlcv_candles, load_ohlcv_parquet_file
from tradeexecutor.strategy.trading_strategy_universe import TradingStrategyUniverse, \

def create_trading_universe() -> TradingStrategyUniverse:

    # Set up fake assets
    mock_chain_id = ChainId.osmosis
    mock_exchange = generate_exchange(
        exchange_id=random.randint(1, 1000),
    # Cosmos tokens use micro token (u-token) as the smallest unit
    osmo = AssetIdentifier(ChainId.osmosis.value, generate_random_ethereum_address(), "OSMO", 6, 1)
    atom = AssetIdentifier(ChainId.osmosis.value, generate_random_ethereum_address(), "ATOM", 6, 2)
    atom_osmo = TradingPairIdentifier(
        internal_id=random.randint(1, 1000),

    time_bucket = TimeBucket.d1

    pair_universe = create_pair_universe_from_code(mock_chain_id, [atom_osmo])

    # Hack
    import tradeexecutor
    ohlcv_sample_path = os.path.join(os.path.dirname(tradeexecutor.__file__), "..", "notebooks", "pool_1_hourly_candles.parquet")

    ohlcv_sample_path = Path(ohlcv_sample_path)

    print(f"Loading data from {ohlcv_sample_path.as_posix()}")

    # Load candles for backtesting
    candles = load_ohlcv_parquet_file(

    # Does not work properly for Osmosis data
    del candles["volume"]

    candle_universe = GroupedCandleUniverse.create_from_single_pair_dataframe(candles)

    universe = Universe(

    return TradingStrategyUniverse(universe=universe, reserve_assets=[osmo])

Running the backtest#

Run backtest using giving trading universe and strategy function.

Running the backtest outputs state object that contains all the information on the backtesting position and trades.

from tradeexecutor.testing.synthetic_exchange_data import generate_simple_routing_model
from tradeexecutor.backtest.backtest_runner import run_backtest_inline

universe = create_trading_universe()

start_candle, end_candle = universe.universe.candles.get_timestamp_range()
print(f"Our universe has synthetic candle data for the period {start_candle} - {end_candle}")

# This function set ups trade routing for our synthetic trading universe.
# Because we have only one trading pair, there is no complicated
# routing needed
routing_model = generate_simple_routing_model(universe)

state, universe,    debug_dump = run_backtest_inline(
    name="ATOM/OSMO backtest",
    client=None,  # None of downloads needed, because we are using synthetic data
    cycle_duration=CycleDuration.cycle_1d,  # Override to use 24h cycles despite what strategy file says

Loading data from /home/alex/work/tradingstrategy-ai/stuff/docs/deps/trade-executor/tradeexecutor/../notebooks/pool_1_hourly_candles.parquet
Our universe has synthetic candle data for the period 2021-12-25 00:00:00 - 2022-04-28 12:00:00

Examine backtest results#

Examine state that contains all actions the trade executor took.

We plot out a chart that shows - The price action - When the strategy made buys or sells

print(f"Positions taken: {len(list(state.portfolio.get_all_positions()))}")
print(f"Trades made: {len(list(state.portfolio.get_all_trades()))}")
print(f"Visualisation entries: {len(list(state.visualisation.plots))}")
Positions taken: 1
Trades made: 2
Visualisation entries: 2
from tradeexecutor.visual.single_pair import visualise_single_pair

figure = visualise_single_pair(state, universe.universe.candles)
figure.update_layout(template="plotly_dark")  # Dark color theme https://plotly.com/python/templates/