Tags: synthetic-data, ema, trend-analysis

Synthetic data backtesting example#

This is an example notebook how to create and run backtests with tradeexecutor framework.

Some highlights of this notebook:

Runs everything within a single notebook
- The backtest code and charts are self-contained in a single file
- The example code is easy to read
- Easy to test different functionalities of tradeexecutor library
Uses generated, synthetic, random price data
- Notebook runs offline
- No downloads needed
- No API keys needed
- Running the notebook completes quickly, making it suitable for low powered devices and demos

Set up#

Set up strategy paramets that will decide its behavior

[1]:

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

trading_strategy_cycle = CycleDuration.cycle_1d

# Strategy keeps its cash in BUSD
reserve_currency = ReserveCurrency.busd

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

#
# Strategy thinking specific parameter
#

slow_ema_candle_count = 20

fast_ema_candle_count = 5

# How many candles to extract from the dataset once
batch_size = 90

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

start_at = datetime.datetime(2021, 6, 1)
start_at_data = datetime.datetime(2021, 1, 1)
end_at = datetime.datetime(2022, 1, 1)

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()

    # 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 fast_ema >= slow_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]:

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
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.ethereum
    mock_exchange = generate_exchange(
        exchange_id=random.randint(1, 1000),
        chain_id=mock_chain_id,
        address=generate_random_ethereum_address())
    usdc = AssetIdentifier(ChainId.ethereum.value, generate_random_ethereum_address(), "USDC", 6, 1)
    weth = AssetIdentifier(ChainId.ethereum.value, generate_random_ethereum_address(), "WETH", 18, 2)
    weth_usdc = TradingPairIdentifier(
        weth,
        usdc,
        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, [weth_usdc])

    candles = generate_ohlcv_candles(time_bucket, start_at_data, end_at, pair_id=weth_usdc.internal_id)
    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=[usdc])

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="Synthetic random data backtest",
    start_at=start_at,
    end_at=end_at,
    client=None,  # None of downloads needed, because we are using synthetic data
    cycle_duration=trading_strategy_cycle,
    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,
)

Our universe has synthetic candle data for the period 2021-01-01 00:00:00 - 2021-12-31 00: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()))}")

Positions taken: 82
Trades made: 163

[6]:

from tradeexecutor.visual.single_pair import visualise_single_pair

figure = visualise_single_pair(state, universe.universe.candles)

figure.show()