Robô de trading multithreaded com machine learning: da concepção à concretização

Este projeto nasceu da tentativa de reviver um dos meus códigos antigos. O código ficou por muito tempo guardado em discos rígidos empoeirados e, há bastante tempo, já não funciona corretamente: muita coisa mudou no ambiente de desenvolvimento ao longo desses anos. Decidi não apenas retomar o desenvolvimento, mas também contar esse processo passo a passo ao longo de vários artigos.

Inicialmente, a ideia original era desenvolver um robô multithreaded usando computação paralela de alto desempenho, talvez até com clusters na nuvem. Imagine a seguinte situação: você está sentado diante de vários monitores, observando o movimento dos pares de moedas, e percebe que o cérebro humano simplesmente não é capaz de acompanhar dezenas de instrumentos ao mesmo tempo, analisar centenas de atributos e tomar decisões na velocidade exigida pelos mercados modernos. É exatamente nesse momento que nasce a ideia de criar um robô de trading inteligente, capaz de operar com um portfólio de instrumentos e analisar décadas de dados em paralelo.

Mas por que exatamente Python, e não o MQL5 nativo? Essa pergunta certamente surge para qualquer pessoa que veja a arquitetura híbrida da nossa solução. A resposta está na natureza das tarefas que resolvemos. Machine learning exige bibliotecas poderosas como scikit-learn, XGBoost e pandas, um ecossistema que, em Python, parece ter sido criado para machine learning.

Outro motivo para essa escolha é que a arquitetura do nosso robô se baseia na filosofia de separação de responsabilidades e divisão de tarefas. O Python fica responsável pelo "cérebro": coleta de dados, pré-processamento, criação de atributos, treinamento dos modelos e geração dos sinais de trading. O MetaTrader 5, por meio de sua API para Python, atua como as "mãos": executa ordens, gerencia posições e fornece dados de mercado em tempo real.

O multithreading deixa de ser apenas uma solução técnica e passa a ser uma necessidade. Quando o robô precisa analisar simultaneamente EURUSD, GBPUSD, AUDUSD e outros pares, o processamento sequencial se transforma em um gargalo. Cada thread recebe seu próprio par de moedas, seu próprio modelo e seu próprio contexto de tomada de decisão. É semelhante ao modo como um trader experiente consegue manter várias ideias de trading em mente ao mesmo tempo, mas com a capacidade computacional de uma máquina.

A particularidade da nossa abordagem está na integração da gestão de risco no nível do portfólio. Em vez de cada instrumento ser operado de forma independente, definimos a variável global TOTAL_PORTFOLIO_RISK, que limita o risco total do portfólio. Isso significa que o tamanho da posição em cada instrumento é calculado não apenas com base em sua volatilidade, mas também levando em conta as posições já abertas em outros instrumentos.

TOTAL_PORTFOLIO_RISK: float = 500.0
POSITION_SIZES: Dict[str, float] = {}
SYMBOL_TRADES: Dict[str, bool] = {}

Sistema de logging e obtenção de dados de mercado

Antes de mergulhar nos algoritmos de machine learning, precisamos criar uma base confiável: um sistema de logging e obtenção de dados. Você já tentou depurar uma aplicação multithreaded sem logs? É como procurar uma agulha no palheiro de olhos vendados.

Nosso sistema de logging se baseia em uma fila de tamanho limitado, uma solução que pode parecer excessivamente complexa, mas faz muito sentido. Imagine a situação: cinco threads tentando escrever informações no console ao mesmo tempo. Sem coordenação, o resultado será uma confusão de mensagens misturadas. A fila garante a ordem das mensagens, enquanto uma thread dedicada à impressão mantém os logs legíveis.

log_queue = queue.Queue(maxsize=1000)

def log_printer():
    while True:
        try:
            log_message = log_queue.get(timeout=10)
            if log_message is None:
                break
            print(log_message)
        except queue.Empty:
            continue

Por que exatamente 1000 mensagens na fila? Isso vem da experiência prática. Uma fila pequena demais levaria à perda de mensagens importantes em momentos de alta atividade; uma fila grande demais, ao consumo excessivo de memória. 1000 mensagens é o equilíbrio ideal para um robô de trading.

Agora passamos ao coração do nosso sistema: a função retrieve_data. Ela parece simples, mas incorpora uma filosofia de confiabilidade. Observe a variável retries_limit = 300. Não é um número aleatório. No trading real, os servidores das corretoras podem ficar indisponíveis por segundos, às vezes por minutos. 300 tentativas com intervalo de um segundo nos dão uma janela de recuperação de 5 minutos.

Uma atenção especial é dedicada à criação dos indicadores técnicos. Não calculamos apenas RSI ou MACD: criamos todo um ecossistema de atributos interligados. O RSI mostra a sobrecompra ou a sobrevenda do ativo. O MACD revela mudanças no momentum. O ATR mede a volatilidade. Mas a mágica acontece quando esses indicadores interagem dentro do modelo de machine learning.

