使用Python和MQL5开发机器人（第一部分）：数据预处理
986
0
概述
市场正变得越来越复杂。如今，它正演变成一场算法之战。超过95%的交易额是由交易机器人产生的。
下面一步是机器学习。这些虽然不是高级人工智能，但也并非简单的线性算法。机器学习模型能够在困难的背景下实现盈利。将机器学习应用于创建交易系统是一件有趣的事情。得益于神经网络，交易机器人将分析大数据，发现规律并预测价格走势。
我们将探讨交易机器人的开发周期：包括数据采集、处理、样本扩展、特征工程、模型选择和训练，通过Python创建交易系统以及监控交易。
使用Python工作有其自身的优势：在机器学习领域速度快，以及能够选择和生成特征。将模型导出到ONNX需要完全相同的特征生成逻辑，就像在Python中一样，这并不容易。这就是我选择通过Python进行在线交易的原因。
设定任务和选择工具
项目的目标是创建一个可持续盈利的Python交易机器学习模型。工作流程：
- 从MetaTrader 5收集数据，加载主要特征。
- 进行数据增强以扩展样本。
- 为数据打标签。
- 特征工程：生成、聚类、选择。
- 选择和训练机器学习模型。如果有可能的话，进行模型集成。
- 根据指标评估模型。
- 开发测试器以评估盈利能力。
- 基于新数据获得正向结果。
- 通过Python MQL5实现交易算法。
- 集成MetaTrader 5
- 优化模型。
工具：Python MQL5，用于提高速度和功能的Python机器学习库。
至此，我们已经明确了目标和宗旨。在本文的框架内，我们将开展以下工作：
- 从MetaTrader 5收集数据，加载主要特征。
- 进行数据增强以扩展样本。
- 为数据打标签。
- 特征工程：生成、聚类、选择。
设置环境和导入模块。数据收集
首先，我们需要通过MetaTrader 5获取历史数据。编写代码通过向初始化方法传递终端文件的路径来建立与交易平台的连接。
在循环中，使用mt5.copy_rates_range()方法获取数据，该方法具有以下参数：交易品类、时间框、开始和结束日期。这里包含了一个统计失败次数的计数器，以及为了稳定连接而设置的延迟。
通过调用mt5.shutdown()方法断开与平台的连接来终止该函数。
这是一个独立的函数，在后续程序中可以调用。
要运行脚本，您需要安装以下库：
-
NumPy：用于处理多维数组和数学函数的库。
-
Pandas：一种数据分析工具，为处理表格和时间序列提供了便捷的数据结构。
-
Random（随机模块）：用于生成随机数并从序列中选择随机元素的模块。
-
Datetime（日期和时间模块）：提供用于处理日期和时间的类和函数。
-
MetaTrader5：与MetaTrader 5交易平台进行交互的库。
-
Time（时间模块）：用于处理时间和程序执行延迟的模块。
-
Concurrent.futures：用于运行并行和异步任务的工具。我们之后在并行多货币交易中会用到它。
-
Tqdm：一个库，用于在执行迭代操作时显示进度指示器。我们之后在并行多货币交易中会用到它。
-
Train_test_split：一个函数，用于在训练机器学习模型时将数据集分割为训练集和测试集。
可以使用pip通过运行以下命令来安装它们：
pip install numpy pandas MetaTrader5 concurrent-futures tqdm sklearn matplotlib imblearn
import numpy as np import pandas as pd import random from datetime import datetime import MetaTrader5 as mt5 import time import concurrent.futures from tqdm import tqdm from sklearn.model_selection import train_test_split import matplotlib.pyplot as plt from sklearn.utils import class_weight from imblearn.under_sampling import RandomUnderSampler # GLOBALS MARKUP = 0.00001 BACKWARD = datetime(2000, 1, 1) FORWARD = datetime(2010, 1, 1) EXAMWARD = datetime(2024, 1, 1) MAX_OPEN_TRADES = 3 symbol = "EURUSD" def retrieve_data(symbol, retries_limit=300): terminal_path = "C:/Program Files/MetaTrader 5/Arima/terminal64.exe" attempt = 0 raw_data = None while attempt < retries_limit: if not mt5.initialize(path=terminal_path): print("MetaTrader initialization failed") return None instrument_count = mt5.symbols_total() if instrument_count > 0: print(f"Number of instruments in the terminal: {instrument_count}") else: print("No instruments in the terminal") rates = mt5.copy_rates_range(symbol, mt5.TIMEFRAME_H1, BACKWARD, EXAMWARD) mt5.shutdown() if rates is None or len(rates) == 0: print(f"Data for {symbol} not available (attempt {attempt+1})") attempt += 1 time.sleep(1) else: raw_data = pd.DataFrame(rates[:-1], columns=['time', 'open', 'high', 'low', 'close', 'tick_volume']) raw_data['time'] = pd.to_datetime(raw_data['time'], unit='s') raw_data.set_index('time', inplace=True) break if raw_data is None: print(f"Data retrieval failed after {retries_limit} attempts") return None # Add simple features raw_data['raw_SMA_10'] = raw_data['close'].rolling(window=10).mean() raw_data['raw_SMA_20'] = raw_data['close'].rolling(window=20).mean() raw_data['Price_Change'] = raw_data['close'].pct_change() * 100 # Additional features raw_data['raw_Std_Dev_Close'] = raw_data['close'].rolling(window=20).std() raw_data['raw_Volume_Change'] = raw_data['tick_volume'].pct_change() * 100 raw_data['raw_Prev_Day_Price_Change'] = raw_data['close'] - raw_data['close'].shift(1) raw_data['raw_Prev_Week_Price_Change'] = raw_data['close'] - raw_data['close'].shift(7) raw_data['raw_Prev_Month_Price_Change'] = raw_data['close'] - raw_data['close'].shift(30) raw_data['Consecutive_Positive_Changes'] = (raw_data['Price_Change'] > 0).astype(int).groupby((raw_data['Price_Change'] > 0).astype(int).diff().ne(0).cumsum()).cumsum() raw_data['Consecutive_Negative_Changes'] = (raw_data['Price_Change'] < 0).astype(int).groupby((raw_data['Price_Change'] < 0).astype(int).diff().ne(0).cumsum()).cumsum() raw_data['Price_Density'] = raw_data['close'].rolling(window=10).apply(lambda x: len(set(x))) raw_data['Fractal_Analysis'] = raw_data['close'].rolling(window=10).apply(lambda x: 1 if x.idxmax() else (-1 if x.idxmin() else 0)) raw_data['Price_Volume_Ratio'] = raw_data['close'] / raw_data['tick_volume'] raw_data['Median_Close_7'] = raw_data['close'].rolling(window=7).median() raw_data['Median_Close_30'] = raw_data['close'].rolling(window=30).median() raw_data['Price_Volatility'] = raw_data['close'].rolling(window=20).std() / raw_data['close'].rolling(window=20).mean() print("\nOriginal columns:") print(raw_data[['close', 'high', 'low', 'open', 'tick_volume']].tail(100)) print("\nList of features:") print(raw_data.columns.tolist()) print("\nLast 100 features:") print(raw_data.tail(100)) # Replace NaN values with the mean raw_data.fillna(raw_data.mean(), inplace=True) return raw_data retrieve_data(symbol)
定义全局变量：点差成本和经纪商佣金、培训和测试样本的日期、最大交易次数、交易品类。
在循环中加载MetaTrader 5的报价。数据被转换为DataFrame，并优化了以下功能：
- 周期为10和20的简单移动平均线（SMA）
- 价格变化
- 标准价格偏差
- 交易量变化
- 日/月价格变化
- 价格中位数
这些特征有助于衡量影响价格的因素。
显示有关DataFrame列和最新条目的信息。I
让我们看看我们的第一个函数是如何执行的：
一切都运行正常。通过代码成功加载了报价数据，准备并加载好特征。接下来，我们进入代码的下一部分。
进行数据增强以扩展样本量。
数据增强是基于现有数据生成新的训练样本，以增加样本量并提高模型质量。这对于有限数据的时间序列预测尤为重要。此外，它还能减少模型误差，提高模型的鲁棒性。
财经数据增强方法：
- 添加噪声（随机偏差）以提高抗噪性
- 时间平移以模拟不同的开发场景
- 根据模型价格的波动进行调整
- 源数据反演
当前已经实现了一个输入增强函数，该函数包括DataFrame和每种方法对应的新样本数量。它使用不同的方法生成新数据，并将其与原始DataFrame连接起来。
def augment_data(raw_data, noise_level=0.01, time_shift=1, scale_range=(0.9, 1.1)): print(f"Number of rows before augmentation: {len(raw_data)}") # Copy raw_data into augmented_data augmented_data = raw_data.copy() # Add noise noisy_data = raw_data.copy() noisy_data += np.random.normal(0, noise_level, noisy_data.shape) # Replace NaN values with the mean noisy_data.fillna(noisy_data.mean(), inplace=True) augmented_data = pd.concat([augmented_data, noisy_data]) print(f"Added {len(noisy_data)} rows after adding noise") # Time shift shifted_data = raw_data.copy() shifted_data.index += pd.DateOffset(hours=time_shift) # Replace NaN values with the mean shifted_data.fillna(shifted_data.mean(), inplace=True) augmented_data = pd.concat([augmented_data, shifted_data]) print(f"Added {len(shifted_data)} rows after time shift") # Scaling scale = np.random.uniform(scale_range[0], scale_range[1]) scaled_data = raw_data.copy() scaled_data *= scale # Replace NaN values with the mean scaled_data.fillna(scaled_data.mean(), inplace=True) augmented_data = pd.concat([augmented_data, scaled_data]) print(f"Added {len(scaled_data)} rows after scaling") # Inversion inverted_data = raw_data.copy() inverted_data *= -1 # Replace NaN values with the mean inverted_data.fillna(inverted_data.mean(), inplace=True) augmented_data = pd.concat([augmented_data, inverted_data]) print(f"Added {len(inverted_data)} rows after inversion") print(f"Number of rows after augmentation: {len(augmented_data)}") # Print dates by years print("Print dates by years:") for year, group in augmented_data.groupby(augmented_data.index.year): print(f"Year {year}: {group.index}") return augmented_data
调用代码并获得以下结果：
在应用数据增强方法后，原始的150,000个H1价格条数据被显著扩展成为可观的747,000个数据串。有关机器学习领域的许多权威研究表明，通过生成合成样本从而大幅增加训练数据的数量，对训练模型的质量指标具有积极影响。预计在我们的案例中，这种技术也将产生预期的效果。
标签数据
数据标注对于监测学习算法的成功至关重要。为源数据提供标签，再让模型学习这些标签。标注数据可以提高准确性、改善归纳能力、加快训练速度，并使模型评估变得更加容易。在EURUSD预测问题中，我们添加了一个名为“labels”的二进制列，该列指示下一个价格变动是否大于点差和佣金。它允许模型学习点差回吐和非回吐的趋势。
标注在机器学习算法发现数据中的复杂模式方面发挥着关键作用，这些模式在原始数据中是不可见的。接下来，让我们进行代码审核。
def markup_data(data, target_column, label_column, markup_ratio=0.00002): data.loc[:, label_column] = np.where(data.loc[:, target_column].shift(-1) > data.loc[:, target_column] + markup_ratio, 1, 0) data.loc[data[label_column].isna(), label_column] = 0 print(f"Number of markups on price change greater than markup: {data[label_column].sum()}") return data
执行代码并获取数据中标签的数量。该函数返回一帧。一切都运行正常。顺便说一句，在70多万个数据点中，只有7万个数据点的价格变化超过了点差。
现在我们用另一个数据标记功能。这次，它更接近于实际收益。
目标标签功能
def label_data(data, symbol, min=2, max=48): terminal_path = "C:/Program Files/MetaTrader 5/Arima/terminal64.exe" if not mt5.initialize(path=terminal_path): print("Error connecting to MetaTrader 5 terminal") return symbol_info = mt5.symbol_info(symbol) stop_level = 100 * symbol_info.point take_level = 800 * symbol_info.point labels = [] for i in range(data.shape[0] - max): rand = random.randint(min, max) curr_pr = data['close'].iloc[i] future_pr = data['close'].iloc[i + rand] min_pr = data['low'].iloc[i:i + rand].min() max_pr = data['high'].iloc[i:i + rand].max() price_change = 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) # Growth elif price_change > take_level and future_pr < curr_pr and max_pr < curr_pr + stop_level: labels.append(0) # Fall else: labels.append(None) 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) return data
该功能用于训练基于交易利润的机器学习模型以获取目标标签。该功能连接到MetaTrader 5，检索交易品类信息并计算止损/止盈水平。然后，为每个入场点随机确定一个未来价格时间点。如果价格变动超过了止盈点且未触及止损点，同时满足上涨/下跌的条件，则相应地添加1.0/0.0标识。如果不满足上述条件，则标记为None。接下来，会创建一个仅包含已标记数据的新数据框。之后，None值会被平均值替换。
最后，显示上涨/下跌标签的数量。
一切都按照预期进行。我们已经接收到了模型数据及其衍生数据。数据得到了增强和补充，并且被标记为两种不同的功能。让我们开始进行下一步——分类平衡。
顺便说一下，我相信有经验的机器学习读者已经明白，我们最终将开发一个分类模型，而不是回归模型。其实我更喜欢回归模型，因为在其中会看到比分类模型更多的预测逻辑。
所以，我们的下一步是进行分类平衡。
平衡类。分类
分类是基于人类根据共同特征对信息进行结构化的一种基本分析方法。分类的最初应用之一是勘探——根据地质特征寻找有前景的区域。
随着计算机的出现，分类达到了一个新的水平，但其本质依然不变——在看似混乱的细节中发现规律。对于金融市场来说，将价格动态分类为上涨/下跌是很重要的。然而，分类可能是不平衡的——趋势走向很少，而平稳偏多。
因此，使用了分类平衡方法：移除占主导地位的样本或生成稀有的样本。这使得模型能够可靠地识别出重要但罕见的价格动态规律。
pip install imblearn
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) 至此，我们的分类达到了平衡。每种分类（价格下降和下跌）的标签数量都是相等的。
接下来，让我们来谈谈数据预测中最重要的一点——特征。
机器学习中的特征是什么？
属性和特征是我们非常熟悉的基本概念。当我们描述一个对象时，我们会列出它的属性——这些就是特征。这样一组特征帮助我们完整描述了一个对象。
在数据分析中也是如此，每个观测值都由一组数值型和分类型特征来描述。选择具有信息量的特征对于成功至关重要。
在更高层次上，我们进行特征工程，即从初始参数中构建新的衍生特征，以便更好地描述研究对象。
因此，人类通过描述对象特征来认识世界的经验，在模式化和自动化的层面上被应用于科学中。
自动化特征提取n
特征工程是将源数据转换为用于训练机器学习模型的一组特征的过程。目标是找到最具信息量的特征。包含两种方法：手动方法（人工选择特征）和自动方法（使用算法）。
我们会使用自动方法。我们应用特征生成方法，自动从我们的数据中提取最佳特征。然后从生成的特征结果集中选择最具信息量的特征。
生成新特征的方法如下：
def generate_new_features(data, num_features=200, random_seed=1): random.seed(random_seed) new_features = {} for _ in range(num_features): feature_name = f'feature_{len(new_features)}' col1_idx, col2_idx = random.sample(range(len(data.columns)), 2) col1, col2 = data.columns[col1_idx], data.columns[col2_idx] operation = random.choice(['add', 'subtract', 'multiply', 'divide', 'shift', 'rolling_mean', 'rolling_std', 'rolling_max', 'rolling_min', 'rolling_sum']) if operation == 'add': new_features[feature_name] = data[col1] + data[col2] elif operation == 'subtract': new_features[feature_name] = data[col1] - data[col2] elif operation == 'multiply': new_features[feature_name] = data[col1] * data[col2] elif operation == 'divide': new_features[feature_name] = data[col1] / data[col2] elif operation == 'shift': shift = random.randint(1, 10) new_features[feature_name] = data[col1].shift(shift) elif operation == 'rolling_mean': window = random.randint(2, 20) new_features[feature_name] = data[col1].rolling(window).mean() elif operation == 'rolling_std': window = random.randint(2, 20) new_features[feature_name] = data[col1].rolling(window).std() elif operation == 'rolling_max': window = random.randint(2, 20) new_features[feature_name] = data[col1].rolling(window).max() elif operation == 'rolling_min': window = random.randint(2, 20) new_features[feature_name] = data[col1].rolling(window).min() elif operation == 'rolling_sum': window = random.randint(2, 20) new_features[feature_name] = data[col1].rolling(window).sum() new_data = pd.concat([data, pd.DataFrame(new_features)], axis=1) print("\nGenerated features:") print(new_data[list(new_features.keys())].tail(100)) return data
请注意以下参数：
random_seed=42 random_seed参数对于生成新特征结果的可复用性是必要的。The generate_new_features函数从源数据中创建新特征。I输入：数据、特征数量、种子值。
使用给定的种子初始化随机数生成器。在循环中：生成一个名称，随机选择两个现有特征和一个操作（加法、减法等）。计算所选操作对应的新特征。
生成后的新特征会被添加到原始数据中。返回更新后的数据，其中包含了自动生成的特征。
这段代码允许我们自动创建新特征，以提高机器学习的质量。
让我们运行代码并查看结果：
已经生成了100个新特征。接下来，我们进入下一阶段——特征聚类。
特征聚类
特征聚类将相似的特征分组，以减少特征的数量。这有助于去除冗余数据、降低相关性，并在不过度拟合的情况下简化模型。
常用的算法包括：k-means（指定聚类数量，特征围绕质心分组）和层次聚类（多层次树结构）。
特征聚类可以帮助我们筛选出大量无用的特征，以此提高模型的效率。
让我们看一下特征聚类的代码：
from sklearn.mixture import GaussianMixture
def cluster_features_by_gmm(data, n_components=4):
X = data.drop(['label', 'labels'], axis=1)
X = X.replace([np.inf, -np.inf], np.nan)
X = X.fillna(X.median())
gmm = GaussianMixture(n_components=n_components, random_state=1)
gmm.fit(X)
data['cluster'] = gmm.predict(X)
print("\nFeature clusters:")
print(data[['cluster']].tail(100))
return data 我们使用高斯混合模型（GMM）算法进行特征聚类。其基本思想是：数据是由多个正态分布的混合生成的，其中每个分布代表一个聚类。
首先，我们设置聚类的数量。然后，我们设置模型的初始参数：均值、协方差矩阵和聚类概率。算法使用最大似然法循环重新计算这些参数，直到它们不再变化为止。
最后，我们获得每个聚类的最终参数，通过这些参数我们可以确定新特征属于哪个聚类。.
GMM是一个非常强大的工具。它被应用于不同的任务之中。对于聚类形状复杂且边界模糊的数据，GMM非常有效。
这段代码使用GMM将特征划分为聚类。首先，我们获取原始数据并移除类别标签。然后，使用GMM将数据划分为给定数量的聚类。将聚类编号作为新列添加到数据中。最后，打印出获得的聚类表格。
让我们运行聚类代码并查看结果：
接下来，我们选择最佳特征的功能部分。
选择最佳特征
from sklearn.feature_selection import RFECV from sklearn.ensemble import RandomForestClassifier import pandas as pd def feature_engineering(data, n_features_to_select=10): # Remove the 'label' column as it is not a feature X = data.drop(['label', 'labels'], axis=1) y = data['labels'] # Create a RandomForestClassifier model clf = RandomForestClassifier(n_estimators=100, random_state=1) # Use RFECV to select n_features_to_select best features 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) # Return a DataFrame with the best features, 'label' column, and 'labels' column selected_features = X.columns[rfecv.get_support(indices=True)] selected_data = data[selected_features.tolist() + ['label', 'labels']] # Convert selected_features to a list # Print the table of best features print("\nBest features:") print(pd.DataFrame({'Feature': selected_features})) return selected_data labeled_data_engineered = feature_engineering(labeled_data_clustered, n_features_to_select=10)
这个函数从我们的数据中提取出最佳、最有用的特征。输入包含类别特征和标签的原始数据，以及需要选择的特征数量。
首先，重置类别标签，因为它们对特征选择没有帮助。然后，启动随机森林算法——该算法在随机特征子集上构建一系列决策树。
训练完所有树后，随机森林会评估每个特征的重要性以及它对分类的影响程度。基于这些重要性得分，函数选择了顶部特征。
最后，将选定的特征添加到数据中，并返回类别标签。该函数打印一个包含选定特征的表格，并返回更新后的数据。
为什么会这么繁琐？这就是我们如何去掉那些只会拖慢进程的无用特征的方法。我们保留了最佳特征，这样模型在训练这类数据时能呈现出更好的结果。
让我们运行这个函数，看一看结果吧：
事实证明，用于价格预测的最佳指标是开盘价本身。顶部特征包括基于移动平均线、价格增量、标准差、日价和月价变化等特征。自动生成的特征结果被证实是无信息量的。
该代码允许自动选择重要特征，这可以提高模型的性能和归纳能力。
结论
我们编写了数据操作代码，预测未来机器学习模型的发展。让我们简要回顾一下所采用的步骤：
- 首先，从MetaTrader 5平台提取欧元对美元（EURUSD）的初始数据。然后，通过随机操作（如添加噪声、位移和缩放）对数据进行变换，以显著增加样本量。
- 接下来，使用特殊的趋势标记（增长和下降）来标记价格值，同时考虑止损和止盈水平。为了平衡类别，去除了冗余的样本。
- 之后，执行复杂的特征工程。首先，从原始时间序列中自动生成了数百个新的衍生特征。然后，使用高斯混合聚类来检测冗余特征。最后，使用随机森林算法对变量进行排序和选择，以确定最具信息量的子集。
- 结果，我们生成了一个具有最优特征的高质量且丰富的数据集。该数据集将作为进一步训练机器学习模型和在Python中开发交易策略的基础，并将与MetaTrader 5进行集成。
在下一篇文章中，我们将重点关注选择最优分类模型，对其进行改进、实施交叉验证以及为Python/MQL5软件包编写测试函数。
本文由MetaQuotes Ltd译自俄文
原文地址： https://www.mql5.com/ru/articles/14350
使用 Python 的深度学习 GRU 模型到使用 EA 的 ONNX，以及 GRU 与 LSTM 模型的比较
