As an Amazon Associate I earn from qualifying purchases.
Today, we will be continue creating a strategy for your algorithmic trading bot.
By now, you should have 3 new files created from the last post. strategy.py that loads in your strategy, myStrategy.json that contains your strategy and a constants.py file that stores all the moving average functions for ta-lib. Today, we will be exploring how to dynamically apply this strategy to our trader. Let’s get started!
Loading the json strategy from the trader
Let’s go back to the main code add imports for the new python files that you have created. Add strategy and constants and sys to your list of imports:
import MetaTrader5 as mt5
from datetime import datetime, timedelta
import pandas as pd
import pytz
import schedule
import talib as ta
import time
import strategy
import constants
import sysAfter adding the imports, modify the get_data method to include a new argument called strategy:
def get_data(time_frame, strategy):On the next line where pairs are defined, replace this with strategy['pairs']. This means that all of our trading pairs will come from the strategy. This is important as you probably will want to trade different pairs with different strategies:
def get_data(time_frame, strategy):
pairs = strategy['pairs']Add the same strategy parameter to the check_trades method as follows:
def check_trades(time_frame, pair_data, strategy):With this done, delete the 2 lines that are calculating SMA and EMA. We will be replacing this with something more dynamic:
data['SMA'] = ta.SMA(data['close'], 10)
data['EMA'] = ta.EMA(data['close'], 50)First of all, let’s get our defined moving averages from the strategy dictionary. Under the method signature for check_trades add the following:
moving_averages = strategy['movingAverages']Here is a reminder of what the moving averages were defined as from the last post:
"movingAverages": {
"SMA": {
"val": 10,
"aboveBelow": "below"
},
"EMA": {
"val": 50,
"aboveBelow": "above"
}
}Now, you will want to get the corresponding moving average function based on your strategy dictionary. This is done by using a for loop, looping over the keys in movingAverages and matching them with the movingAverageFunctions dictionary created in constants
for m in moving_averages :
ma_func = constants.movingAveragesFunctions[m]You will also need to extract the value from the current moving average as this is used by the moving average function:
val = moving_averages [m]['val']The last step is to run the function with the close price and the val and add it to the data frame. You will add the moving averages to the dataframe by the key from the moving_averages dictionary:
data[m] = maFunc(close, val)The beginning of your check_trades method should look like the following:
def check_trades(time_frame, pair_data, strategy):
moving_averages = strategy['movingAverages']
for pair, data in pair_data.items():
for m in moving_averages:
ma_func = constants.movingAveragesFunctions[m]
val = moving_averages [m]['val']
data[m] = ma_func(data['close'], val)This code will get all the moving averages you want to calculate specified in the strategy, calculate them using the stored functions in constants and then add the results to the data dataframe with key as the column name.
Implementing the strategy stop loss and take profit
Now that we have re-implemented our strategy for entering trades dynamically let’s see how we can add our stop loss and take profits dynamically.
Find the following line in your code:
open_position(pair, "BUY", 1, 300, 100)300 is the current take profit and 100 is the current stop loss. You need to replace these values with the values from your strategy dictionary and make them a float:
open_position(pair, "BUY", 1, float (strategy['takeProfit']), float(strategy['stopLoss']))Dynamically loading the strategy
The last step for creating a strategy for your algorithmic trading bot, is to load the strategy dynamically – You will be passing the strategy file as an argument to the script. But first, let’s implement some code that will take this argument and set it to a variable named current_strategy. You will be using sys.argv[1] to capture the input parameter.
Find the main entry point to your script. If you have been following this series, this is at the bottom of your trader.py file:
if __name__ == '__main__':
live_trading()Start by using sys.argv[1] to capture the strategy file name and assign it to a variable named current_strategy then add some print statements to the beginning of the program to let you know what strategy is currently being run with the trader:
if __name__ == '__main__':
current_strategy = sys.argv[1]
print("Trading bot started with strategy: ", current_strategy)From strategy.py you can now load the json file using the load_strategy method defined earlier:
if __name__ == '__main__':
current_strategy = sys.argv[1]
print("Trading bot started with strategy: ", current_strategy)
current_strategy = strategy.load_strategy(current_strategy)Now pass this into the live_trading method:
if __name__ == '__main__':
current_strategy = sys.argv[1]
print("Trading bot started with strategy: ", current_strategy)
current_strategy = strategy.load_strategy(current_strategy)
live_trading(current_strategy)Add an argument called strategy to the live_trading method and in the schedule code, pass this argument after mt5.TIMEFRAME_M15.
def live_trading(strategy):
schedule.every().hour.at(":00").do(run_trader, mt5.TIMEFRAME_M15, strategy)
schedule.every().hour.at(":15").do(run_trader, mt5.TIMEFRAME_M15, strategy)
schedule.every().hour.at(":30").do(run_trader, mt5.TIMEFRAME_M15, strategy)
schedule.every().hour.at(":45").do(run_trader, mt5.TIMEFRAME_M15, strategy)Finally, add the same strategy argument to run_trader and pass this argument into the get_data and check_trades calls:
Testing the code
Let’s try running our updated code passing in the strategy created from the previous post. As you can see below the trader ran with the strategy ‘strategy’ and also managed to find 2 trades and open them!
> python trader.py strategy
Trading bot started with strategy: strategy
Connected: Connecting to MT5 Client
EURUSD found!
Order successfully placed!
GBPUSD found!
Order successfully placed!
If you are interested in learning more about algo trading and trading systems, I highly recommend reading this book. I have taken some of my own trading ideas and strategies from this book. It also provided me a great insight into effective back testing. Check it out here.
That’s all for now! Check back on Friday to see how you can dynamically calculate your position size of a trade based on risk tolerance and account size! As always, if you have any questions or comments please feel free to post them below. Additionally, if you run into any issues please let me know.
I’m having an issue when I try to run it. After it “gets the data” and proceeds to check the data, it does nothing and loops back. It doesn’t open any trades.
import MetaTrader5 as mt5
from datetime import datetime, timedelta
import pandas as pd
import pytz
import schedule
import talib as ta
import time
import strategy
import constants
import sys
def connect(account):
account = int(account)
mt5.initialize()
authorized=mt5.login(account)
if authorized:
print(“Connected: Connecting to MT5 Client”)
else:
print(“Failed to connect at account #{}, error code: {}”
.format(account, mt5.last_error()))
def open_position(pair, order_type, size, tp_distance=None, stop_distance=None):
symbol_info = mt5.symbol_info(pair)
if symbol_info is None:
print(pair, “not found”)
return
if not symbol_info.visible:
print(pair, “is not visible, trying to switch on”)
if not mt5.symbol_select(pair, True):
print(“symbol_select({}}) failed, exit”,pair)
return
print(pair, “found!”)
point = symbol_info.point
if(order_type == “BUY”):
order = mt5.ORDER_TYPE_BUY
price = mt5.symbol_info_tick(pair).ask
if(stop_distance):
sl = price – (stop_distance * point)
if(tp_distance):
tp = price + (tp_distance * point)
if(order_type == “SELL”):
order = mt5.ORDER_TYPE_SELL
price = mt5.symbol_info_tick(pair).bid
if(stopDistance):
sl = price + (stop_distance * point)
if(tpDistance):
tp = price – (tp_distance * point)
request = {
“action”: mt5.TRADE_ACTION_DEAL,
“symbol”: pair,
“volume”: float(size),
“type”: order,
“price”: price,
“sl”: sl,
“tp”: tp,
“magic”: 234000,
“comment”: “”,
“type_time”: mt5.ORDER_TIME_GTC,
“type_filling”: mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
print(“Failed to send order :(“)
else:
print (“Order successfully placed!”)
def positions_get(symbol=None):
if(symbol is None):
res = mt5.positions_get()
else:
res = mt5.positions_get(symbol=symbol)
if(res is not None and res != ()):
df = pd.DataFrame(list(res),columns=res[0]._asdict().keys())
df[‘time’] = pd.to_datetime(df[‘time’], unit=’s’)
return df
return pd.DataFrame()
def close_position(deal_id):
open_positions = positions_get()
open_positions = open_positions[open_positions[‘ticket’] == deal_id]
order_type = open_positions[“type”][0]
symbol = open_positions[‘symbol’][0]
volume = open_positions[‘volume’][0]
if(order_type == mt5.ORDER_TYPE_BUY):
order_type = mt5.ORDER_TYPE_SELL
price = mt5.symbol_info_tick(symbol).bid
else:
order_type = mt5.ORDER_TYPE_BUY
price = mt5.symbol_info_tick(symbol).ask
close_request={
“action”: mt5.TRADE_ACTION_DEAL,
“symbol”: symbol,
“volume”: float(volume),
“type”: order_type,
“position”: deal_id,
“price”: price,
“magic”: 234000,
“comment”: “Close trade”,
“type_time”: mt5.ORDER_TIME_GTC,
“type_filling”: mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(close_request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
print(“Failed to close order :(“)
else:
print (“Order successfully closed!”)
def close_positon_by_symbol(symbol):
open_positions = positions_get(symbol)
open_positions[‘ticket’].apply(lambda x: close_position(x))
def get_data(time_frame, strategy):
print(“Getting data now…”)
pairs = strategy[‘pairs’]
pair_data = dict()
for pair in pairs:
utc_from = datetime(2021, 1, 1, tzinfo=pytz.timezone(‘Europe/London’))
date_to = datetime.now().astimezone(pytz.timezone(‘Europe/London’))
date_to = datetime(date_to.year, date_to.month, date_to.day, hour=date_to.hour, minute=date_to.minute)
rates = mt5.copy_rates_range(pair, time_frame, utc_from, date_to)
rates_frame = pd.DataFrame(rates)
rates_frame[‘time’] = pd.to_datetime(rates_frame[‘time’], unit=’s’)
rates_frame.drop(rates_frame.tail(1).index, inplace = True)
pair_data[pair] = rates_frame
print(pair_data[pair])
return pair_data
def check_trades(time_frame, pair_data, strategy):
print(“Checking trades now…”)
moving_averages = strategy[‘movingAverages’]
for pair, data in pair_data.items():
for m in moving_averages:
ma_func = constants.movingAveragesFunctions[m]
val = moving_averages [m][‘val’]
data[m] = ma_func(data[‘close’], val)
last_row = data.tail(1)
open_positions = positions_get(pair)
#open_positions = positions_get()
#current_dt = datetime.now().astimezone(pytz.timezone(‘Europe/London’))
#for index, position in open_positions.iterrows():
# Check to see if the trade has exceeded the time limit
#trade_open_dt = position[‘time’].replace(tzinfo = pytz.timezone(‘Europe/London’))
#deal_id = position[‘ticket’]
#if(current_dt – trade_open_dt >= timedelta(hours = 2)):
for index, last in last_row.iterrows():
#Exit strategy
if(last[‘close’] last[‘SMA’]):
close_positon_by_symbol(pair)
#Entry strategy
#if(last[‘close’] > last[‘EMA’] and last[‘close’] < last['SMA']):
open_position(pair, "BUY", 1, float (strategy['takeProfit']), float(strategy['stopLoss']))
def run_trader(time_frame, strategy):
print("Running trader at", datetime.now())
connect(12345678)
pair_data = get_data(time_frame, strategy)
#open_position("USDCAD", "BUY", 1, 300, 100)
check_trades(time_frame, pair_data, strategy)
def live_trading(strategy):
schedule.every().hour.at(":00").do(run_trader, mt5.TIMEFRAME_M15, strategy)
schedule.every().hour.at(":15").do(run_trader, mt5.TIMEFRAME_M15, strategy)
schedule.every().hour.at(":30").do(run_trader, mt5.TIMEFRAME_M15, strategy)
schedule.every().hour.at(":45").do(run_trader, mt5.TIMEFRAME_M15, strategy)
while True:
schedule.run_pending()
time.sleep(1)
if __name__ == '__main__':
#print(sys.argv)
#current_strategy = sys.argv[1]
print("C:\Python\Python39\Scripts\strategies\MyStrategy.json")
current_strategy = "MyStrategy"
print("Trading bot started with strategy: ", current_strategy)
current_strategy = strategy.load_strategy(current_strategy)
live_trading(current_strategy)
Could you tell, how to use correctly aboveBelow parameter from myStrategy in check_trades function.
The aboveBelow parameter is just specifying if the price should be above or below the moving average specified to open/close a trade