Backtesting
Before putting your code onto Cybotrade Cloud, you have the ability to backtest it locally on your machine.
Do note that there is no 'multi-exchange' support in Backtesting. All exchange-credentials, and API exchange parameters will be ignored.
The following code sample gives you an overview on how to backtest your strategy locally.
from cybotrade.strategy import Strategy as BaseStrategy
from cybotrade.models import (
RuntimeConfig,
RuntimeMode,
)
from cybotrade.permutation import Permutation
from datetime import datetime, timezone
import asyncio
import colorlog
import logging
# This is the strategy class you've written.
class Strategy(BaseStrategy):
def __init__(self):
handler = colorlog.StreamHandler()
handler.setFormatter(
colorlog.ColoredFormatter(f"%(log_color)s{Strategy.LOG_FORMAT}")
)
file_handler = logging.FileHandler("example.log")
file_handler.setLevel(logging.INFO)
super().__init__(log_level=logging.INFO, handlers=[handler, file_handler])
async def on_candle_closed(self, strategy, topic, symbol):
logging.info(f"{topic} = {super().data_map[topic][-1]}")
async def on_datasource_interval(self, strategy, topic, data_list):
logging.info(f"{topic} = {super().data_map[topic][-1]}")
permutation = Permutation(RuntimeConfig(
mode=RuntimeMode.Backtest,
candle_topics=["candles-1d-BTC/USDT-bybit", "candles-1d-AVAX/USDT-bybit"],
datasource_topics=["coinglass|1m|futures/fundingRate/ohlc-history?exchange=Binance&symbol=BTCUSDT&interval=1h", "coinglass|1m|futures/openInterest/ohlc-history?exchange=Binance&symbol=BTCUSDT&interval=4h"],
active_order_interval=1,
start_time=datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
end_time=datetime(2023, 1, 5, 0, 0, 0, tzinfo=timezone.utc),
data_count=1000,
api_key="YOUR_CYBOTRADE_API_KEY",
api_secret="YOUR_CYBOTRADE_API_SECRET",
))
hyper_parameters = {}
async def start():
await permutation.run(hyper_parameters, Strategy)
asyncio.run(start())The best part? changing this to Live or LiveTestnet is as simple as changing the mode value to RuntimeMode.Live/RuntimeMode.LiveTestnet and setting up your
exchange credentials following this guide!
Permutation
When performing backtests, you are able to leverage the powerful capabialities of Hyper-parameter tuning with the use of the Permutation class.
A simple strategy example of this is if you were to have a Strategy that relies on SMA and RSI but you'd like to see the results of different SMA and RSI values.
Attached below is an example of the strategy mentioned above:
from datetime import datetime, timezone
from cybotrade.strategy import Strategy as BaseStrategy
from cybotrade.runtime import Runtime
from cybotrade.models import (
Exchange,
OrderSide,
RuntimeConfig,
RuntimeMode,
)
from cybotrade.permutation import Permutation
import talib
import numpy as np
import asyncio
import logging
class Strategy(BaseStrategy):
async def set_param(self, identifier, value):
logging.info(f"Setting {identifier} to {value}")
if identifier == "sma":
self.sma = int(value)
elif identifier == "rsi":
self.rsi = int(value)
else:
logging.error(f"Could not set {identifier}, not found")
async def on_candle_closed(self, strategy, topic, symbol):
# Retrieve list of candle data for corresponding symbol that candle closed.
candles = super().data_map[topic];
# Retrieve close data from list of candle data.
close = np.array(list(map(lambda c: float(c["close"]), candles)))
sma = talib._ta_lib.SMA(close, self.sma)
rsi = talib._ta_lib.RSI(close, 10)
price_changes = (float(candles[-1]["close"]) / sma[-1]) - 1.0
current_pos = await strategy.position(exchange=Exchange.BybitLinear, symbol=symbol)
if price_changes > 0.045 and rsi[-1] > self.rsi:
if current_pos.long.quantity == 0.0:
try:
await strategy.open(
exchange=Exchange.BybitLinear,
side=OrderSide.Buy,
quantity=1,
symbol=symbol,
limit=None,
take_profit=None,
stop_loss=None,
is_hedge_mode=False
is_post_only=False
)
except Exception as e:
logging.error(f"Failed to open long: {e}")
else:
if current_pos.long.quantity != 0.0:
try:
await strategy.close(
exchange=Exchange.BybitLinear,
side=OrderSide.Buy,
quantity=current_pos.long.quantity,
symbol=symbol
is_hedge_mode=False
is_post_only=False
)
except Exception as e:
logging.error(f"Failed to close entire position: {e}")
config = RuntimeConfig(
mode=RuntimeMode.Backtest,
datasource_topics=[],
active_order_interval=1,
initial_capital=10000000.0,
candle_topics=["candles-1d-BTC/USDT-bybit"],
start_time=datetime(2020, 5, 11, 0, 0, 0, tzinfo=timezone.utc),
end_time=datetime(2024, 1, 5, 0, 0, 0, tzinfo=timezone.utc),
api_key="YOUR_CYBOTRADE_API_KEY",
api_secret="YOUR_CYBOTRADE_API_SECRET",
data_count=40,
)
permutation = Permutation(config)
hyper_parameters = {}
hyper_parameters["rsi"] = [10, 20, 30]
hyper_parameters["sma"] = [39, 40, 41]
async def start_backtest():
await permutation.run(hyper_parameters, Strategy)
asyncio.run(start_backtest())It is recommended to use the Permutation class to run a Backtest even with a single parameter. Such a Backtest Strategy will be ran once. An example of achieving this is shown below.
permutation = Permutation(config)
hyper_parameters = {}
hyper_parameters["rsi"] = [10]
hyper_parameters["sma"] = [39]
async def start_backtest():
await permutation.run(hyper_parameters, Strategy)
asyncio.run(start_backtest())When attempting to start a Strategy you would typically define a function similar to the one shown below. This is used when you want to run the Backtest once or run the strategy as Live or LiveTestnet.
async def start_strategy():
runtime = await Runtime.connect(config, Strategy())
await runtime.start()
asyncio.run(start_strategy())However, with Permutations you will have to create a new Permutation class first.
permutation = Permutation(config)The Permutation class only has one useful function which is .run(). Notice in the example below that we define a Dictionary with key value pairs which are your hyper-parameters.
hyper_parameters = {}
hyper_parameters["rsi"] = [10, 20, 30]
hyper_parameters["sma"] = [39, 40, 41]
async def start_backtest():
await permutation.run(hyper_parameters, Strategy)We will be passing this Dictionary into the .run() along with the runtime value that is returned from the .connect() method as defined in the start_backtest() function.
This will run all the hyper parameters with the Strategy.
Assigning the hyper_parameters occur with the set_param() handler. For an example, setting the rsi and sma can be done as such
async def set_param(identifier, value):
if identifier == "rsi":
self.rsi = int(value)
elif identifier == "sma":
self.sma = init(value)
else:
raise Exception("Unrecognized identifier")The set_param() handler will be called by the Permutation class and your rsi/sma values will be set for you. This is a dynamic field assignment.
Therefore, if you'd like to assign any type of parameters you may do so with 'self.{parameter_identifier}'.
For an example:
async def set_param(identifier, value):
if identifier == "weather":
self.weather = value
elif identifier == "water temperature":
self.water_temp = value
...Upon completion of the backtest you may run the following line in your terminal
ls | grep "performance_*"You should be able to see a performance_YYYYMMDD-HHMMSSmmm.json file.
For an example:
performance_20240215-171002942.jsonthe performance json file will be placed in the directory that you ran the backtest in.