# RSI (Relative Strength Index)
delta = raw_data['close'].diff()
gain = delta.where(delta > 0, 0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss
raw_data['RSI_14'] = (100 - (100 / (1 + rs))).astype(np.float64)

Observe que usamos np.float64 em vez de float32. Isso pode parecer desperdício de memória, mas as funções subsequentes exigem alta precisão numérica. Economizar na precisão pode levar à instabilidade na clusterização de atributos.

Cada indicador técnico tem sua própria "personalidade". O Williams %R reage de forma agressiva às variações de preço. O Aroon Oscillator mostra a força da tendência. O Efficiency Ratio mede a "pureza" do movimento do preço. Quando todos esses indicadores são combinados em modelos de machine learning, eles criam uma representação multidimensional do mercado inacessível à percepção humana.

Data augmentation e criação de amostras sintéticas

Machine learning é voraz por dados, mas os mercados financeiros não são generosos em exemplos históricos. Como resolver esse dilema? A data augmentation passa a ser nossa chave para criar um conjunto de dados de treinamento rico a partir de um histórico limitado.

A função augment_data implementa quatro estratégias de expansão do volume de dados, cada uma com uma justificativa sólida. A primeira estratégia é a adição de ruído. Você pode perguntar: por que estragar dados limpos? A resposta está na natureza dos mercados reais. Os preços nunca se movem de forma perfeitamente suave: sempre existem microflutuações, slippage e picos aleatórios. Ao adicionar ruído controlado, preparamos o modelo para as condições reais.

noisy_data = raw_data.astype(np.float32) + np.random.normal(0, noise_level, raw_data.shape).astype(np.float32)

O parâmetro noise_level = 0.01 não foi escolhido por acaso. Isso corresponde a aproximadamente 1% do valor do atributo: o bastante para criar variabilidade, mas não a ponto de distorcer os padrões. Pense nisso como uma leve ondulação na água: ela não muda a direção da corrente, mas acrescenta realismo.

A segunda estratégia é o deslocamento temporal. Criamos uma cópia dos dados adiantada em uma hora. Isso pode parecer artificial, mas os padrões financeiros muitas vezes se repetem com pequenos deslocamentos no tempo. A volatilidade da manhã pode aparecer à noite, e ciclos semanais podem se deslocar em um ou dois dias.

A terceira estratégia é o reescalonamento. Multiplicamos todos os valores por um coeficiente aleatório entre 0.9 e 1.1. Isso simula diferentes condições de mercado: períodos de alta e baixa volatilidade, mercados de alta e de baixa. Um modelo treinado apenas com dados de um único período pode se tornar incapaz de lidar com uma mudança no regime de mercado.

A quarta estratégia é a inversão das colunas de preços. Multiplicamos os dados de preços por -1, o que, na prática, inverte o gráfico verticalmente. Isso ensina o modelo a reconhecer padrões espelhados. O padrão "ombro-cabeça-ombro", quando invertido, se torna um "ombro-cabeça-ombro invertido", mas as relações matemáticas entre os atributos são preservadas.

for col in price_columns:
    if col in inverted_data.columns:
        inverted_data[col] *= -1

É essencial processar corretamente os resultados da data augmentation. Operações em ponto flutuante podem gerar valores infinitos ou NaN. Nossa estratégia de limpeza usa medianas para preencher valores problemáticos, uma abordagem mais robusta do que as médias, que podem ser distorcidas por valores atípicos.

Após a data augmentation, o volume de dados aumenta cinco vezes. De 1000 horas históricas, obtemos 5000 exemplos de treinamento. Mas isso não é apenas um aumento quantitativo: é um enriquecimento qualitativo. O modelo vê não só o que aconteceu, mas também variações do que poderia ter acontecido em condições ligeiramente diferentes.

Damos atenção especial ao tratamento do volume de negociações. Ao contrário dos preços, o volume não pode ser negativo; por isso, na inversão, não o alteramos e, ao adicionar ruído, limitamos inferiormente a zero. Isso mantém a coerência econômica dos dados.

Rotulagem dos dados e criação das variáveis-alvo

Machine learning sem dados corretamente rotulados é como tentar ensinar uma criança a ler mostrando letras sem explicar seu significado. As funções markup_data e label_data implementam a etapa fundamental de transformar dados brutos de preço em exemplos de treinamento com rótulos claros da variável-alvo.

A função markup_data implementa a abordagem mais simples de rotulagem: ela prevê se o preço, após um período, estará acima do valor atual em uma margem definida por markup_ratio. Essa abordagem parece primitiva, mas sua simplicidade engana. O parâmetro markup_ratio = 0.00002, ou 2 pips para a maioria dos pares de moedas, leva em conta os custos reais de trading.

def markup_data(
    data: pd.DataFrame,
    target_column: str,
    label_column: str,
    markup_ratio: float = 0.00002
) -> pd.DataFrame:
    log("Starting markup_data function")
    data[label_column] = np.where(
        data[target_column].shift(-1) > data[target_column] + markup_ratio,
        1,
        0
    ).astype(np.int8)
    data.loc[data[label_column].isna(), label_column] = 0
    log(f"Number of labels set for price change greater than markup ratio: {data[label_column].sum()}")
    return data

A função label_data representa uma abordagem mais sofisticada. Em vez de simplesmente comparar os preços após um período, ela modela uma estratégia de trading real com stop-loss e take-profit. Isso é fundamental, porque os mercados financeiros são não lineares. O preço pode alcançar seu alvo, mas, no caminho, tocar o stop-loss.

def label_data(
    data: pd.DataFrame,
    symbol: str,
    min_days: int = 2,
    max_days: int = 72
) -> pd.DataFrame:
    if not mt5.initialize(path=TERMINAL_PATH):
        log("Terminal connection error")
        return data

    symbol_info = mt5.symbol_info(symbol)
    stop_level: float = 300 * symbol_info.point
    take_level: float = 800 * symbol_info.point
    labels: List[Optional[int]] = []

    for i in range(data.shape[0] - max_days):
        rand: int = random.randint(min_days, max_days)
        curr_pr: float = data['close'].iloc[i]
        future_pr: float = data['close'].iloc[i + rand]
        min_pr: float = data['low'].iloc[i:i + rand].min()
        max_pr: float = data['high'].iloc[i:i + rand].max()
        price_change: float = abs(future_pr - curr_pr)

        if (price_change > take_level and
            future_pr > curr_pr and
            min_pr > curr_pr - stop_level):
            labels.append(1)
        elif (price_change > take_level and
              future_pr < curr_pr and
              max_pr < curr_pr + stop_level):
            labels.append(0)
        else:
            labels.append(None)

O algoritmo de rotulagem funciona da seguinte maneira: para cada instante, escolhemos um horizonte aleatório entre min_days e max_days, de 2 a 72 horas. Em seguida, analisamos o comportamento do preço nesse intervalo. A lógica de verificação tem três níveis: se a variação do preço é suficientemente grande (price_change > take_level), se a direção está correta (future_pr > curr_pr para compras) e se não houve acionamento prematuro do stop-loss (min_pr > curr_pr - stop_level).

Por que usar horizontes de tempo aleatórios? Isso evita que o modelo se sobreajuste a intervalos temporais fixos. Os mercados não obedecem aos nossos períodos de análise: um padrão pode se desenvolver em 5 horas ou em 50. A aleatoriedade torna o modelo mais adaptável a diferentes escalas temporais dos movimentos de mercado.

data = data.iloc[:len(labels)].copy()
    data['labels'] = labels
    data.dropna(inplace=True)

    X = data.drop('labels', axis=1)
    y = data['labels']

    rus = RandomUnderSampler(random_state=2)
    X_balanced, y_balanced = rus.fit_resample(X, y)

    data_balanced = pd.concat([X_balanced, y_balanced], axis=1)
    log(f"Number of growth labels (1.0): {data_balanced['labels'].value_counts().get(1.0, 0)}")
    log(f"Number of decline labels (0.0): {data_balanced['labels'].value_counts().get(0.0, 0)}")
    return data_balanced

Os parâmetros stop_level = 300 pontos e take_level = 800 pontos definem uma relação risco/retorno de aproximadamente 1:2.67. Esses valores são multiplicados por symbol_info.point, o que os adapta automaticamente ao instrumento específico. Para o EURUSD, com point = 0.00001, isso significa um stop-loss de 300 pips e um take-profit de 800 pips.

O RandomUnderSampler resolve o problema do desbalanceamento de classes de forma radical: ele remove exemplos aleatórios da classe majoritária até alcançar o equilíbrio. Em dados financeiros, sempre há mais períodos "sem sinal claro" do que oportunidades de trading bem definidas. Sem balanceamento, o modelo aprenderá a prever "sem sinal" em 90% dos casos e terá boa acurácia formal, mas será inútil na prática.

Geração de atributos e clusterização com Gaussian Mixture Models

Após criar o conjunto básico de indicadores técnicos, surge a pergunta: como extrair ainda mais informação dos dados disponíveis? A função generate_new_features faz isso por meio da geração automática de atributos derivados, usando combinações aleatórias dos atributos já existentes.

A filosofia dessa abordagem é simples, mas poderosa: se dois atributos, isoladamente, contêm informações úteis, suas combinações matemáticas podem revelar padrões ocultos. Imagine que um indicador mostre a força da tendência, enquanto outro mostre sua estabilidade. O produto dos dois pode identificar momentos em que uma tendência forte começa a se tornar instável.

def generate_new_features(
    data: pd.DataFrame,
    num_features: int = 10,
    random_seed: int = 1
) -> pd.DataFrame:
    random.seed(random_seed)
    new_features: Dict[str, pd.Series] = {}

    columns = data.columns
    for i in range(num_features):
        feature_name = f'feature_{i}'
        col1_idx, col2_idx = random.sample(range(len(columns)), 2)
        col1, col2 = columns[col1_idx], columns[col2_idx]
        operation = random.choice([
            'add', 'subtract', 'multiply', 'divide',
            'shift', 'rolling_mean', 'rolling_std',
            'rolling_max', 'rolling_min', 'rolling_sum'
        ])

A aleatoriedade na escolha das colunas e das operações pode parecer caótica, mas garante diversidade. Uma abordagem determinística poderia deixar passar combinações pouco óbvias, porém úteis. A operação de divisão exige cuidado especial: adicionamos um pequeno valor, 1e-8, ao denominador para evitar divisão por zero.

elif operation == 'divide':
            new_features[feature_name] = (data[col1] / (data[col2] + 1e-8)).astype(np.float32)
        elif operation == 'shift':
            shift = random.randint(1, 10)
            new_features[feature_name] = data[col1].shift(shift).fillna(method='ffill').fillna(method='bfill').astype(np.float32)

As operações de defasagem criam atributos defasados com uma defasagem aleatória de 1 a 10 períodos. Isso permite que o modelo incorpore informações do passado em diferentes horizontes temporais. As estatísticas em janela móvel acrescentam versões suavizadas dos atributos originais, o que pode revelar tendências de longo prazo escondidas no ruído de curto prazo.

A etapa seguinte é a clusterização de atributos com Gaussian Mixture Models. Esta é uma das partes mais elegantes do nosso sistema. O GMM pressupõe que os dados são gerados por uma mistura de várias distribuições normais. No contexto dos mercados financeiros, isso significa identificar regimes: mercado de alta, mercado de baixa, movimento lateral, alta volatilidade e baixa volatilidade.

def cluster_features_by_gmm(
    data: pd.DataFrame,
    n_components: int = 6
) -> pd.DataFrame:
    X = data.drop(['label', 'labels'], axis=1, errors='ignore').astype(np.float32)
    X = X.replace([np.inf, -np.inf], np.nan).fillna(X.median(numeric_only=True))

    gmm = GaussianMixture(
        n_components=n_components,
        covariance_type='full',
        reg_covar=0.1,
        random_state=1
    )
    gmm.fit(X)
    data['cluster'] = gmm.predict(X).astype(np.int16)
    return data

O parâmetro n_components = 6 foi escolhido como um compromisso entre detalhamento e interpretabilidade. Clusters de menos significam perda de nuances significam perda de nuances; clusters demais, sobreajuste e instabilidade. Seis clusters permitem capturar os principais regimes de mercado: forte tendência de alta, tendência de alta fraca, movimento lateral com baixa volatilidade, movimento lateral com alta volatilidade, tendência de baixa fraca e forte tendência de baixa.

O parâmetro covariance_type='full' permite que o GMM modele as correlações entre os atributos. Isso é essencial em finanças, onde os atributos raramente são independentes. O RSI e o MACD podem se comportar de maneiras diferentes em condições de mercado distintas, e a matriz de covariância completa leva isso em conta.

A regularização da covariância (reg_covar=0.1) impede a degeneração do modelo quando os dados são quase linearmente dependentes. Em dados financeiros, situações desse tipo surgem com frequência, por exemplo, quando vários indicadores apontam valores extremos ao mesmo tempo.

A atribuição ao cluster se torna um novo atributo poderoso. O modelo de machine learning recebe informações não apenas sobre os valores específicos dos indicadores, mas também sobre o regime de mercado em que a situação atual se enquadra. Isso acrescenta informação contextual que pode mudar radicalmente a interpretação de outros atributos.

Eliminação recursiva de atributos e treinamento do XGBoost com ensemble

Tendo à disposição dezenas de indicadores técnicos, atributos gerados e rótulos de cluster, enfrentamos o problema clássico da "maldição da dimensionalidade". Mais atributos nem sempre significam um modelo melhor. Pelo contrário, atributos redundantes e correlacionados podem levar ao sobreajuste e reduzir a capacidade de generalização.

A função feature_engineering lida com esse dilema por meio da eliminação recursiva de atributos com validação cruzada (RFECV). Esse método funciona como um trader experiente, que gradualmente remove indicadores desnecessários do seu arsenal e mantém apenas os mais informativos.

def feature_engineering(
    data: pd.DataFrame,
    n_features_to_select: int = 15
) -> pd.DataFrame:
    X = data.drop(['label', 'labels'], axis=1, errors='ignore').astype(np.float32)
    y = data['labels'].astype(np.int8)
    X = X.replace([np.inf, -np.inf], np.nan).fillna(X.median(numeric_only=True))

    unique_classes = y.unique()
    if len(unique_classes) < 2:
        log(f"Error in feature_engineering: Only one class found in labels: {unique_classes}")
        raise ValueError(f"The target 'y' needs to have more than 1 class. Got {len(unique_classes)} class instead")

    clf = RandomForestClassifier(n_estimators=100, random_state=1)
    rfecv = RFECV(
        estimator=clf,
        step=1,
        cv=5,
        scoring='accuracy',
        n_jobs=1,
        verbose=1,
        min_features_to_select=n_features_to_select
    )
    rfecv.fit(X, y)

O RFECV funciona de forma iterativa: começa com todos os atributos, treina um modelo, estima a importância de cada um, remove o menos importante e repete o ciclo. O parâmetro step=1 significa que, em cada iteração, apenas um atributo é removido: devagar, mas com precisão. Uma alternativa seria usar step=5 para acelerar, mas, em finanças, o custo de um erro é alto demais.

A escolha do RandomForestClassifier como estimador-base não é aleatória. A floresta aleatória funciona bem com atributos heterogêneos e é robusta a valores atípicos. Seu feature_importances_ se baseia na redução da Gini impurity, o que fornece uma métrica de importância intuitiva.

selected_features = X.columns[rfecv.get_support()]
    selected_data = data[selected_features.tolist() + ['label', 'labels']]
    log("\nBest features:")
    log(str(pd.DataFrame({'Feature': selected_features})))
    return selected_data

Passamos agora ao coração do nosso sistema: o treinamento do XGBoost com ensemble. O XGBoost é reconhecido por sua eficiência em problemas financeiros, graças à sua capacidade de modelar interações não lineares entre atributos e à sua robustez ao sobreajuste.

def train_xgboost_classifier(
    data: pd.DataFrame,
    num_boost_rounds: int = 500
) -> BaggingClassifier:
    if data.empty:
        raise ValueError("Data should not be empty")

    X = data.drop(['label', 'labels'], axis=1).astype(np.float32)
    y = data['labels'].astype(np.int8)

    clf = xgb.XGBClassifier(
        objective='binary:logistic',
        random_state=1,
        max_depth=5,
        learning_rate=0.2,
        n_estimators=300,
        subsample=0.01,
        colsample_bytree=0.1,
        reg_alpha=1,
        reg_lambda=1
    )

Os parâmetros do XGBoost foram escolhidos para trabalhar com dados financeiros: max_depth=5 limita a profundidade das árvores, evitando sobreajuste em dados ruidosos; learning_rate=0.2 garante um treinamento rápido o suficiente mantendo a estabilidade. Os parâmetros de regularização são essenciais: subsample=0.01 significa que cada árvore é treinada com apenas 1% dos dados, o que reduz radicalmente o sobreajuste; colsample_bytree=0.1 indica que cada árvore usa apenas 10% dos atributos. Isso pode parecer extremo, mas, em combinação com um grande número de árvores (n_estimators=300), cria um ensemble poderoso de aprendizes fracos diversificados.

bagging_clf = BaggingClassifier(estimator=clf, random_state=1)
    param_grid = {
        'n_estimators': [10, 20, 30],
        'max_samples': [0.5, 0.7, 1.0],
        'max_features': [0.5, 0.7, 1.0]
    }

    grid_search = GridSearchCV(bagging_clf, param_grid, cv=5, scoring='accuracy')
    grid_search.fit(X, y)
    accuracy = grid_search.best_score_
    log(f"Average cross-validation accuracy: {accuracy:.2f}")
    return grid_search.best_estimator_

Sim, esqueci de acrescentar: o BaggingClassifier adiciona mais uma camada de ensemble acima do XGBoost. Criamos um meta-ensemble: cada elemento já é um ensemble de gradient boosting, e o bagging combina vários desses modelos XGBoost. Esse ensemble em dois níveis proporciona robustez excepcional ao sobreajuste.

O GridSearchCV seleciona automaticamente os parâmetros ideais do bagging por meio de validação cruzada. O max_samples controla qual fração dos dados é usada para treinar cada modelo no ensemble, enquanto o max_features determina a fração de atributos disponível para cada modelo. A combinação desses parâmetros cria diversidade no ensemble, a chave para sua eficiência.

Gestão de risco no nível do portfólio e cálculo dinâmico do tamanho das posições

Imagine a seguinte situação: seu robô apresenta excelentes resultados em cada instrumento separadamente, mas, ao colocar o portfólio em operação, ocorre uma catástrofe. A razão é simples: falta de coordenação entre as posições. Cinco operações simultâneas de 1 lote cada, em condições de alta correlação entre pares de moedas, podem criar um risco equivalente a uma única posição de 5 lotes.

A função calculate_portfolio_position_sizes revoluciona a abordagem à gestão de risco. Em vez de usar tamanhos de posição fixos, nós os calculamos dinamicamente, com base na volatilidade atual de cada instrumento e no limite total de risco do portfólio.

def calculate_portfolio_position_sizes(symbols: List[str]) -> None:
    global POSITION_SIZES
    
    if not mt5.initialize(path=TERMINAL_PATH):
        log("Ошибка подключения к терминалу для расчета позиций")
        return
    
    try:
        symbol_risks = {}
        for symbol in symbols:
            symbol_info = mt5.symbol_info(symbol)
            if symbol_info is None:
                continue
                
            # Получаем ATR за последние 14 дней
            rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_D1, 0, 20)
            if rates is None or len(rates) < 14:
                continue
                
            df = pd.DataFrame(rates)
            df['tr'] = np.maximum(df['high'] - df['low'], 
                                 np.maximum(abs(df['high'] - df['close'].shift(1)),
                                           abs(df['low'] - df['close'].shift(1))))
            atr = df['tr'].rolling(14).mean().iloc[-1]
            
            # Риск на стандартный лот
            risk_per_lot = (atr / symbol_info.point) * symbol_info.trade_tick_value
            symbol_risks[symbol] = risk_per_lot

