The second stage of backtesting a python trading robot. I’m adding methods to allow the same trading robot to run in both a backtest mode and to run live against Metatrader 5.
My approach involves using child classes of the Backtest and Strategy classes, to intercept the trading actions and direct calls to either backtesting or Metatrader.
I began by copying the files from the last part where I introduced backtesting and just renaming files as MA Cross, then adding the file for the new classes
ma_cross.py
classes\
ma_cross_strategy.py
custom_backtesting.py
In custom_backtesting.py I created 2 classes, CustomBacktest and CustomStrategy. There are some imports at the beginning of the file
import MetaTrader5 as mt5 from datetime import datetime import pandas as pd import time from typing import Optional, Tuple, Type, Union from backtesting import Backtest, Strategy
CustomBacktest class
Then add the CustomBacktest class inheriting from Backtest and a constructor that sets up testing mode and fills with dummy data in live mode, finally calling the parent constructor
class CustomBacktest(Backtest):
def __init__(self,
data: pd.DataFrame,
strategy: Type[Strategy],
*,
cash: float = 10_000,
spread: float = .0,
commission: Union[float, Tuple[float, float]] = .0,
margin: float = 1.,
trade_on_close=False,
hedging=False,
exclusive_orders=False,
finalize_trades=False,
):
self.is_testing = True
# create a blank data set if needed
if data is None:
columns = ['Time', 'Open', 'High', 'Low', 'Close', 'Volume']
values = {col: [0] for col in columns}
values['Time'] = [datetime.now()]
data = pd.DataFrame(values)
data.set_index('Time', inplace=True)
super().__init__(data, strategy,
cash=cash,
spread=spread,
commission=commission,
margin=margin,
trade_on_close=trade_on_close,
hedging=hedging,
exclusive_orders=exclusive_orders,
finalize_trades=finalize_trades)
To switch into live mode add a function
def set_live_params(self, *, symbol='', timeframe=mt5.TIMEFRAME_H1, cycle=60):
self.is_testing = False
self.symbol = symbol
self.timeframe = timeframe
self.cycle = cycle
Then override the run function from the parent class if running in live mode, or just call the parent class if in test mode
def run(self, **kwargs) -> pd.Series:
if self.is_testing:
return super().run(**kwargs)
if not mt5.symbol_select(self.symbol):
return None
strategy: Strategy = self._strategy(None, self._data, kwargs)
strategy.set_live_params(symbol=self.symbol, timeframe=self.timeframe)
strategy.init()
while True:
strategy.next()
time.sleep(self.cycle)
return None
CustomStrategy class
The constructor also sets up testing mode and just calls the parent constructor
class CustomStrategy(Strategy):
def __init__(self, broker, data, params):
super().__init__(broker, data, params)
self.is_testing = True
There is also a function to set live mode
def set_live_params(self, *, symbol='', timeframe=mt5.TIMEFRAME_H1):
self.is_testing = False
self.symbol = symbol
self.timeframe = timeframe
The implementation of the strategy is in the MACross class which currently inherits from the Strategy class. That will change to inherit from CustomStrategy but the MACross class calls some functions to get trade information and to execute buy and sell. These functions need to be handled differently in CustomStrategy depending on if they are live or test
def buy(self, *,
size: float = 1.0,
limit: Optional[float] = None,
stop: Optional[float] = None,
sl: Optional[float] = None,
tp: Optional[float] = None,
tag: object = None) -> 'Order':
if self.is_testing:
return super().buy(size=size, limit=limit, stop=stop, sl=sl, tp=tp, tag=tag)
return self.open_position(mt5.ORDER_TYPE_BUY, size, sl, tp, 0, 0)
def sell(self, *,
size: float = 1.0,
limit: Optional[float] = None,
stop: Optional[float] = None,
sl: Optional[float] = None,
tp: Optional[float] = None,
tag: object = None) -> 'Order':
if self.is_testing:
return super().sell(size=size, limit=limit, stop=stop, sl=sl, tp=tp, tag=tag)
return self.open_position(mt5.ORDER_TYPE_SELL, size, sl, tp, 0, 0)
def open_position(self, type, size, sl, tp, deviation, magic):
request = self.get_request(type, size, sl, tp, deviation, magic)
if request is None:
return
result = mt5.order_send(request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
print(f"order send failed with {result.retcode}")
print(result)
# return False
else:
print("we had success")
print(result)
def get_request(self, type, size, sl, tp, deviation, magic):
# current price information
price_info = mt5.symbol_info_tick(self.symbol)
if price_info is None:
print(f"Failed to get price information for {self.symbol}")
return None
if type == mt5.ORDER_TYPE_BUY:
price = price_info.ask
else:
price = price_info.bid
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": self.symbol,
"magic": magic,
"volume": size,
"type": type,
"price": price,
"sl": sl,
"tp": tp,
"deviation": deviation,
"type_time": mt5.ORDER_TIME_GTC,
}
if not self.set_filling_mode(request):
return None
return request
def set_filling_mode(self, request):
for filling_mode in range(2):
request['type_filling'] = filling_mode
result = mt5.order_check(request)
if result.comment == "Done":
return True
return False
def get_rates_from_pos(self, symbol, timeframe, bar_count):
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, bar_count)
if rates is None:
print('No rates data retrieved')
return None
rates_frame = pd.DataFrame(rates)
rates_frame.rename(columns={'open':'Open', 'high':'High', 'low':'Low', 'close':'Close', 'tick_volume':'Volume'}, inplace=True)
return rates_frame
# common functions that will take different paths on test
def get_open_trade_count(self, type):
open_trades = 0
if self.is_testing:
if type==mt5.ORDER_TYPE_BUY:
open_trades = sum(1 for trade in self.trades if trade.is_long)
elif type==mt5.ORDER_TYPE_SELL:
open_trades = sum(1 for trade in self.trades if trade.is_long)
return open_trades
positions = mt5.positions_get(symbol=self.symbol)
if positions == None:
return 0
open_trades = sum(1 for position in positions if position['type']==type)
return open_trades
ma_cross.py
Here I have to setup the mode as testing or live. I also replaced references to Backtest with CustomBacktest and added some constants to use for flagging test mode and setting the environment for live trading
Imports and constants
import MetaTrader5 as mt5 import pandas as pd import datetime # from backtesting import Backtest from classes.custom_backtesting import CustomBacktest from classes.ma_cross_strategy import MACrossStrategy TESTING=False SYMBOL = 'EURUSD' TIMEFRAME = mt5.TIMEFRAME_M1 CYCLE = 1
Change the section setting up history to only build the test data is in testing mode, only print test results in testing mode, and replace the Backtest call with CustomBacktest
def main():
if not mt5.initialize():
log("terminal initialisation failed")
return
log("MT5 successfully initialised")
# get history data from mt, only using one instrument / tf for now
if TESTING:
history = pd.DataFrame(mt5.copy_rates_from_pos("EURUSD", mt5.TIMEFRAME_H1, 0, 1000))
history['time'] = pd.to_datetime(history['time'], unit='s')
history.set_index('time', inplace=True)
history.rename(columns={'open':'Open', 'high':'High', 'low':'Low', 'close':'Close', 'tick_volume':'Volume'}, inplace=True)
print(history)
else:
history = None
test = CustomBacktest(history, MACrossStrategy, cash=10000, hedging=True, finalize_trades=True)
if not TESTING:
test.set_live_params(symbol=SYMBOL, timeframe=TIMEFRAME, cycle=CYCLE)
result = test.run()
if TESTING:
print(result)
print(f'buy count = {result._strategy.buy_count}')
print(f'sell count = {result._strategy.sell_count}')
mt5.shutdown()
return
ma_cross_strategy.py
The code changes in ma_cross_strategy.py are untidy and will be improved in future versions.
First change the imports, note that I’ve used pandas_ta here so listen to the warnings in this and the last video.
import MetaTrader5 as mt5 import pandas as pd import pandas_ta as ta # take care with this from backtesting._util import _Data from classes.custom_backtesting import CustomStrategy
Then modify init to only calculate indicators if in testing mode
def init(self):
self.fast_ma_period = 5
self.slow_ma_period = 10
self.lot_size = 0.1
self.stop_loss_amount = 0.00100
self.take_profit_amount = 0.00150
self.buy_count = 0
self.sell_count = 0
# Calculate fast and slow moving averages
if self.is_testing:
self.fast_ma = self.I(SMA, self.data.Close, self.fast_ma_period)
self.slow_ma = self.I(SMA, self.data.Close, self.slow_ma_period)
Change the next function to calculate indicators and fill data on each call in live mode, and use a function to count the number of open trades
def next(self):
# get fresh data if in live mode
if not self.is_testing:
bar_count = self.fast_ma_period + self.slow_ma_period + 3
rates = self.get_rates_from_pos(self.symbol, self.timeframe, bar_count)
if rates is None:
return
self._data = _Data(rates.copy(deep=False))
self.fast_ma = SMA(rates['Close'], self.fast_ma_period)
self.slow_ma = SMA(rates['Close'], self.slow_ma_period)
# Check for crossover
if self.fast_ma[-1] < self.slow_ma[-1] and self.fast_ma[-2] >= self.slow_ma[-2]:
# Fast MA crosses above Slow MA: Buy signal
open_trades = self.get_open_trade_count(mt5.ORDER_TYPE_BUY)
if open_trades > 0:
return
stop_loss_price = self.data.Close[-1]-self.stop_loss_amount
take_profit_price = self.data.Close[-1]+self.take_profit_amount
self.buy(size = self.lot_size, sl = stop_loss_price, tp = take_profit_price)
self.buy_count += 1
elif self.slow_ma[-1] < self.fast_ma[-1] and self.slow_ma[-2] >= self.fast_ma[-2]:
# Slow MA crosses above Fast MA: Sell signal
open_trades = self.get_open_trade_count(mt5.ORDER_TYPE_SELL)
if open_trades > 0:
return
stop_loss_price = self.data.Close[-1]+self.stop_loss_amount
take_profit_price = self.data.Close[-1]-self.take_profit_amount
self.sell(size = self.lot_size, sl = stop_loss_price, tp = take_profit_price)
self.sell_count += 1
Finally I changed the SMA function from last time to now use pandas_ta. Remember that at the time of recording pandas_ta does not work with versions of numpy from 2 onward so I created my own modified but untested version which you can install with the supplied commands, at your own risk, if you wish.
# Define Simple Moving Average (SMA) function
def SMA(data, period):
return ta.sma(pd.Series(data), length = period).to_numpy()
# pip uninstall pandas_ta
# pip install -U git+https://github.com/OrchardForexTutorials/pandas-ta.git --no-cache-dir
Looking for a broker? Support the channel by using one of these links. We receive a commission on your trades from the broker at no cost to you:
For Australian customers without trading restrictions – Black Bull
Outside Australia – FXTrading