Integrando o MQL5 com pacotes de processamento de dados (Parte 2): Aprendizado de Máquina e Análise Preditiva

Introdução

Neste artigo, focamos especificamente em Aprendizado de Máquina (ML) e Análise Preditiva. Pacotes de processamento de dados abrem novas fronteiras para traders quantitativos e analistas financeiros. Ao incorporar capacidades de aprendizado de máquina no MQL5, traders podem elevar suas estratégias de trading de sistemas baseados em regras tradicionais para modelos sofisticados, orientados por dados, que se adaptam continuamente às condições de mercado em evolução.

O processo envolve o uso das poderosas bibliotecas de processamento de dados e aprendizado de máquina do Python, como o scikit-learn, em conjunto com o MQL5. Essa integração permite que traders treinem modelos preditivos usando dados históricos, testem sua eficácia com técnicas de back-testing e, em seguida, implementem esses modelos para tomar decisões de trading em tempo real. A flexibilidade para combinar essas ferramentas permite a criação de estratégias que vão além dos indicadores técnicos típicos, incorporando análise preditiva e reconhecimento de padrões que podem aprimorar significativamente os resultados das operações.

Coletar Dados Históricos

Para começar, precisamos dos dados históricos do MetaTrader 5 salvos no formato .csv. Para isso, basta abrir sua plataforma MetaTrader e, no topo do painel do MetaTrader 5, navegar até > Ferramentas e depois > Opções, e você será direcionado para as opções de Gráficos. Em seguida, você deverá selecionar a quantidade de Barras no gráfico que deseja baixar. É melhor escolher a opção de barras ilimitadas, pois estaremos trabalhando com data e não saberíamos quantas barras há em um determinado período de tempo.

Depois disso, você precisará baixar os dados reais. Para fazer isso, navegue até > Exibir e depois para > Símbolos, e você será direcionado para a aba Especificações. Simplesmente navegue até > Barras ou Tiques, dependendo do tipo de dado que deseja baixar. Prossiga e insira o período de data inicial e final dos dados históricos que você deseja baixar. Após isso, clique no botão de solicitação para baixar os dados e salvá-los no formato .csv.

Após todos esses passos, você terá baixado com sucesso os dados históricos da sua plataforma MetaTrader. Agora, você precisa baixar e configurar o ambiente Jupyter Lab para análise. Para baixar e configurar o Jupyter Lab, você pode acessar o site oficial aqui e seguir os passos simples para fazer o download. Dependendo do tipo de sistema operacional que você usa, você terá uma variedade de opções para instalar utilizando pip, conda ou brew.

Processar Dados Históricos do MetaTrader 5 no Jupyter Lab

Para carregar com sucesso os dados históricos do MetaTrader 5 no Jupyter Lab, você precisará saber a pasta que selecionou para baixar os dados. Em seguida, no Jupyter Lab, navegue até essa pasta. Para começar, você deverá carregar os dados e inspecionar os nomes das colunas. Precisamos inspecionar os nomes das colunas para que possamos manipulá-las corretamente e evitar erros que possam surgir ao utilizarmos nomes incorretos.

Código Python:

import pandas as pd

# assign variable to the historical data
file_path = '/home/int_junkie/Documents/ML/predi/XAUUSD.m_H1_201510010000_202408052300.csv'

data = pd.read_csv(file_path, delimiter='\t')

# Display the first few rows and column names
print(data.head())
print(data.columns)

Saída:

Estamos trabalhando com dados históricos do MetaTrader 5 do ano de 2015 até 2024, ou seja, cerca de 9 anos de dados históricos. Esse tipo de dado ajuda a capturar grandes ciclos de mercado. O conjunto de dados provavelmente capturará diferentes fases do mercado, o que permitirá uma melhor compreensão e modelagem desses ciclos. Conjuntos de dados mais longos reduzem a probabilidade de overfitting, proporcionando uma gama mais abrangente de cenários.