O ATR (Average True Range) se torna a base do nosso sistema de gestão de risco. Esse indicador mede a volatilidade média do instrumento ao longo de 14 dias, levando em conta não apenas as oscilações diárias, mas também os gaps entre o fechamento de um período e a abertura do período seguinte. O True Range de cada dia é calculado como o máximo entre três valores: a diferença entre a máxima do dia e a mínima do dia, a diferença entre a máxima e o fechamento anterior, e a diferença entre o fechamento anterior e a mínima do dia.

A fórmula risk_per_lot = (atr / symbol_info.point) * symbol_info.trade_tick_value converte a volatilidade em valor monetário. Se o ATR do EURUSD for 0.00080 (80 pips), o risco de um lote padrão será de 80 dólares, considerando o valor de 1 dólar por pip.

# Риск на инструмент
        risk_per_instrument = TOTAL_PORTFOLIO_RISK / len(symbols)
        
        # Рассчитываем размеры позиций
        for symbol, risk_per_lot in symbol_risks.items():
            symbol_info = mt5.symbol_info(symbol)
            base_size = risk_per_instrument / risk_per_lot
            
            # Нормализуем под требования брокера
            min_lot = symbol_info.volume_min
            lot_step = symbol_info.volume_step
            normalized_size = max(min_lot, round(base_size / lot_step) * lot_step)
            
            POSITION_SIZES[symbol] = normalized_size
            SYMBOL_TRADES[symbol] = False
            
            log(f"{symbol}: лот={normalized_size:.3f}, риск={normalized_size * risk_per_lot:.2f}")

