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:
Trades on ATOM/OSMO pair on Osmosis
Make trade decisions based hourly OHCLV candles
Uses simple fast/slow EMA crossover strategy
Set up#
Set up strategy paramets that will decide its behavior
[1]:
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.
[2]:
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)
else:
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.
[3]:
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, \
create_pair_universe_from_code
def create_trading_universe() -> TradingStrategyUniverse:
# Set up fake assets
mock_chain_id = ChainId.osmosis
mock_exchange = generate_exchange(
exchange_id=random.randint(1, 1000),
chain_id=mock_chain_id,
address=generate_random_ethereum_address())
# 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(
atom,
osmo,
generate_random_ethereum_address(),
mock_exchange.address,
internal_id=random.randint(1, 1000),
internal_exchange_id=mock_exchange.exchange_id,
fee=0.0005
)
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(
ohlcv_sample_path,
mock_chain_id,
mock_exchange.exchange_id,
atom_osmo.internal_id)
# Does not work properly for Osmosis data
del candles["volume"]
candle_universe = GroupedCandleUniverse.create_from_single_pair_dataframe(candles)
universe = Universe(
time_bucket=time_bucket,
chains={mock_chain_id},
exchanges={mock_exchange},
pairs=pair_universe,
candles=candle_universe,
liquidity=None
)
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.
[4]:
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",
start_at=start_at,
end_at=end_at,
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
decide_trades=decide_trades,
universe=universe,
initial_deposit=10_000,
reserve_currency=ReserveCurrency.busd,
trade_routing=TradeRouting.user_supplied_routing_model,
routing_model=routing_model,
log_level=logging.WARNING,
)
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
[5]:
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
[6]:
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/
figure.show()