Um modelo treinado com um conjunto de dados mais amplo tem maior chance de generalizar bem para dados não vistos, especialmente se o conjunto de dados for de menor período, como 1H. Isso é particularmente importante na análise de séries temporais, onde ter mais observações aumenta a confiabilidade dos resultados. Por exemplo, você pode detectar tendências seculares (direções de mercado de longo prazo) ou efeitos sazonais recorrentes que são relevantes para previsões.

Gráfico de Linhas com Dados Históricos

data.plot.line(y = "<CLOSE>", x = "<DATE>", use_index = True)

Saída:

Usamos o código acima para visualizar dados de séries temporais, como ativos financeiros ao longo do tempo. Usamos o código acima para visualizar dados de séries temporais, como ativos financeiros ao longo do tempo.

del data["<VOL>"]
del data["<SPREAD>"]

Em seguida, deletamos as colunas especificadas dos nossos dados históricos, utilizando a biblioteca pandas para deletá-las.

data.head()

Saída:

A partir da saída acima, podemos ver que de fato as colunas especificadas foram deletadas.

# We add a colunm for tommorows price
data["<NexH>"] = data["<CLOSE>"].shift(-1)

1. 'data["<NexH>"]':

• Isso adiciona uma nova coluna chamada '"<NexH>"' (PRÓXIMA HORA) ao DataFrame 'data'. O valor dessa coluna representará os preços de fechamento para a próxima hora em relação a cada linha.

2. 'data["<CLOSE>"].shift(-1)':

• 'data["<CLOSE>"]' refere-se à coluna existente no DataFrame que contém os preços de fechamento para cada data e hora.
• O método '.shift(-1)' desloca a data e hora na coluna "<CLOSE>" para cima em 1 linha (porque o argumento é '-1'), o que efetivamente move cada valor para a linha anterior.
• Como resultado, o valor que originalmente correspondia a uma data específica agora aparecerá na linha correspondente à data anterior.

Saída:

data["<TRGT>"] = (data["<NexH>"] > data["<CLOSE>"]).astype(int)

data

Em seguida, usamos o código acima para criar uma nova coluna no DataFrame 'data' que contém valores binários (0 ou 1) indicando se o preço máximo do próximo período ('"NexH"') é maior que o preço de fechamento atual ('"<CLOSE>"').

1. 'data["<TRGT>"]':

• Essa é a nova coluna chamada '"<TRGT>"' (ALVO) no DataFrame 'data'. Essa coluna armazenará os valores binários do alvo (0 ou 1) com base em uma condição.

2. '(data["<NexH>"] > data["<CLOSE>"])':

• Essa expressão compara o valor da coluna '"<NexH>"' (preço máximo do próximo período) com o valor da coluna '"<CLOSE>"' (preço de fechamento atual) para cada linha.
• Essa é uma série booleana, onde cada valor é 'True' (se o próximo máximo for maior que o fechamento atual) ou 'False' (se não for).

3. '.astype(int)':

• Essa função converte os valores booleanos ('True' ou 'False') em inteiros ('1' ou '0', respectivamente).
• 'True' se torna '1', e 'False' se torna '0'.

Saída:

from sklearn.ensemble import RandomForestClassifier

model = RandomForestClassifier(n_estimators = 50, min_samples_split = 50, random_state = 1)

train = data.iloc[:-50]
test = data.iloc[-50:]

predictors = ["<CLOSE>","<TICKVOL>", "<OPEN>", "<HIGH>", "<LOW>"]
model.fit(train[predictors], train["<TRGT>"])

Saída:

1. Importando o Classificador Random Forest:

• O 'Random-Forest-Classifier' é um modelo de aprendizado de máquina baseado em assembleia, que constrói várias árvores de decisão e combina suas saídas para melhorar a precisão preditiva e controlar o overfitting.

2. Inicialização do Modelo:

• 'estimator' especifica o número de árvores de decisão na floresta. Neste caso, o modelo construirá 50 árvores.
• 'min_sample_split' define o número mínimo de amostras necessárias para dividir um nó interno. Um valor mais alto reduz o overfitting, garantindo que as divisões ocorram somente quando houver dados suficientes disponíveis.
• 'random_state' fixa a semente aleatória para garantir a reprodutibilidade dos resultados. Usar a mesma semente (por exemplo, '1') gerará os mesmos resultados sempre que o código for executado.