A distribuição uniforme do risco entre os instrumentos (risk_per_instrument = TOTAL_PORTFOLIO_RISK / len(symbols)) pode parecer simplificada, mas garante o equilíbrio do portfólio. Com TOTAL_PORTFOLIO_RISK = 500 dólares e cinco pares de moedas, cada instrumento recebe um limite de risco de 100 dólares.

Adequar o tamanho da posição às exigências da corretora é essencial. Na prática, as corretoras impõem restrições: lote mínimo (geralmente 0.01) e incremento do lote (geralmente 0.01). A função round(base_size / lot_step) * lot_step ajusta o tamanho calculado ao valor permitido mais próximo.

O sistema de flags SYMBOL_TRADES impede a abertura de múltiplas posições no mesmo instrumento. Isso é especialmente importante em um ambiente multithreaded, no qual cada thread pode receber um sinal de trading de forma independente.

def check_and_update_position_flags() -> None:
    global SYMBOL_TRADES
    
    if not mt5.initialize(path=TERMINAL_PATH):
        return
    
    try:
        positions = mt5.positions_get()
        if positions is None:
            positions = []
        
        open_symbols = set()
        for pos in positions:
            if pos.magic == 123456:  # Наш magic number
                open_symbols.add(pos.symbol)
        
        # Сбрасываем флаги для символов без открытых позиций
        for symbol in list(SYMBOL_TRADES.keys()):
            if symbol not in open_symbols and SYMBOL_TRADES[symbol]:
                SYMBOL_TRADES[symbol] = False
                log(f"Флаг сброшен для {symbol} - позиция закрыта")

