Em posts recentes, ensinamos diversas técnicas de trading utilizando o Python, bem como também mostramos como utilizar o MetaTrader5 para a importação de dados de ativos financeiros e a realização de requisições de ordem e venda com a linguagem. No post de hoje, estaremos interessados em criar um ambiente automatizado, que pretende criar as requisições de acordo com os sinais de um sistema de trading em tempo real.
O objetivo será criar um ambiente totalmente automatizado, portanto, será necessário usar diversas opções de funções e métodos disponíveis na biblioteca MetaTrader 5 do Python, de forma que atenda os principais requisitos de criação de ordem e venda.
Iremos criar esse ambiente em uma classe definida com diversas funções em um arquivo .py. Após isso, utilizaremos um arquivo .ipnyb para definir o indicador utilizado, no caso, o RSI, para definir as ordens de compra e venda. Os dois arquivos são separados pelos blocos de código abaixo.
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime
class MT5:
max_price = dict()
min_price = dict()
summary = None
def get_ticks(symbol, number_of_data = 10000):
# Data e horário atual
from_date = datetime.now()
# Extrai os ticks (instrumentos financeiros)
ticks = mt5.copy_ticks_from(symbol, from_date, number_of_data, mt5.COPY_TICKS_ALL)
# Transforma a tupla em data frame
df_ticks = pd.DataFrame(ticks)
# Converte a data em datetime
df_ticks["time"] = pd.to_datetime(df_ticks["time"], unit="s")
# Coloca a data no index
df_ticks = df_ticks.set_index("time")
return df_ticks
def get_rates(symbol, number_of_data = 10000, timeframe=mt5.TIMEFRAME_D1):
# Cria a data atual
from_date = datetime.now()
# Extrai os ticks (instrumentos financeiros)
rates = mt5.copy_rates_from(symbol, timeframe, from_date, number_of_data)
# Transforma a tupla em data frame
df_rates = pd.DataFrame(rates)
# Converte a data em datetime
df_rates["time"] = pd.to_datetime(df_rates["time"], unit="s")
# Coloca a data no index
df_rates = df_rates.set_index("time")
return df_rates
def risk_reward_threshold(symbol, buy=True, risk=0.01, reward=0.02):
# Extrai o leverage (alavancagem)
leverage = mt5.account_info().leverage
# Computa o preço
price = mt5.symbol_info(symbol).ask
# Extrai o número de decimais
nb_decimal = str(price)[::-1].find(".")
# Computa as variação em porcentagem
var_down = risk/leverage
var_up = reward/leverage
# Acha o limite do TP e SL em preço absoluto
if buy:
price = mt5.symbol_info(symbol).ask
# Computa as variações em preço absoluto
price_var_down = var_down*price
price_var_up = var_up * price
tp = np.round(price + price_var_up, nb_decimal)
sl = np.round(price - price_var_down, nb_decimal)
else:
price = mt5.symbol_info(symbol).bid
# Computa as variações em preço absoluto
price_var_down = var_down*price
price_var_up = var_up * price
tp = np.round(price - price_var_up, nb_decimal)
sl = np.round(price + price_var_down, nb_decimal)
return tp, sl
def find_filling_mode(symbol):
for i in range(2):
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": mt5.symbol_info(symbol).volume_min,
"type": mt5.ORDER_TYPE_BUY,
"price": mt5.symbol_info_tick(symbol).ask,
"type_filling": i,
"type_time": mt5.ORDER_TIME_GTC}
result = mt5.order_check(request)
if result.comment == "Feito":
break
return i
def send_order(symbol, lot, buy, sell, id_position=None, pct_tp=0.02, pct_sl=0.01, comment=" No specific comment", magic=0):
# Inicializa o mt5 com o Python
mt5.initialize()
# Extrai o filling_mode
filling_type = MT5.find_filling_mode(symbol)
""" ABRE O TRADE """
if buy and id_position==None:
tp, sl = MT5.risk_reward_threshold(symbol, buy=True, risk=pct_sl, reward=pct_tp)
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": lot,
"type": mt5.ORDER_TYPE_BUY,
"price": mt5.symbol_info_tick(symbol).ask,
"deviation": 10,
"tp": tp,
"sl": sl,
"magic": magic,
"comment": comment,
"type_filling": filling_type,
"type_time": mt5.ORDER_TIME_GTC}
result = mt5.order_send(request)
print(mt5.symbol_info_tick(symbol).ask, tp, sl)
return result
if sell and id_position==None:
tp, sl = MT5.risk_reward_threshold(symbol, buy=False, risk=pct_sl, reward=pct_tp)
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": lot,
"type": mt5.ORDER_TYPE_SELL,
"price": mt5.symbol_info_tick(symbol).bid,
"deviation": 10,
"tp": tp,
"sl": sl,
"magic": magic,
"comment": comment,
"type_filling": filling_type,
"type_time": mt5.ORDER_TIME_GTC}
result = mt5.order_send(request)
print(mt5.symbol_info_tick(symbol).bid, tp, sl)
return result
""" FECHA O TRADE """
if buy and id_position!=None:
request = {
"position": id_position,
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": lot,
"type": mt5.ORDER_TYPE_SELL,
"price": mt5.symbol_info_tick(symbol).bid,
"deviation": 10,
"magic": magic,
"comment": comment,
"type_filling": filling_type,
"type_time": mt5.ORDER_TIME_GTC}
result = mt5.order_send(request)
return result
if sell and id_position!=None:
request = {
"position": id_position,
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": lot,
"type": mt5.ORDER_TYPE_BUY,
"price": mt5.symbol_info_tick(symbol).ask,
"deviation": 10,
"magic": magic,
"comment": comment,
"type_filling": filling_type,
"type_time": mt5.ORDER_TIME_GTC}
result = mt5.order_send(request)
return result
def resume():
""" Retorna as posições atuais. Posição=0 --> Compra """
# Define o nome das colunas que serão criadas
colonnes = ["ticket", "position", "symbol", "volume", "magic", "profit", "price", "tp", "sl","trade_size"]
# Pega as posições em aberto
liste = mt5.positions_get()
# Cria um dataframe vazio
summary = pd.DataFrame()
# Cria o loop para adicionar cada linha no dataframe
for element in liste:
element_pandas = pd.DataFrame([element.ticket, element.type, element.symbol, element.volume, element.magic,
element.profit, element.price_open, element.tp,
element.sl, mt5.symbol_info(element.symbol).trade_contract_size],
index=colonnes).transpose()
summary = pd.concat((summary, element_pandas), axis=0)
try:
summary["profit %"] = summary.profit / (summary.price * summary.trade_size * summary.volume)
summary = summary.reset_index(drop=True)
except:
pass
return summary
def trailing_stop_loss():
# Extrai as posições abertas atuais
MT5.summary = MT5.resume()
# Verificaçã: Tem alguma posição em aberto?
if MT5.summary.shape[0] >0:
for i in range(MT5.summary.shape[0]):
# Extrai as informação
row = MT5.summary.iloc[i]
symbol = row["symbol"]
""" CASO 1: Altera dinâmicamente o stop loss para uma ORDEM DE COMPRA """
# Trailing stop loss para um posição de compra
if row["position"] == 0:
if symbol not in MT5.max_price.keys():
MT5.max_price[symbol]=row["price"]
# Extrai o preço atual
current_price = (mt5.symbol_info(symbol).ask + mt5.symbol_info(symbol).bid ) / 2
# Computa a distância entre o preço atual e preço máximo
from_sl_to_curent_price = current_price - row["sl"]
from_sl_to_max_price = MT5.max_price[symbol] - row["sl"]
# Se o preço atual é maior que o preço máximo anterior --> novo preço máximo
if current_price > MT5.max_price[symbol]:
MT5.max_price[symbol] = current_price
# Encontra a diferença entre a diferença do preço menor e máximo
if from_sl_to_curent_price > from_sl_to_max_price:
difference = from_sl_to_curent_price - from_sl_to_max_price
# Seta filling mode
filling_type = mt5.symbol_info(symbol).filling_mode
# Seta o point
point = mt5.symbol_info(symbol).point
# Altera o sl
request = {
"action": mt5.TRADE_ACTION_SLTP,
"symbol": symbol,
"position": row["ticket"],
"volume": row["volume"],
"type": mt5.ORDER_TYPE_BUY,
"price": row["price"],
"sl": row["sl"] + difference,
"type_filling": filling_type,
"type_time": mt5.ORDER_TIME_GTC,
}
information = mt5.order_send(request)
print(information)
""" CASO 2: Altera dinâmicamente o stop loss para uma ORDEM DE VENDA """
# Trailing stop loss para uma ordem de venda
if row["position"] == 1:
if symbol not in MT5.min_price.keys():
MT5.min_price[symbol]=row["price"]
# Extrai o preço atual
current_price = (mt5.symbol_info(symbol).ask + mt5.symbol_info(symbol).bid ) / 2
# Computa a distância entre o preço atual e preço máximo
from_sl_to_curent_price = row["sl"] - current_price
from_sl_to_min_price = row["sl"] - MT5.min_price[symbol]
# Se o preço atual é maior que o preço máximo anterior --> novo preço máximo
if current_price < MT5.min_price[symbol]:
MT5.min_price[symbol] = current_price
# Encontra a diferença entre a diferença do preço menor e máximo
if from_sl_to_curent_price > from_sl_to_min_price:
difference = from_sl_to_curent_price - from_sl_to_min_price
# Seta filling mode
filling_type = mt5.symbol_info(symbol).filling_mode
# Seta o point
point = mt5.symbol_info(symbol).point
# Altera o sl
request = {
"action": mt5.TRADE_ACTION_SLTP,
"symbol": symbol,
"position": row["ticket"],
"volume": row["volume"],
"type": mt5.ORDER_TYPE_SELL,
"price": row["price"],
"sl": row["sl"] - difference,
"type_filling": filling_type,
"type_time": mt5.ORDER_TIME_GTC,
}
information = mt5.order_send(request)
print(information)
def verif_tsl():
if len(MT5.summary)>0:
buy_open_positions = MT5.summary.loc[MT5.summary["position"]==0]["symbol"]
sell_open_positions = MT5.summary.loc[MT5.summary["position"]==0]["symbol"]
else:
buy_open_positions = []
sell_open_positions = []
""" SE FECHAR UMA DAS POSIÇÃO É NECESSÁRIO DELETAR O PREÇO NO DICIONÁRIO DE PREÇO MAX E MIN """
if len(MT5.max_price) != len(buy_open_positions) and len(buy_open_positions) >0:
symbol_to_delete = []
for symbol in MT5.max_price.keys():
if symbol not in list(buy_open_positions):
symbol_to_delete.append(symbol)
for symbol in symbol_to_delete:
del MT5.max_price[symbol]
if len(MT5.min_price) != len(sell_open_positions) and len(sell_open_positions) >0:
symbol_to_delete = []
for symbol in MT5.min_price.keys():
if symbol not in list(sell_open_positions):
symbol_to_delete.append(symbol)
for symbol in symbol_to_delete:
del MT5.min_price[symbol]
if len(buy_open_positions) == 0:
MT5.max_price={}
if len(sell_open_positions) == 0:
MT5.min_price={}
def run(symbol, buy, sell, lot, pct_tp=0.02, pct_sl=0.01, comment="", magic=23400):
# Inicializa
mt5.initialize()
# Escolhe o símbolo
print("------------------------------------------------------------------")
print("Date: ", datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "\tSYMBOL:", symbol)
# Inicializa
ouvertures = MT5.resume()
# Compra ou Vende
print(f"BUY: {buy} \t SELL: {sell}")
""" Fecha o trade eventualmente """
# Tipo de extração do preço
try:
position = ouvertures.loc[ouvertures["symbol"] == symbol].values[0][1]
identifier = ouvertures.loc[ouvertures["symbol"] == symbol].values[0][0]
except:
position = None
identifier = None
if position!=None:
print(f"POSITION: {position} \t ID: {identifier}")
# Verifica os trades
if buy == True and position == 0:
buy = False
elif buy == False and position == 0:
before = mt5.account_info().balance
res = MT5.send_order(symbol, lot, True, False, id_position=identifier,pct_tp=pct_tp, pct_sl=pct_sl, comment=" No specific comment", magic=0)
after = mt5.account_info().balance
print(f"CLOSE BUY POSITION: {res.comment}")
pct = np.round(100*(after-before)/before, 3)
if res.comment != "Request executed":
print("WARNINGS", res.comment)
elif sell == True and position == 1:
sell = False
elif sell == False and position == 1:
before = mt5.account_info().balance
res = MT5.send_order(symbol, lot, False, True, id_position=identifier,pct_tp=pct_tp, pct_sl=pct_sl, comment=" No specific comment", magic=0)
print(f"CLOSE SELL POSITION: {res.comment}")
after = mt5.account_info().balance
pct = np.round(100*(after-before)/before, 3)
if res.comment != "Request executed":
print("WARNINGS", res.comment)
else:
pass
""" Compra ou Vende """
if buy == True:
res = MT5.send_order(symbol, lot, True, False, id_position=None,pct_tp=pct_tp, pct_sl=pct_sl, comment=" No specific comment", magic=0)
print(f"OPEN BUY POSITION: {res.comment}")
if res.comment != "Request executed":
print("WARNINGS", res.comment)
if sell == True:
res = MT5.send_order(symbol, lot, False, True, id_position=None,pct_tp=pct_tp, pct_sl=pct_sl, comment=" No specific comment", magic=0)
print(f"OPEN SELL POSITION: {res.comment}")
if res.comment != "Request executed":
print("WARNINGS", res.comment)
print("------------------------------------------------------------------")
<pre></pre>
<pre>
</pre>
# Importa as bibliotecas
import talib as ta
import time
from datetime import datetime
import numpy as np
import pandas as pd
from algo_mt5 import *
import MetaTrader5 as mt5
mt5.initialize()
# Define o indicador (rsi)
def rsi(symbol):
overbuy = 70
oversell = 30
neutral = 50
df = MT5.get_data(symbol, 3500).dropna()
df["rsi"] = ta.RSI(df["close"], timeperiod=14)
today = df["rsi"].iloc[-1]
# Long buy signal
buy = (today > neutral) & (today < overbuy)
# Short selling signal
sell = (today < neutral) & (today > oversell)
return buy, sell
# Define os ativos financeiros
symbols_list = {
"Petrobras": ["PETR4", 1]}
# Constrói a mensagem
current_account_info = mt5.account_info()
print("------------------------------------------------------------------")
print(f"Login: {mt5.account_info().login} \tserver: {mt5.account_info().server}")
print(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Balance: {current_account_info.balance} USD, \t Equity: {current_account_info.equity} USD, \t Profit: {current_account_info.profit} USD")
print(f"Run time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("------------------------------------------------------------------")
# Em qual dia e horário irá rodar o algortimo?
launching = [[0, "00:15:59"],
[0, "15:59:50"],
[1, "15:59:50"],
[2, "15:59:50"],
[3, "15:59:50"],
[4, "15:59:50"]]
# Lança o algoritmo
while True:
# TSL
MT5.trailing_stop_loss()
MT5.verif_tsl()
# Verficação para o lançamento
current_time = [datetime.now().weekday(), datetime.now().strftime("%H:%M:%S")]
if current_time in launching:
is_time = True
else:
is_time = False
# is_time=True # Apenas para rodar o trading AGORA
if is_time:
# Abre os trades
for asset in symbols_list.keys():
# Inicializa os inputs
symbol = symbols_list[asset][0]
lot = symbols_list[asset][1]
selected = mt5.symbol_select(symbol)
if not selected:
print(f"\nERROR - Failed to select '{symbol}' in MetaTrader 5 with error :",mt5.last_error())
else:
# Cria os sinais
buy, sell = rsi(symbol)
# Roda o algoritimo
MT5.run(symbol, buy, sell, lot, pct_tp=0.1, pct_sl=0.05)
______________________________________
Quer saber mais?
Veja nosso curso de Python para Investimentos.
________________________________
Referências
Inglese, Lucas. Python for Finance and Algorithmic trading (2nd edition): Machine Learning, Deep Learning, Time series Analysis, Risk and Portfolio Management for MetaTrader™5 Live Trading