3. Dividindo os Dados em Conjuntos de Treinamento e Teste:

• 'data.iloc[:-50]' seleciona todas as linhas, exceto as últimas 50, como dados de treinamento.
• 'data.iloc[-50:]' seleciona as últimas 50 linhas como dados de teste.
• Essa divisão é comumente usada em dados de séries temporais, onde o modelo é treinado com dados históricos e testado com os dados mais recentes para avaliar o desempenho nas previsões futuras.

4. Especificando as Variáveis Preditivas:

• A lista 'predictors' contém os nomes das colunas que representam os recursos usados pelo modelo para fazer previsões. Eles incluem ('"<CLOSE>", '"<TICKVOL>"', '"<OPEN>"', '"<HIGH>"', e '"<LOW>"').
  

O código prepara um classificador Random Forest para prever o comportamento futuro do mercado com base nos dados passados. O modelo é treinado usando características como preço de fechamento, volume de ticks, e outras. Após dividir os dados em conjuntos de treinamento e teste, o modelo é ajustado aos dados de treinamento, aprendendo com os padrões históricos para fazer previsões futuras.

Medir a precisão do modelo

from sklearn.metrics import precision_score

prcsn = model.predict(test[predictors])

Importamos a função 'precision-score' do módulo 'sklearn.metrics'. O score de precisão é uma métrica usada para avaliar modelos de classificação, particularmente útil quando as classes são desbalanceadas. Ela mede quantos dos resultados positivos previstos são realmente positivos. Alta precisão indica taxas baixas de falsos positivos.

prcsn = pd.Series(prcsn, index = test.index)

Em seguida, convertemos as previsões ('prcsn') em uma 'Series' do Pandas, mantendo o índice do conjunto de dados de teste.

precision_score(test["<TRGT>"], prcsn)

Obtivemos a precisão das previsões do modelo comparando os valores reais do alvo no conjunto de teste com os valores previstos.

Saída:

cmbnd = pd.concat([test["<TRGT>"], prcsn], axis = 1)

cmbnd.plot()

Combinamos os valores reais do alvo e os valores previstos pelo modelo em um único DataFrame para facilitar a análise.

Saída:

def predors(train, test, predictors, model):
    model.fit(train[predictors], train["<TRGT>"])
    prcsn = model.predict(test[predictors])
    prcsn = pd.Series(prcsn, index = test.index, name = "Predictions")
    cmbnd = pd.concat([test["<TRGT>"], prcsn], axis = 1)
    return cmbnd

Esta função recebe os conjuntos de dados de treinamento e teste, uma lista de variáveis preditoras e um modelo de aprendizado de máquina. A função treina o modelo nos dados de treinamento, faz previsões nos dados de teste e retorna um DataFrame que contém tanto os valores reais do alvo quanto os valores previstos lado a lado.

def backtestor(data, model, predictors, start = 2500, step = 250):
    all_predictions = []

    for i in range(start, data.shape[0], step):
        train = data.iloc[0:i].copy()
        test = data.iloc[i:(i + step)].copy()
        predictions = predors(train, test, predictors, model)
        all_predictions.append(predictions)
    return pd.concat(all_predictions)

Esta função realiza um back-test rolante em um conjunto de dados de séries temporais usando um modelo de aprendizado de máquina. O back-test avalia o desempenho do modelo simulando previsões como se fossem feitas em um ambiente real de trading, onde os dados são revelados gradualmente ao longo do tempo.

predictions = backtestor(data, model, predictors)

Executa a função 'backtestor' usando o conjunto de dados especificado ('data'), o modelo de aprendizado de máquina ('model') e as variáveis preditoras ('predictors'). Realiza um back-test rolante, e as previsões resultantes são armazenadas na variável 'predictions'.

predictions["Predictions"].value_counts()

Contamos o número de ocorrências de cada valor único na coluna '"Predictions"' do DataFrame 'predictions'.

Saída:

precision_score(predictions["<TRGT>"], predictions["Predictions"])

Calcula a precisão da previsão do modelo. Precisão é uma métrica que mede a acurácia das previsões positivas.

Saída:

predictions["<TRGT>"].value_counts() / predictions.shape[0]

Calcula a proporção de cada valor único na coluna "<TRGT>" em relação ao número total de previsões.

Saída:

horizons = [2, 5, 55, 125, 750]
new_predictors = []

# Ensure only numeric columns are used for rolling calculations
numeric_columns = data.select_dtypes(include=[float, int]).columns

for i in horizons:
    # Calculate rolling averages for numeric columns only
    rolling_averages = data[numeric_columns].rolling(i).mean()
    
    # Generate the ratio column
    ratio_column = f"Close_Ratio_{i}"
    data[ratio_column] = data["<CLOSE>"] / rolling_averages["<CLOSE>"]
    
    # Generate the trend column
    trend_column = f"Trend_{i}"
    data[trend_column] = data["<TRGT>"].shift(1).rolling(i).sum()
    
    new_predictors += [ratio_column, trend_column]
data

Gera novos recursos baseados em médias móveis e tendências ao longo de diferentes horizontes temporais. Os preditores adicionais ajudam a melhorar o desempenho dos modelos, fornecendo mais informações sobre o mercado ao longo de períodos variados.

Saída:

data = data.dropna()

Removemos qualquer linha do DataFrame com valores ausentes.

def predict(train, test, predictors, model):
    model.fit(train[predictors], train["<TRGT>"])
    prcsn = model.predict_proba(test[predictors])[:1]
    prcsn[prcsn >= .6] = 1
    prcsn[prcsn < .6] = 0
    prcsn = pd.Series(prcsn, index = test.index, name = "Predictions")
    cmbnd = pd.concat([test["<TRGT>"], prcsn], axis = 1)
    return cmbnd

O modelo é treinado com o conjunto de dados de treinamento usando os preditores selecionados e a variável alvo. Para o limite e as previsões, aplicamos um limite personalizado de 0,6. Se a probabilidade para a classe 1 for 0,6 ou superior, o modelo prevê "1". Caso contrário, prevê "0". Esse ajuste permite que o modelo seja mais conservador, exigindo maior confiança antes de sinalizar uma operação.

predictions = backtestor(data, model, new_predictors)

predictions["Predictions"].value_counts()

precision_score(predictions["<TRGT>"], predictions["Predictions"])

Saída:

Nossa pontuação de precisão subiu ligeiramente, 0,52 se arredondarmos.

Treinando o Modelo e exportando para ONNX

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import onnx
import skl2onnx
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType

# Load and preprocess your data (example)
# Replace this with your actual data loading process
#data = pd.read_csv('your_data.csv')  # Replace with your actual data source
#data = data.dropna()

# Define predictors and target
predictors = ["<CLOSE>", "<TICKVOL>", "<OPEN>", "<HIGH>", "<LOW>"]
target = "<TRGT>"

# Split data into train and test sets
train, test = train_test_split(data, test_size=0.2, shuffle=False)

# Define and train the model
model = RandomForestClassifier(n_estimators=50, min_samples_split=50, random_state=1)
model.fit(train[predictors], train[target])

# Export the trained model to ONNX format
initial_type = [('float_input', FloatTensorType([None, len(predictors)]))]
onnx_model = convert_sklearn(model, initial_types=initial_type)

# Save the ONNX model to a file
with open("random_forest_model.onnx", "wb") as f:
    f.write(onnx_model.SerializeToString())

Treinamos nosso modelo, e ele é convertido, exportado e salvo no formato ONNX usando o skl2onnx. Depois, copiamos nosso modelo salvo para a pasta "Files" do MQL5, onde poderemos acessá-lo.

Colocando tudo junto no MQL5

Carregue o modelo no Oninit().

#include <Trade/Trade.mqh>
#define   ModelName          "RandomForestClassifier"
#define   ONNXFilename       "random_forest_model.onnx"

// Single ONNX model resource
#resource "\\Files\\random_forest_model.onnx" as const uchar ExtModelDouble[];

input double lotsize = 0.1;     // Trade lot size
input double stoploss = 20;     // Stop loss in points
input double takeprofit = 50;   // Take profit in points

// Trading functions
CTrade m_trade;

Os includes e variáveis globais no escopo global. Também especificamos que o modelo ONNX é embutido no MQL5 como um recurso binário. O #resource é usado para incluir arquivos externos.

//+------------------------------------------------------------------+
//| Run classification using double values                           |
//+------------------------------------------------------------------+
bool RunModel(long model, vector &input_vector, vector &output_vector)
{
    ulong batch_size = input_vector.Size() / 5; // Assuming 5 input features
    if (batch_size == 0)
        return (false);

    output_vector.Resize((int)batch_size);

    // Prepare input tensor
    double input_data[];
    ArrayResize(input_data, input_vector.Size());

    for (int k = 0; k < input_vector.Size(); k++)
        input_data[k] = input_vector[k];

    // Set input shape
    ulong input_shape[] = {batch_size, 5}; // 5 input features for each prediction
    OnnxSetInputShape(model, 0, input_shape);

    // Prepare output tensor
    double output_data[];
    ArrayResize(output_data, (int)batch_size);

    // Set output shape (binary classification)
    ulong output_shape[] = {batch_size, 2}; // Output shape for probability (0 or 1)
    OnnxSetOutputShape(model, 0, output_shape);

    // Run the model
    bool res = OnnxRun(model, ONNX_DEBUG_LOGS, input_data, output_data);

    if (res)
    {
        // Copy output to vector (only keeping the class with highest probability)
        for (int k = 0; k < batch_size; k++)
            output_vector[k] = (output_data[2 * k] < output_data[2 * k + 1]) ? 1.0 : 0.0;
    }

    return (res);
}

Esta função RunModel é um modelo ONNX que treinamos para realizar classificações binárias. A função determina a classe prevista (0 ou 1) com base em qual classe tem a maior probabilidade e armazena os resultados em um vetor de saída.

//+------------------------------------------------------------------+
//| Generate input data for prediction                               |
//+------------------------------------------------------------------+
vector input_data()
{
    vector input_vector;
    MqlRates rates[];

    // Get the last 5 bars of data
    if (CopyRates(Symbol(), PERIOD_H1, 5, 1, rates) > 0)
    {
        input_vector.Resize(5 * 5); // 5 input features for each bar

        for (int i = 0; i < 5; i++)
        {
            input_vector[i * 5] = rates[i].open;
            input_vector[i * 5 + 1] = rates[i].high;
            input_vector[i * 5 + 2] = rates[i].low;
            input_vector[i * 5 + 3] = rates[i].close;
            input_vector[i * 5 + 4] = rates[i].tick_volume;
        }
    }

    return (input_vector);
}

//+------------------------------------------------------------------+
//| Check if there is a new bar                                      |
//+------------------------------------------------------------------+
bool NewBar()
{
    static datetime last_time = 0;
    datetime current_time = iTime(Symbol(), Period(), 0);

    if (current_time != last_time)
    {
        last_time = current_time;
        return (true);
    }
    return (false);
}

//+------------------------------------------------------------------+
//| Check if a position of a certain type exists                     |
//+------------------------------------------------------------------+
bool PosExists(int type)
{
    for (int i = PositionsTotal() - 1; i >= 0; i--)
    {
        if (PositionGetInteger(POSITION_TYPE) == type && PositionGetString(POSITION_SYMBOL) == Symbol())
            return (true);
    }
    return (false);
}

//+------------------------------------------------------------------+
//| Script program initialization                                    |
//+------------------------------------------------------------------+
int OnInit()
{
    Print("Initializing ONNX model...");

    // Initialize the ONNX model
    long model = OnnxCreateFromBuffer(ExtModelDouble, ONNX_DEFAULT);
    if (model == INVALID_HANDLE)
    {
        Print("Error loading ONNX model: ", GetLastError());
        return INIT_FAILED;
    }

    // Store the model handle for further use
    GlobalVariableSet("model_handle", model);
    return (INIT_SUCCEEDED);
}

//+------------------------------------------------------------------+
//| Expert tick function                                             |
//+------------------------------------------------------------------+
void OnTick()
{
    if (NewBar()) // Trade at the opening of a new candle
    {
        vector input_vector = input_data();
        vector output_vector;

        // Retrieve the model handle
        long model = GlobalVariableGet("model_handle");
        if (model == INVALID_HANDLE)
        {
            Print("Invalid model handle.");
            return;
        }

        bool prediction_success = RunModel(model, input_vector, output_vector);
        if (!prediction_success || output_vector.Size() == 0)
        {
            Print("Prediction failed.");
            return;
        }

        long signal = output_vector[0]; // The predicted class (0 or 1)

        MqlTick ticks;
        if (!SymbolInfoTick(Symbol(), ticks))
            return;

        if (signal == 1) // Bullish signal
        {
            if (!PosExists(POSITION_TYPE_BUY)) // No buy positions exist
            {
                if (!m_trade.Buy(lotsize, Symbol(), ticks.ask, ticks.bid - stoploss * Point(), ticks.ask + takeprofit * Point())) // Open a buy trade
                    Print("Failed to open a buy position, error = ", GetLastError());
            }
        }
        else if (signal == 0) // Bearish signal
        {
            if (!PosExists(POSITION_TYPE_SELL)) // No sell positions exist
            {
                if (!m_trade.Sell(lotsize, Symbol(), ticks.bid, ticks.ask + stoploss * Point(), ticks.bid - takeprofit * Point())) // Open a sell trade
                    Print("Failed to open a sell position, error = ", GetLastError());
            }
        }
    }
}

//+------------------------------------------------------------------+
//| Script program deinitialization                                  |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
    // Release the ONNX model
    long model = GlobalVariableGet("model_handle");
    if (model != INVALID_HANDLE)
    {
        OnnxRelease(model);
    }
}