A função check_and_update_position_flags garante a sincronização entre o estado interno do robô e as posições reais no terminal. Uma posição pode ser fechada manualmente pelo trader, por stop-loss ou por take-profit. Sem verificação regular, o robô pode "pensar" que a posição ainda está aberta e deixar passar novas oportunidades de trading.

O magic number 123456 serve para identificar nossas posições em meio a posições de outros robôs de trading ou operações manuais. Isso garante que gerenciemos apenas as posições que nós mesmos abrimos.

Lógica de trading e execução de ordens em tempo real

O momento da verdade de qualquer robô de trading chega no momento da execução das ordens. Todas as etapas anteriores: coleta de dados, criação de atributos e treinamento dos modelos, servem a um único objetivo: tomar a decisão de "comprar", "vender" ou "esperar". A função online_trading materializa essas decisões em posições reais de mercado.

def online_trading(
    symbol: str,
    features: np.ndarray,
    model: Any
) -> Optional[Any]:
    global SYMBOL_TRADES, POSITION_SIZES
    
    # Проверяем и обновляем флаги позиций
    check_and_update_position_flags()
    
    if not mt5.initialize(path=TERMINAL_PATH):
        log("Error: Failed to connect to MetaTrader 5 terminal")
        return None

    # Проверяем, открыта ли уже позиция по этому символу
    if SYMBOL_TRADES.get(symbol, False):
        log(f"Позиция по {symbol} уже открыта")
        mt5.shutdown()
        return None

    # Получаем размер позиции из расчета портфеля
    volume = POSITION_SIZES.get(symbol, 0.1)

