本文按工作流程逐环节讲解期魔方机器学习模型的开发方法：环境配置、数据获取、数据预处理与特征工程、模型训练、评估指标、可视化分析、模型保存与加载。完整可运行代码见案例：铁矿石预测螺纹钢。
环境配置
在期魔方的机器学习功能中，以下库已在系统内置，用户按需导入即可：
| 库 | 导入语句 | 用途 |
|---|---|---|
| pandas | import pandas as pd |
数据处理：读取、清洗、分析期货时间序列与价格数据 |
| numpy | import numpy as np |
科学计算：数学计算（np.sqrt）、数组操作、统计计算（np.mean） |
| matplotlib.pyplot | import matplotlib.pyplot as plt |
数据可视化：实际值 VS 预测值散点图、残差分布图、残差直方图等 |
| sklearn | from sklearn.linear_model import LinearRegression 等 |
机器学习建模；metrics 模块提供评估指标（R²、均方误差、平均绝对误差） |
| joblib | import joblib |
模型持久化：保存（joblib.dump）与加载（joblib.load）训练好的模型 |
| seaborn | import seaborn as sns |
基于 matplotlib 的统计数据可视化库，图表样式更美观 |
| os | import os |
操作系统功能：文件路径处理、模型保存路径管理 |
| scipy.stats | from scipy import stats |
统计分析：绘制 Q-Q 图（正态性检验） |
| qmf_model_sdk / qmf_data | import qmf_model_sdk / from qmf_data import load_kline |
期魔方模型模块与数据模块，用于调用期魔方的数据 |
| asyncio | import asyncio |
支持期魔方数据的异步调用 |
此外，matplotlib 需要以下图形、中文及负号显示设置：
matplotlib.use("QtAgg") # 期魔方平台图形显示必须要有这行代码
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False
这些库的组合使用，构成了一个完整的数据分析和机器学习工作流程：
- 数据获取和处理 (qmf_data, pandas)
- 数值计算和统计 (numpy, scipy)
- 机器学习建模 (sklearn)
- 模型评估和可视化 (matplotlib, seaborn)
- 模型保存和部署 (joblib)
数据获取
通过期魔方系统下载数据
qmf_model_sdk 是期魔方自带的库。使用方式：进入 策略模块，点击 机器学习子模块，在右上角找到 获取数据的按钮，点击该按钮会打开 VSCode 编辑器，可以看到编辑器中已经预先写好了获取数据的代码。直接修改代码中的参数，然后保存代码，执行程序，待日志打印获取数据结束，则完成下载数据。
import asyncio
import qmf_model_sdk
# 获取数据配置
symbol = "rb888" # 合约代码
begin_time = "2020-01-01" # 开始时间，格式 YYYY-mm-dd
end_time = "2025-05-01" # 结束时间，格式 YYYY-mm-dd
symbol_cycle = "1天" # 周期
asyncio.run(qmf_model_sdk.get_futures_data(symbol, begin_time, end_time, symbol_cycle))
symbol_cycle 可选值：'1分钟','3分钟','5分钟','10分钟','15分钟','30分钟','45分钟','1小时','2小时','4小时','1天','1周','1月','1季','1年'。
通过 qmf_data 读取数据
def get_futures_data(symbol, start_time="2023-01-01 00:00:00", end_time="2025-05-01 00:00:00"):
# 通过期魔方数据模块获取期货数据
data = load_kline(product=symbol, cycle="1D", start_time=start_time, end_time=end_time)
# 转换日期格式并筛选时间范围
data["date"] = pd.to_datetime(data["date"], errors="coerce")
# 重命名收盘价列
data.rename(columns={"close": "收盘"}, inplace=True)
# 重置索引
data.reset_index(inplace=True, drop=True)
return data
# 获取螺纹钢和铁矿石主力合约数据
rbm_data = get_futures_data("rb888") # 螺纹钢主连
im_data = get_futures_data("i888") # 铁矿石主连
数据预处理与特征工程
以铁矿石预测螺纹钢为例：首先合并数据，基于日期将铁矿石和螺纹钢数据对齐；然后创建前一天的铁矿石价格列，并删除第一行（因为它没有前一天的数据）；接下来确保数据按时间排序，并按时间顺序将数据集划分为训练集（80%）和测试集（20%）；最终将前一天铁矿石的收盘价作为特征（X），当天螺纹钢的收盘价作为预测目标（y），分别准备训练集和测试集的特征与目标变量。
要点：
- 使用前一天铁矿石收盘价作为特征预测当天螺纹钢收盘价
- 按时间顺序将数据集分为训练集（80%）和测试集（20%），不打乱时序
- 通过内连接（inner join）确保两个品种的数据完全对齐
# 合并数据，基于日期，这样可以确保铁矿石和螺纹钢数据对齐
df = pd.merge(
rbm_data[["date", "收盘"]],
im_data[["date", "收盘"]],
on="date",
how="inner",
suffixes=("_螺纹钢", "_铁矿石"),
)
# 创建前一天的铁矿石价格列
df["前一天铁矿石收盘"] = df["收盘_铁矿石"].shift(1)
# 删除第一行，因为它没有前一天的数据
df = df.dropna()
# 建立预测模型，利用前一天铁矿石的价格预测当天螺纹钢的收盘价
X = df[["前一天铁矿石收盘"]].values
y = df["收盘_螺纹钢"].values
# 确保数据按时间排序
df = df.sort_values("date")
# 按时间顺序划分训练集和测试集（80%训练，20%测试）
train_size = int(len(df) * 0.8)
train_data = df.iloc[:train_size]
test_data = df.iloc[train_size:]
# 准备特征和目标变量
X_train = train_data[["前一天铁矿石收盘"]].values
y_train = train_data["收盘_螺纹钢"].values
X_test = test_data[["前一天铁矿石收盘"]].values
y_test = test_data["收盘_螺纹钢"].values
模型训练
- 使用 sklearn 中的 LinearRegression 类构建线性回归模型
- 用铁矿石价格（X_train）作为特征，训练模型预测螺纹钢价格（y_train）
- 训练完成后，分别在训练集和测试集上进行预测，得到预测值
# 创建并训练线性回归模型
model = LinearRegression()
model.fit(X_train, y_train)
# 在训练集和测试集上进行预测
y_train_pred = model.predict(X_train)
y_test_pred = model.predict(X_test)
评估指标
- R方分数(R²)：衡量模型对数据变异性的解释程度，范围0-1，越接近1表示模型拟合效果越好
- 均方根误差(RMSE)：反映预测值与实际值的平均偏差大小，单位与原始数据相同
- 平均绝对误差(MAE)：表示预测值与实际值的平均绝对差距
- 平均绝对百分比误差(MAPE)：以百分比形式表示预测误差的相对大小
def calculate_metrics(y_true, y_pred):
# R方分数：反映模型解释数据变异性的程度，越接近1越好
r2 = r2_score(y_true, y_pred)
# 均方根误差(RMSE)：反映预测值与真实值的偏差，越小越好
mse = mean_squared_error(y_true, y_pred)
rmse = np.sqrt(mse)
# 平均绝对误差(MAE)：反映预测值与真实值的平均差距，越小越好
mae = mean_absolute_error(y_true, y_pred)
# 平均绝对百分比误差(MAPE)：反映预测误差的相对大小，越小越好
mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100
return r2, rmse, mae, mape
# 分别计算训练集和测试集的评估指标
train_metrics = calculate_metrics(y_train, y_train_pred)
test_metrics = calculate_metrics(y_test, y_test_pred)
# 打印评估结果
print("\n模型评估结果：")
print("训练集：")
print(f"R²分数: {train_metrics[0]:.4f}")
print(f"RMSE: {train_metrics[1]:.2f}")
print(f"MAE: {train_metrics[2]:.2f}")
print(f"MAPE: {train_metrics[3]:.2f}%")
print("\n测试集：")
print(f"R²分数: {test_metrics[0]:.4f}")
print(f"RMSE: {test_metrics[1]:.2f}")
print(f"MAE: {test_metrics[2]:.2f}")
print(f"MAPE: {test_metrics[3]:.2f}%")
可视化分析
通过图表全面展示模型表现：
- 实际值vs预测值散点图：直观展示预测准确度，点越接近对角线说明预测越准确
- 残差分布图：展示预测误差随预测值的变化情况，用于检测是否存在系统性偏差
- 残差直方图：显示预测误差的分布特征，理想情况应呈正态分布
- Q-Q图：用于检验残差是否服从正态分布，点越接近直线表示越符合正态分布
- 时间序列预测图：展示螺纹钢价格随时间的实际变化与模型预测值的对比，并标明训练集和测试集的分界点
- 铁矿石与螺纹钢价格关系散点图：展示前一天铁矿石收盘价与当天螺纹钢收盘价的相关性，包含拟合的回归线
# 创建评估图
plt.figure(figsize=(20, 15))
# 实际值vs预测值图
plt.subplot(2, 2, 1)
plt.scatter(y_train, y_train_pred, c="blue", alpha=0.5, label="训练集")
plt.scatter(y_test, y_test_pred, c="red", alpha=0.5, label="测试集")
plt.plot([y.min(), y.max()], [y.min(), y.max()], "k--", lw=2)
plt.xlabel("实际值")
plt.ylabel("预测值")
plt.title("实际值 vs 预测值")
plt.legend()
# 残差图
residuals_train = y_train - y_train_pred
residuals_test = y_test - y_test_pred
plt.subplot(2, 2, 2)
plt.scatter(y_train_pred, residuals_train, c="blue", alpha=0.5, label="训练集")
plt.scatter(y_test_pred, residuals_test, c="red", alpha=0.5, label="测试集")
plt.axhline(y=0, color="k", linestyle="--")
plt.xlabel("预测值")
plt.ylabel("残差")
plt.title("残差分布图")
plt.legend()
# 残差直方图
plt.subplot(2, 2, 3)
sns.histplot(residuals_train, color="blue", alpha=0.5, label="训练集", bins=30)
sns.histplot(residuals_test, color="red", alpha=0.5, label="测试集", bins=30)
plt.xlabel("残差")
plt.ylabel("频数")
plt.title("残差直方图")
plt.legend()
# Q-Q图
plt.subplot(2, 2, 4)
stats.probplot(residuals_train, dist="norm", plot=plt)
plt.title("残差Q-Q图")
plt.tight_layout()
plt.show()
# 添加时间序列预测图
plt.figure(figsize=(15, 7))
plt.plot(train_data["date"], train_data["收盘_螺纹钢"], 'b-', label='训练集实际值')
plt.plot(train_data["date"], y_train_pred, 'g--', label='训练集预测值')
plt.plot(test_data["date"], test_data["收盘_螺纹钢"], 'r-', label='测试集实际值')
plt.plot(test_data["date"], y_test_pred, 'y--', label='测试集预测值')
plt.axvline(x=train_data["date"].iloc[-1], color='k', linestyle='--', label='训练/测试分割点')
plt.xlabel('日期')
plt.ylabel('螺纹钢收盘价')
plt.title('时间序列预测结果')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# 添加散点图显示铁矿石价格与螺纹钢价格的关系
plt.figure(figsize=(10, 6))
plt.scatter(train_data["前一天铁矿石收盘"], train_data["收盘_螺纹钢"],
c="blue", alpha=0.5, label="训练集")
plt.scatter(test_data["前一天铁矿石收盘"], test_data["收盘_螺纹钢"],
c="red", alpha=0.5, label="测试集")
# 添加回归线
x_range = np.linspace(
min(df["前一天铁矿石收盘"]),
max(df["前一天铁矿石收盘"]),
100
).reshape(-1, 1)
plt.plot(x_range, model.predict(x_range), 'k-', lw=2)
plt.xlabel('前一天铁矿石收盘价')
plt.ylabel('螺纹钢收盘价')
plt.title('前一天铁矿石价格与螺纹钢价格关系')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
模型保存与加载
模型保存路径设置
此为期魔方平台的内置函数，使用默认设置即可！
def get_save_model_path(file_name):
# 获取当前文件的绝对路径
current_file_path = os.path.abspath(__file__)
# 获取不带扩展名的文件名
filename_without_ext = os.path.splitext(os.path.basename(current_file_path))[0]
# 返回完整的模型保存路径
return f"{os.path.dirname(current_file_path)}\\{filename_without_ext}__{file_name}"
保存模型
将训练好的模型保存到本地，以供后续模型的调试及在实盘策略中进行价格预测。
# 生成模型文件名（文件名需与实盘策略中 load_model 的名称一致）
model_filename = get_save_model_path("model_lr.joblib")
# 使用 joblib 保存模型
joblib.dump(model, model_filename)
print(f"模型已保存至: {model_filename}")
加载模型
此阶段为将训练好的模型加载进来，进一步进行价格预测，比如新传入一个铁矿石价格为 800 元，让模型来预测螺纹钢的价格。
# 加载保存的模型
loaded_model = joblib.load(model_filename)
# 使用加载的模型进行预测
test_value = np.array([[800]]) # 假设前一天铁矿石价格为 800 时
predicted_price = loaded_model.predict(test_value)[0]
print("使用加载的模型预测：")
print(f"当铁矿石前一日价格为 800 时，螺纹钢当日预测价格为: {predicted_price:.2f}")
相关阅读
- • 案例：铁矿石预测螺纹钢
- • 实盘策略集成
评论
登录后参与讨论，与站内账号体系共用。