Durante a inicialização, o modelo é carregado e seu identificador é armazenado para uso posterior. Na função 'OnTick()', o script executa o modelo quando uma nova barra é detectada. Usamos a previsão do modelo (0 para tendência de baixa e 1 para tendência de alta) e executamos as operações de acordo. Operações de compra são feitas quando o modelo prevê uma tendência de alta, enquanto operações de venda são feitas para uma tendência de baixa.

Conclusão

Em resumo, usamos pacotes de processamento de dados (Jupyter Lab) para processar dados históricos, desenvolvemos e treinamos o modelo utilizando aprendizado de máquina para poder fazer previsões. Exploramos etapas cruciais necessárias para uma integração e operação sem falhas. Depois, focamos em carregar e manipular o modelo dentro do MQL5, embutindo-o como um recurso e garantindo que o modelo esteja devidamente inicializado e disponível durante a execução.

Concluindo, integramos um modelo ONNX em um ambiente de trading MQL5 para aprimorar a tomada de decisões usando aprendizado de máquina. O processo começou com o carregamento do modelo no ambiente MQL5. Em seguida, configuramos o Expert Advisor para coletar dados relevantes do mercado, pré-processá-los em vetores de características e alimentá-los no modelo para previsões. A lógica foi projetada para executar operações com base na saída do modelo. As posições só são abertas quando uma nova barra é detectada e nenhuma operação conflitante está ativa. Além disso, o sistema lida com verificações de posição, gerenciamento de erros e desalocação de recursos para garantir uma solução robusta e eficiente de trading. Essa implementação demonstra uma fusão perfeita entre análise financeira e insights baseados em IA, permitindo estratégias de trading automatizadas que se adaptam às condições do mercado em tempo real.

Referências

Criar um modelo