As primeiras linhas da função demonstram programação defensiva. A verificação das flags de posição evita a abertura de operações duplicadas. A inicialização do terminal pode falhar inesperadamente devido a problemas de rede ou a uma reinicialização. O uso do tamanho da posição a partir dos POSITION_SIZES pré-calculados garante conformidade com a estratégia geral de gestão de risco.

O laço de obtenção das informações do instrumento merece atenção especial. No trading real, symbol_info pode retornar None temporariamente por problemas de conexão com o servidor da corretora. Em vez de desistir imediatamente, o robô realiza até 30000 tentativas em intervalos de 5 segundos.

attempts: int = 30000
    symbol_info = None
    for _ in range(attempts):
        symbol_info = mt5.symbol_info(symbol)
        if symbol_info is not None:
            break
        log(f"Error: Instrument not found. Attempt {_ + 1} of {attempts}")
        time.sleep(5)

    if symbol_info is None:
        mt5.shutdown()
        return None

    tick = mt5.symbol_info_tick(symbol)
    price_bid: float = tick.bid
    price_ask: float = tick.ask
    signal = model.predict(features)
    positions_total: int = mt5.positions_total()

30000 tentativas podem parecer excessivas, mas, no contexto do trading algorítmico, isso representa aproximadamente 41 horas de tentativas contínuas. Os mercados funcionam 24 horas por dia, e problemas técnicos de curta duração não devem fazer o robô perder oportunidades de trading.

A lógica de tomada de decisões de trading é elegante em sua simplicidade. O modelo retorna uma probabilidade, e usamos o threshold de 0.5 para separar compras e vendas. A verificação adicional positions_total < MAX_OPEN_TRADES evita exposição excessiva.

request = None
    if positions_total < MAX_OPEN_TRADES and signal[-1] > 0.5:
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": symbol,
            "volume": volume,
            "type": mt5.ORDER_TYPE_BUY,
            "price": price_ask,
            "sl": price_ask - 50 * symbol_info.point,
            "tp": price_ask + 100 * symbol_info.point,
            "deviation": 20,
            "magic": 123456,
            "comment": "Portfolio Buy",
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_FOK,
        }

A estrutura da requisição de ordem contém vários parâmetros essenciais. TRADE_ACTION_DEAL significa execução imediata ao preço de mercado. ORDER_TYPE_BUY/SELL define a direção. O stop-loss de 50 pontos e o take-profit de 100 pontos criam uma relação risco/recompensa de 1:2.

O parâmetro deviation=20 permite que a corretora execute a ordem com desvio de até 20 pontos em relação ao preço solicitado. Em mercados em movimento rápido, isso evita a rejeição das ordens devido a variações microscópicas de preço entre o momento em que a cotação é recebida e o envio da ordem.

ORDER_TIME_GTC (Good Till Cancelled) significa que a ordem permanece válida até ser cancelada. ORDER_FILLING_FOK (Fill Or Kill) exige a execução integral da ordem ou sua rejeição: a execução parcial não é permitida.

for _ in range(attempts):
        result = mt5.order_send(request)
        if result.retcode == mt5.TRADE_RETCODE_DONE:
            # Помечаем, что по этому символу позиция открыта
            SYMBOL_TRADES[symbol] = True
            log(f"Позиция открыта {symbol}: {'Buy' if signal[-1] > 0.5 else 'Sell'}, лот={volume}")
            mt5.shutdown()
            return result.order
        log(f"Error: Trade request not executed, retcode={result.retcode}. Attempt {_ + 1}/{attempts}")
        time.sleep(3)

O laço de execução da ordem também realiza múltiplas tentativas. TRADE_RETCODE_DONE significa execução bem-sucedida. Outros códigos podem indicar fundos insuficientes, problemas de liquidez ou falhas técnicas. Intervalos de 3 segundos entre as tentativas dão ao mercado tempo para se estabilizar.

Após a execução bem-sucedida, definimos a flag SYMBOL_TRADES[symbol] = True, o que é essencial para evitar a abertura de posições duplicadas em um ambiente multithreaded. O logging, registrando a direção da posição e o volume, garante rastreabilidade completa da atividade de trading.

Arquitetura multithreaded e coordenação das threads

Imagine uma orquestra em que cada músico toca sua própria parte, mas todos seguem o mesmo maestro. Nossa arquitetura multithreaded funciona pelo mesmo princípio: cada thread processa seu próprio par de moedas de forma independente, mas todas se coordenam por meio de variáveis globais e funções compartilhadas.

A função process_symbol atua como o principal "executor" de cada thread. Ela encapsula todo o ciclo de vida do processamento de um instrumento: da obtenção dos dados históricos ao laço infinito de trading em tempo real.

def process_symbol(symbol: str) -> None:
    global all_symbols_done
    try:
        raw_data = retrieve_data(symbol)
        if raw_data is None:
            log(f"Data not found for symbol {symbol}")
            return

        labeled_data_engineered = process_data(raw_data)
        train_data = labeled_data_engineered[labeled_data_engineered.index <= FORWARD]
        test_data = labeled_data_engineered[labeled_data_engineered.index > FORWARD]

        if train_data.empty or len(train_data['labels'].unique()) < 2:
            log(f"Skipping symbol {symbol}: Insufficient data or single class in labels")
            return

        xgb_clf = train_xgboost_classifier(train_data, num_boost_rounds=1000)
        test_features = test_data.drop(['label', 'labels'], axis=1)
        test_labels = test_data['labels']
        accuracy = evaluate_xgboost_classifier(xgb_clf, test_features, test_labels)
        log(f"Accuracy for symbol {symbol}: {accuracy * 100:.2f}%")

