Agentes Python no Mercado Financeiro com MetaTrader 5: Tecnologia, Integração e Realidade
Automatizar operações no mercado financeiro com Python e MetaTrader 5 é tecnicamente viável e engenheiramente interessante. Mas o mercado não é um problema de software — é um sistema adaptativo adversarial. Este artigo foca na tecnologia, nas bibliotecas e nas boas práticas, sem vender ilusões.
Automatizar operações no Mercado Financeiro é um dos projetos mais sedutores para um engenheiro de software. A combinação de dados em tempo real, tomada de decisão algorítmica, execução de ordens e gestão de risco tem todos os elementos de um sistema distribuído complexo — e a promessa de que o código vai "trabalhar enquanto você dorme" é irresistível.
Mas antes de escrever uma linha de código, uma declaração honesta: o mercado financeiro não é um problema de engenharia esperando ser resolvido. É um sistema adaptativo adversarial onde cada participante tenta extrair valor dos outros. Estratégias que funcionam deixam de funcionar quando outros as descobrem. Backtests otimistas não predizem performance futura. A maioria dos traders algorítmicos perde dinheiro.
Este artigo não é sobre ganhar dinheiro. É sobre como construir sistemas de automação financeira corretamente do ponto de vista técnico — porque a engenharia é genuinamente interessante, independentemente do resultado financeiro.
MetaTrader 5 e a Biblioteca Python
MetaTrader 5 (MT5) é uma plataforma de trading amplamente usada por corretoras. A biblioteca Python oficial, disponível desde 2019, permite controlar o MT5 a partir de scripts externos via IPC — não é uma API REST, é uma bridge direta com o processo do MT5 rodando localmente no Windows.
pip install MetaTrader5 pandas numpy ta-lib backtrader python-dotenv
Essa limitação é importante: a biblioteca MetaTrader5 só funciona em Windows. Para rodar em Linux ou cloud, é necessário usar Wine ou uma VM Windows — o que aumenta significativamente a complexidade operacional.
Conexão e Autenticação
import MetaTrader5 as mt5
import os
from dotenv import load_dotenv
load_dotenv()
def connect_to_mt5() -> bool: if not mt5.initialize( login=int(os.getenv("MT5_LOGIN")), password=os.getenv("MT5_PASSWORD"), server=os.getenv("MT5_SERVER"), ): print(f"initialize() failed: {mt5.last_error()}") return False info = mt5.account_info() print(f"Connected: {info.name} | Balance: {info.balance} {info.currency}") return True
def shutdown_mt5() -> None: mt5.shutdown()
Nunca coloque credenciais hardcoded. Use variáveis de ambiente ou um gerenciador de segredos. O arquivo .env nunca deve ir para o repositório — adicione ao .gitignore imediatamente.
Obtendo Dados de Mercado
import pandas as pd
from datetime import datetime, timezone
def get_ohlcv(symbol: str, timeframe: int, bars: int = 500) -> pd.DataFrame: rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, bars) if rates is None: raise ValueError(f"Failed to get rates for {symbol}: {mt5.last_error()}") df = pd.DataFrame(rates) df["time"] = pd.to_datetime(df["time"], unit="s", utc=True) df.set_index("time", inplace=True) return df[["open", "high", "low", "close", "tick_volume"]]
df_h1 = get_ohlcv("PETR4", mt5.TIMEFRAME_H1, bars=1000) df_m15 = get_ohlcv("WIN$N", mt5.TIMEFRAME_M15, bars=500)
Para dados tick-by-tick:
def get_ticks(symbol: str, count: int = 1000) -> pd.DataFrame:
ticks = mt5.copy_ticks_from(
symbol,
datetime.now(timezone.utc),
count,
mt5.COPY_TICKS_ALL,
)
if ticks is None:
raise ValueError(f"Failed to get ticks: {mt5.last_error()}")
df = pd.DataFrame(ticks)
df["time"] = pd.to_datetime(df["time"], unit="s", utc=True)
return df
Indicadores Técnicos com TA-Lib
TA-Lib é o wrapper Python de uma biblioteca C com mais de 150 indicadores, usada como referência para análise técnica:
import talib
def calculate_indicators(df: pd.DataFrame) -> pd.DataFrame: close = df["close"].values high = df["high"].values low = df["low"].values
df["sma_20"] = talib.SMA(close, timeperiod=20) df["sma_50"] = talib.SMA(close, timeperiod=50) df["rsi_14"] = talib.RSI(close, timeperiod=14) df["atr_14"] = talib.ATR(high, low, close, timeperiod=14)
macd, signal, _ = talib.MACD(close, fastperiod=12, slowperiod=26, signalperiod=9) df["macd"] = macd df["macd_signal"] = signal
upper, middle, lower = talib.BBANDS(close, timeperiod=20, nbdevup=2, nbdevdn=2) df["bb_upper"] = upper df["bb_middle"] = middle df["bb_lower"] = lower
return df.dropna()
Execução de Ordens
A execução de ordens é a parte mais crítica — e onde mais erros acontecem. O MT5 usa uma estrutura de request explícita:
def send_market_order(
symbol: str,
side: int,
volume: float,
sl: float,
tp: float,
magic: int = 12345,
) -> dict:
tick = mt5.symbol_info_tick(symbol)
price = tick.ask if side == mt5.ORDER_TYPE_BUY else tick.bid
request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume, "type": side, "price": price, "sl": sl, "tp": tp, "deviation": 10, "magic": magic, "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, }
result = mt5.order_send(request)
if result is None: raise RuntimeError(f"order_send returned None: {mt5.last_error()}")
if result.retcode != mt5.TRADE_RETCODE_DONE: raise RuntimeError(f"Order rejected: retcode={result.retcode} | {result.comment}")
return {"ticket": result.order, "price": result.price, "volume": result.volume}
O campo deviation (slippage máximo em points) é crítico em mercados voláteis. Os retcodes mais comuns: 10009 (DONE — sucesso), 10004 (REQUOTE — preço mudou), 10006 (REJECT — rejeitada), 10014 (INVALID_VOLUME — volume inválido para o símbolo).
Gestão de Risco
O que o código pode fazer:
def calculate_position_size(
balance: float,
risk_pct: float,
entry: float,
stop_loss: float,
pip_value: float,
) -> float:
risk_amount = balance * (risk_pct / 100)
stop_distance = abs(entry - stop_loss)
if stop_distance == 0:
raise ValueError("Stop loss cannot equal entry price")
return round(risk_amount / (stop_distance * pip_value), 2)
def within_drawdown_limit(initial: float, current: float, limit_pct: float = 10.0) -> bool: drawdown = ((initial - current) / initial) * 100 return drawdown <= limit_pct
O que o código não pode fazer: prever o mercado. Stop loss e position sizing limitam a perda por operação, mas não garantem lucratividade no longo prazo.
Backtesting com Backtrader
Antes de operar com capital real, toda estratégia deve ser testada em dados históricos:
import backtrader as bt
class SmaCrossStrategy(bt.Strategy): params = (("fast", 20), ("slow", 50))
def __init__(self): self.cross = bt.indicators.CrossOver( bt.indicators.SMA(period=self.p.fast), bt.indicators.SMA(period=self.p.slow), )
def next(self): if not self.position and self.cross > 0: self.buy() elif self.position and self.cross < 0: self.sell()
cerebro = bt.Cerebro() cerebro.addstrategy(SmaCrossStrategy) cerebro.broker.setcash(10000) cerebro.broker.setcommission(commission=0.001) cerebro.adddata(bt.feeds.PandasData(dataname=df_h1)) cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe") cerebro.addanalyzer(bt.analyzers.DrawDown, _name="dd")
strat = cerebro.run()[0] print(f"Sharpe: {strat.analyzers.sharpe.get_analysis()['sharperatio']:.2f}") print(f"Max DD: {strat.analyzers.dd.get_analysis()['max']['drawdown']:.2f}%")
Aviso crítico: quase toda estratégia performa bem no backtest e mal em produção. Os motivos principais são overfitting (a estratégia foi otimizada para dados históricos específicos), look-ahead bias (usar informação que não estaria disponível no momento da decisão), survivorship bias e subestimação de custos reais (spread, slippage, corretagem, emolumentos, IR).
Estrutura de um Agente de Produção
Um agente que opera em produção precisa de ciclo de vida robusto, logging estruturado e circuit breaker:
import logging
import time
logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s", handlers=[logging.FileHandler("agent.log"), logging.StreamHandler()], ) log = logging.getLogger(__name__)
class TradingAgent: def __init__(self, symbol: str, max_errors: int = 5, max_trades_day: int = 10): self.symbol = symbol self.max_errors = max_errors self.max_trades_day = max_trades_day self._errors = 0 self._trades_today = 0 self._running = True
def run(self) -> None: log.info(f"Agent starting for {self.symbol}") while self._running: try: self._tick() self._errors = 0 time.sleep(60) except KeyboardInterrupt: log.info("Shutdown requested") self._running = False except Exception as e: self._errors += 1 log.error(f"Error {self._errors}/{self.max_errors}: {e}") if self._errors >= self.max_errors: log.critical("Circuit breaker — stopping agent") self._running = False time.sleep(30)
def _tick(self) -> None: if self._trades_today >= self.max_trades_day: log.info("Daily trade limit reached — skipping tick") return df = get_ohlcv(self.symbol, mt5.TIMEFRAME_M15, bars=200) df = calculate_indicators(df) signal = self._evaluate_signal(df) if signal: log.info(f"Signal: {signal} on {self.symbol}")
def _evaluate_signal(self, df: pd.DataFrame) -> str | None: last = df.iloc[-1] prev = df.iloc[-2] if prev["macd"] < prev["macd_signal"] and last["macd"] > last["macd_signal"]: if last["rsi_14"] < 70: return "BUY" if prev["macd"] > prev["macd_signal"] and last["macd"] < last["macd_signal"]: if last["rsi_14"] > 30: return "SELL" return None
O circuit breaker é fundamental — um agente que entra em loop de erros pode causar danos severos. Limite o número de erros consecutivos, o número de operações por dia e o drawdown máximo.
Machine Learning Aplicado: scikit-learn, TensorFlow e PyTorch
Além dos indicadores técnicos clássicos, é possível usar modelos de machine learning para classificar sinais ou prever direção de preço. A abordagem mais honesta: ML não resolve o problema de prever mercados — mas pode complementar estratégias baseadas em regras quando usado com rigor científico.
Feature Engineering com Indicadores
O primeiro passo é transformar dados OHLCV em features que modelos supervisionados consigam consumir:
import numpy as np
from sklearn.preprocessing import StandardScaler
def build_features(df: pd.DataFrame) -> pd.DataFrame: df = calculate_indicators(df)
df["return_1"] = df["close"].pct_change(1) df["return_5"] = df["close"].pct_change(5) df["return_10"] = df["close"].pct_change(10)
df["rsi_norm"] = df["rsi_14"] / 100.0 df["macd_norm"] = df["macd"] / df["close"] df["atr_norm"] = df["atr_14"] / df["close"]
df["bb_pos"] = (df["close"] - df["bb_lower"]) / (df["bb_upper"] - df["bb_lower"])
df["vol_ratio"] = df["tick_volume"] / df["tick_volume"].rolling(20).mean()
return df.dropna()
def build_target(df: pd.DataFrame, horizon: int = 5, threshold: float = 0.002) -> pd.Series: future_return = df["close"].shift(-horizon) / df["close"] - 1 target = pd.Series(0, index=df.index) target[future_return > threshold] = 1 target[future_return < -threshold] = -1 return target
O target aqui é ternário: 1 (alta), -1 (queda), 0 (neutro). A definição do threshold é crítica — thresholds baixos geram muitos sinais com baixo sinal-ruído; altos demais resultam em poucos exemplos positivos/negativos.
Modelos Clássicos com scikit-learn
Random Forest e XGBoost são os pontos de partida mais robustos para dados tabulares financeiros. São resistentes a overfitting quando bem validados e interpretáveis:
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import classification_report
import joblib
FEATURE_COLS = ["return_1", "return_5", "return_10", "rsi_norm", "macd_norm", "atr_norm", "bb_pos", "vol_ratio"]
def train_and_evaluate(df: pd.DataFrame, target: pd.Series) -> RandomForestClassifier: X = df[FEATURE_COLS].values y = target.values
tscv = TimeSeriesSplit(n_splits=5)
for fold, (train_idx, val_idx) in enumerate(tscv.split(X)): X_train, X_val = X[train_idx], X[val_idx] y_train, y_val = y[train_idx], y[val_idx]
model = RandomForestClassifier( n_estimators=200, max_depth=6, min_samples_leaf=50, class_weight="balanced", random_state=42, n_jobs=-1, ) model.fit(X_train, y_train)
preds = model.predict(X_val) print(f"Fold {fold + 1}:") print(classification_report(y_val, preds, target_names=["DOWN", "NEUTRAL", "UP"]))
model.fit(X, y) joblib.dump(model, "rf_signal_model.pkl") return model
O ponto crítico: usar TimeSeriesSplit em vez de train_test_split aleatório. Em séries temporais, validação aleatória causa data leakage — o modelo treina em dados futuros e parece funcionar quando na verdade está trapaceando.
LSTM com PyTorch para Séries Temporais
LSTMs capturam dependências sequenciais que modelos tabulares ignoram. São mais complexos de treinar e validar, e mais propensos a overfitting:
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
class PriceDataset(Dataset): def __init__(self, X: np.ndarray, y: np.ndarray, seq_len: int = 20): self.seq_len = seq_len self.X = torch.FloatTensor(X) self.y = torch.LongTensor(y + 1)
def __len__(self) -> int: return len(self.X) - self.seq_len
def __getitem__(self, idx: int): return self.X[idx:idx + self.seq_len], self.y[idx + self.seq_len]
class LSTMSignalModel(nn.Module): def __init__(self, input_size: int, hidden_size: int = 64, num_layers: int = 2, dropout: float = 0.3): super().__init__() self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, dropout=dropout, batch_first=True, ) self.dropout = nn.Dropout(dropout) self.fc = nn.Linear(hidden_size, 3)
def forward(self, x: torch.Tensor) -> torch.Tensor: out, _ = self.lstm(x) out = self.dropout(out[:, -1, :]) return self.fc(out)
def train_lstm(model: LSTMSignalModel, loader: DataLoader, epochs: int = 50) -> None: optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5) criterion = nn.CrossEntropyLoss() scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5)
for epoch in range(epochs): model.train() total_loss = 0.0 for X_batch, y_batch in loader: optimizer.zero_grad() preds = model(X_batch) loss = criterion(preds, y_batch) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item()
avg_loss = total_loss / len(loader) scheduler.step(avg_loss)
if (epoch + 1) % 10 == 0: print(f"Epoch {epoch + 1}/{epochs} | Loss: {avg_loss:.4f}")
O gradient clipping (clip_grad_norm_) é essencial para LSTMs — sem ele, gradientes explodem em sequências longas.
Integrando o Modelo ao Agente MT5
O modelo treinado pode substituir ou complementar os indicadores técnicos na tomada de decisão:
class MLTradingAgent(TradingAgent):
def __init__(self, symbol: str, model_path: str, **kwargs):
super().__init__(symbol, **kwargs)
self.model = joblib.load(model_path)
self.scaler = joblib.load(model_path.replace(".pkl", "_scaler.pkl"))
def _evaluate_signal(self, df: pd.DataFrame) -> str | None: features = df[FEATURE_COLS].iloc[-1:].values features_scaled = self.scaler.transform(features)
proba = self.model.predict_proba(features_scaled)[0] pred = self.model.predict(features_scaled)[0] confidence = proba.max()
if confidence < 0.6: return None
if pred == 1: return "BUY" if pred == -1: return "SELL" return None
O threshold de confiança (0.6) é fundamental: o modelo só age quando está suficientemente confiante. Sinais com baixa confiança são mais ruído do que sinal.
O Aviso Mais Importante sobre ML Financeiro
TensorFlow e PyTorch produzem modelos impressionantes em backtests. As métricas de validação parecem promissoras. E então o modelo vai a mercado e performa mal ou pior que aleatório.
Os motivos são estruturais:
Não-estacionariedade: mercados financeiros mudam de regime constantemente. Um modelo treinado em dados de 2020-2023 não generaliza para 2024 porque as relações entre features e target mudaram.
Data leakage sutil: normalização usando estatísticas de toda a série, features calculadas com janelas que cruzam o ponto de divisão treino/teste, survivorship bias nos dados históricos — qualquer um desses erros faz o backtest parecer bom quando não é.
Overfitting de mercado: com features suficientes e dados históricos limitados, qualquer modelo complexo vai memorizar ruído específico do período de treino.
Adverse selection: quando o modelo sinaliza compra, alguém está disposto a vender. Esse alguém frequentemente tem mais informação.
A recomendação prática: use ML como filtro adicional sobre uma estratégia baseada em regras, não como oráculo independente. Valide sempre com walk-forward testing, nunca com split aleatório. Monitore drift do modelo em produção com as técnicas de observabilidade descritas no artigo anterior desta série.
O que o Mercado Ensina que o Código Não Ensina
Depois de construir sistemas assim, a conclusão mais honesta é que a parte técnica é a mais simples. Python, MT5, pandas, backtrader — tudo bem documentado e funcional.
O que é difícil: o mercado não se importa com a elegância do seu código. Mudanças de regime (trending para sideways, por exemplo) invalidam estratégias sem aviso. Custos ocultos — spread, slippage, corretagem, emolumentos, imposto de renda — são sistematicamente subestimados em backtests.
A tecnologia descrita aqui é sólida e adequada para construir sistemas de automação financeira sérios. O que não existe é uma estratégia garantidamente lucrativa esperando ser implementada. Se existisse, o mercado a arbitraria até ela deixar de funcionar.
Construa bem. Teste rigorosamente. Dimensione o risco de forma conservadora. Nunca opere com capital que não pode perder.