A divisão dos dados em conjuntos de treino e teste com base no eixo temporal é essencial em finanças. O ponto de corte FORWARD = datetime(2025, 1, 1) significa que o modelo é treinado com dados anteriores a 2025 e testado em dados posteriores. Isso simula condições reais, nas quais o modelo precisa prever o futuro com base no passado.

A verificação len(train_data['labels'].unique()) < 2 evita tentativas de treinamento com dados de uma única classe. Em períodos de baixa volatilidade, todos os exemplos podem receber o rótulo "sem sinal", o que torna o treinamento impossível.

O laço infinito de trading começa após o treinamento bem-sucedido do modelo:

features = test_features.values
        position_id = None
        while not all_symbols_done:
            position_id = online_trading(symbol, features, xgb_clf)
            time.sleep(6)

        all_symbols_done = True
    except Exception as e:
        log(f"Error processing symbol {symbol}: {e}")

A variável global all_symbols_done serve como mecanismo de encerramento elegante de todas as threads. Qualquer thread pode definir essa flag como True, sinalizando às demais threads que devem encerrar a execução. Isso pode acontecer em caso de erro crítico, quando o limite diário de perdas for atingido ou por comando do usuário.

O intervalo time.sleep(6) entre os laços de trading foi escolhido como um compromisso entre tempo de resposta e carga no sistema. Intervalos curtos demais criam carga excessiva sobre a API da corretora e podem levar ao bloqueio das requisições; intervalos longos demais podem levar à perda de oportunidades de trading.

A função principal coordena a criação e o encerramento das threads:

if __name__ == "__main__":
    symbols = ["EURUSD", "GBPUSD", "AUDUSD", "NZDUSD", "USDCAD"]
    
    # Сначала рассчитываем размеры позиций для всего портфеля
    calculate_portfolio_position_sizes(symbols)
    
    threads = []
    for symbol in symbols:
        thread = threading.Thread(target=process_symbol, args=(symbol,))
        thread.start()
        threads.append(thread)

    for thread in threads:
        thread.join()

    log_queue.put(None)

A chamada calculate_portfolio_position_sizes(symbols) antes de iniciar as threads garante que todos os tamanhos de posição sejam calculados e estejam disponíveis em POSITION_SIZES antes do início das operações. Isso evita situações em que uma thread tenta operar com um tamanho de posição indefinido.

O método thread.join() para cada thread garante o encerramento correto do programa. A thread principal aguarda o término de todas as threads filhas antes de finalizar. O log_queue.put(None) final sinaliza à thread de logging que ela deve encerrar sua execução.

A escolha dos pares específicos de moedas ["EURUSD", "GBPUSD", "AUDUSD", "NZDUSD", "USDCAD"] não é aleatória. Todos os pares incluem o dólar dos EUA como moeda base ou moeda cotada, o que proporciona certo grau de correlação, mas também diversidade suficiente para a diversificação dos riscos.

A arquitetura demonstra uma solução elegante para um problema clássico da programação paralela: como garantir a independência das threads quando a coordenação ainda é necessária. As variáveis globais POSITION_SIZES, SYMBOL_TRADES e all_symbols_done servem como canais de comunicação entre as threads, enquanto o sistema de logging por meio de uma fila evita conflitos na saída simultânea de mensagens.

O tratamento de exceções em cada thread é essencial. Uma exceção não tratada em uma thread não deve fazer todo o sistema crashar. Cada thread "morre com dignidade", registrando o erro no log e liberando recursos.

Teste do sistema e planos ambiciosos de desenvolvimento

A função test_model atual demonstra uma abordagem básica para a validação de estratégias de trading, simulando o P&L com dados históricos com base no markup e nos horizontes temporais de saída:

def test_model(
    model: Any,
    X_test: pd.DataFrame,
    y_test: pd.Series,
    markup: float,
    initial_balance: float = 10000.0,
    point_cost: float = 0.00001
) -> None:
    balance = initial_balance
    trades = 0
    profits = []

    predicted_labels = model.predict(X_test.values)
    close = X_test['close'].values

    for i in range(len(predicted_labels) - 10):
        entry_price = close[i]
        exit_price = close[i + 10]
        # Детальная торговая логика с учетом направления и markup
        ...
    total_profit = balance - initial_balance
    log(f"Total accumulated profit or loss: {total_profit:.2f}")

Mas isso é apenas o começo. O futuro real do sistema está no escalonamento radical e na passagem para um patamar qualitativamente novo.

Por enquanto, a acurácia do sistema no trecho de teste fora da amostra fica entre 64% e 65%, enquanto o lucro médio é muito maior do que a perda média.

Foi isto que consegui obter em alguns dias de trading:

As simulações de rentabilidade com um win rate definido e risco calculado tanto para o portfólio de posições quanto para uma posição isolada, também são muito animadoras:

Ainda assim, não confio muito em testadores caseiros em Python. Por isso, será necessário testar o sistema por bastante tempo em trading real ou reescrever a lógica em MQL5 para ter acesso ao poderoso testador de estratégias com base de dados de ticks reais e emulação de requotes.

Migração completa para MQL5: uma revolução arquitetural

A arquitetura híbrida entre Python e MetaTrader 5 tem limitações fundamentais de velocidade. A migração completa para MQL5 exigirá a criação de bibliotecas próprias de machine learning. Os principais módulos incluem operações matriciais de alto desempenho para o XGBoost, uma implementação de boosting de gradiente com otimização nativa para arquitetura x64 e uma versão adaptada de Gaussian Mixture Models para clusterização de atributos. É uma tarefa tecnicamente complexa, mas eliminará toda a latência da API e trará um ganho de desempenho de 10 a 50 vezes.

Interface gráfica de próxima geração

Está prevista a criação de uma interface gráfica completa em tempo real, com arquitetura modular. O dashboard exibirá as posições atuais de todos os instrumentos, o P&L em tempo real, o uso dos limites de risco e o estado de cada thread de trading. O módulo de gestão dinâmica de parâmetros será essencial: a possibilidade de alterar TOTAL_PORTFOLIO_RISK, thresholds dos modelos e horizontes temporais sem reiniciar o sistema. O sistema de alertas incluirá notificações quando houver ultrapassagem dos limites diários de perdas, sinais anômalos dos modelos ou perda de conexão com a corretora.

Risk manager de nível superior: um salto quântico na gestão de portfólio

O sistema atual de gestão de risco parece primitivo em comparação com o planejado. O novo risk manager analisará em tempo real as correlações entre os instrumentos, usando janelas móveis de diferentes tamanhos. Os cálculos de Value-at-Risk baseados em simulações de Monte Carlo preverão as perdas potenciais do portfólio com diferentes níveis de probabilidade. O sistema de hedge dinâmico abrirá automaticamente posições compensatórias quando o risco de correlação ultrapassar os limites definidos.

Treinamento multithreaded: um exército de modelos

A transição para o treinamento paralelizado em massa de modelos será revolucionária. Cada thread de CPU testará diferentes combinações de hiperparâmetros do XGBoost, períodos de treinamento e conjuntos de atributos. O sistema escolherá automaticamente o melhor modelo para cada período temporal com base nos resultados fora da amostra. O ensemble com Random Forest, SVM, Neural Networks e LightGBM criará um metamodelo superior a qualquer modelo individual.

GAN para geração de datasets de 25 anos: uma história sintética dos mercados

O plano mais ambicioso é usar Generative Adversarial Networks para criar séries de preços sintéticas de 25 anos dos pares de moedas majors. A GAN será treinada em dados históricos reais e, em seguida, gerará datasets enormes, preservando as propriedades estatísticas dos mercados reais: clusterização de volatilidade, caudas pesadas nas distribuições e autocorrelações. Isso resolverá o problema da escassez de dados para aprendizado profundo.

Ensemble de 100 modelos especializados: segmentação do espaço temporal-preço

A culminação será a criação de um ensemble de 100 modelos, em que cada um é responsável por um "segmento" específico das condições de mercado. Cada modelo será treinado com dados de um regime específico: alta volatilidade, movimento lateral, tendências fortes, períodos de notícias e sessões pré-feriado. Regimes desse tipo podem ser obtidos, por exemplo, por meio de clusterização de atributos, cadeias de Markov ou modelo de espaço de estado (SSM). O BaggingClassifier de cada modelo garantirá robustez adicional. O metaclassificador determinará o regime de mercado atual com altíssima precisão e ativará os modelos especializados correspondentes. Essa arquitetura permitirá alcançar uma precisão preditiva sem precedentes, adaptando-se em tempo real a quaisquer condições de mercado.

Integração com Google Cloud Platform: a capacidade computacional de supercomputadores com mil núcleos

Os recursos computacionais locais se tornam um gargalo na execução de planos tão ambiciosos. A Google Cloud Platform oferece uma solução por meio de integração direta com a API para Python. A biblioteca google-cloud-compute permite criar programaticamente máquinas virtuais com centenas de núcleos de processamento e terabytes de memória para o treinamento paralelo de modelos.

from google.cloud import compute_v1
from google.cloud import storage

# Создание высокопроизводительного кластера
instance_client = compute_v1.InstancesClient()
operation = instance_client.insert(
    project="trading-ai-project",
    zone="us-central1-a", 
    instance_resource={
        "name": "trading-cluster-node",
        "machine_type": "zones/us-central1-a/machineTypes/c2-standard-60",
        "scheduling": {"preemptible": True}  # Снижение стоимости в 5 раз
    }
)

O AI Platform Training permite executar treinamento distribuído de XGBoost e redes neurais em dezenas de máquinas simultaneamente. O Cloud Storage permite armazenar petabytes de dados históricos com acesso instantâneo. O BigQuery ML se integra diretamente aos algoritmos de trading para análise de padrões em tempo real.

Uma vantagem crítica é a escalabilidade automática dos recursos computacionais. Durante períodos de retreinamento dos modelos, o sistema provisiona automaticamente centenas de instâncias e, em períodos mais tranquilos, reduz a capacidade alocada ao mínimo, otimizando os custos. As Preemptible instances reduzem os custos de computação em 5 vezes, o que torna economicamente viável treinar nossos futuros modelos ultracomplexos.

Conclusão

Neste artigo, demos os primeiros passos rumo à criação de um robô de trading multithreaded usando métodos de machine learning. A arquitetura analisada abrange todo o pipeline: da coleta e data augmentation à geração de atributos, clusterização, treinamento de modelos XGBoost e integração da gestão de risco no nível do portfólio. Vimos como a divisão de papéis entre Python e MetaTrader 5 permite usar as melhores ferramentas dos dois ambientes: a flexibilidade e o rico ecossistema de bibliotecas Python para machine learning e a confiabilidade do MetaTrader 5 para executar ordens em tempo real.

No entanto, esta é apenas a primeira parte de uma série de artigos. Nas próximas publicações, pretendo abordar em detalhes a migração do sistema para MQL5, o desenvolvimento da interface gráfica, a integração com serviços em nuvem, o uso de GAN para geração de datasets sintéticos e a construção de ensembles de modelos especializados. Essas etapas permitirão ampliar significativamente a escala e aumentar a eficiência do robô de trading, abrindo caminho para sua plena concretização.