如何使用ApacheDolphinScheduler来打造一个机器学习选股系统(组图)

Apache DolphinScheduler 在 DataOps 领域提供了强大的分布式可视化工作流调度能力。 2022年我们将为其添加机器学习任务调度能力，逐步支持主流MLops项目/服务商开箱即用的功能。

Apache DolphinScheduler 目前支持 MLOps 工具，包括 MLflow、DVC、Jupyter、OpenMLDB 和其他任务组件，让用户可以更轻松地以低成本编排机器学习系统。

本文介绍如何使用 Apache DolphinScheduler 搭建机器学习选股系统，每天自动更新选股模型，智能选股，并在交易时间内持续监控模型的选股效果。

p>

Apache DolphinScheduler

1

系统介绍概述

01

概览

系统显示

截至 2022 年 6 月 24 日，系统显示以下内容：

系统如何工作

选股逻辑

计算整个股票市场中满足5日均线高于10日均线信号的股票作为目标池，对上述股票池中的每只股票构造如下特征进行训练型号：

1.股价与布林带三个轨道的相对价值

2. 股价相对于多条移动平均线的价值；多个移动平均线的多个区间的斜率

3.当前K线的形状是通过talib模式计算出来的，比如是否是三只乌鸦，是否是十字星等，具体见K线模式识别()

注意：您还可以添加任何您认为有用的信息作为功能。

模型训练

根据第二天是否上涨，进行第二次分类。使用AutoML工具flaml进行5分钟的训练，每天构建120天的数据集，其中最后7天用于评估。

胜率和盈亏比的影响如下：

• 单纯用5日均线高于10日均线时：胜率0.46，盈亏比1.15，期望是关于-0.011;

• 5日均线高于10日均线并加入机器学习模型作为信号后，胜率0.58，盈亏比1.35，期望值约为0. 363。

以上是同期无模型策略与有模型策略的对比。预计会从-0.011提升到0.363。

备注：以上对比是2022-06-23模型训练（训练集220000，测试集1.70000）后的评价指标，其中模型每天选取置信度得分前10（可以认为是模型认为最有可能上涨的前1个0）个股票；

对于真实行情的表现，您可以在系统显示中看到2022-06-24的实时表现。

任务调度

系统中所有任务调度均使用Apache DolphinScheduler完成，包括数据处理、特征工程、模型训练、模型评估、模型上线、批量实时预测、效果监控等。

Apache DolphinScheduler 可以每晚自动更新模型并上线预测。当模型当天表现不佳时，您可以调整参数和特征，一键重新训练新模型并在线评估。

对任务也有容错机制，可以保证系统稳定运行。

前端展示

前端显示是使用 Observable 实现的。得益于其notebook丰富易用的可视化数据分析功能，构建了实时选股系统的效果监控。

技术栈

涉及的主要技术能力如下：

关于量化知识，由于用户级别不同，可以进行不同的选股逻辑和特征工程。

本系统旨在帮助用户建立一个符合自己认知的选股系统（当你有很多选股参考信息，不知道如何结合它们做出更好的决策时）。

项目中涉及的工作流任务的实现可以在这里找到 GitHub - Apache DolphinScheduler-MLOps-Stock-Analysis()

02

模块介绍

模块介绍章节涉及到整个系统各个模块的具体实现代码和细节。您也可以根据自己的阅读习惯跳到文章的最后一个总结章节，然后回到这里继续阅读。

后端模块介绍

下图是一个整体的工作流程run_system，每晚定时启动，执行数据更新、模型训练、模型评估、模型部署和推理，然后推荐每日库存池。

run_system 工作流程包含 4 个子工作流程：

• prepare_datas：每日数据下载、信号计算、特征（量化交易中的因子）计算

• training_model：生成训练数据、训练模型和评估模型

• 部署：部署模型

• batch_inference：生成要通过批处理算法预测的股票并进行预测。

准备数据

下图为数据准备工作流程prepare_datas，下载股票数据，进行信号计算和特征计算：

• download_data : 下载整个市场的每日股票数据

• calc_signals：进行信号计算（每天计算满足信号条件的股票，如5日均线和10日均线金叉的股票）

• calc_features：特征（量化交易中称为因子）计算（每天计算每只股票的特征值，如收盘价与5日均线的相对值，股票是否为十字星，等等）

下载数据

任务类型为Shell任务类型，用于下载股市每日数据。

#添加项目路径，即Git克隆项目Apache DolphinScheduler-MLOps-Stock-Analysis的地址，这样Python就可以直接在另一个目录运行该项目的脚本了

export PYTHONPATH=${project}

#激活Python环境

source ${project}/env/bin/activate

#设置下载股票数据的路径

data_path=${project}/data/daily

#将数据下载到指定路径

python -m dmsa.data.download ${data_path}

自定义参数中的设置是指启动工作流时传入的项目参数的值，可以替换shell中的${project}，比如传入/home/user/Apache DolphinScheduler-MLOps-Stock - 分析，第一行会变成export PYTHONPATH=/home/user/Apache DolphinScheduler-MLOps-Stock-Analysis。如果下面没有特别的功能，Apache DolphinScheduler 的功能就不介绍了。

计算信号

任务类型为Shell任务类型，所有信号根据上一个任务下载的数据计算。

#添加项目路径，即Git克隆项目Apache DolphinScheduler-MLOps-Stock-Analysis的地址，这样python就可以直接在另一个目录运行该项目的脚本了

export PYTHONPATH=${project}

#激活python环境

source ${project}/env/bin/activate

#设置下载股票数据的路径

data_path=${project}/data/daily

# 根据feature_signal.txt的配置计算信号。详情见项目中的实现

python -m dmsa.data_processing.calc_signals \
    --data_path ${data_path} \
    --name_file ${project}/feature_signal.txt

计算出来的信号会自动存储在Mysql中。 Mysql的配置请参考Apache DolphinScheduler-MLOps-Stock-Analysis()进行简单配置。

计算特征

任务类型为Shell任务类型，所有特征都是根据上一个任务下载的数据计算出来的。

#添加项目路径，即Git克隆项目Apache DolphinScheduler-MLOps-Stock-Analysis的地址，这样python就可以直接在另一个目录运行该项目的脚本了

export PYTHONPATH=${project}

#激活python环境

source ${project}/env/bin/activate

#设置下载股票数据的路径

data_path=${project}/data/daily

#根据feature_signal.txt的配置，计算特征，详见项目中的实现

python -m dmsa.data_processing.calc_features \
    --data_path ${data_path} \
    --name_file ${project}/feature_signal.txt

计算出来的特征会自动存储在Mysql中。 Mysql的配置请参考Apache DolphinScheduler-MLOps-Stock-Analysis()进行简单配置。

如果需要更强大的特征计算和存储能力，可以使用 Apache DolphinScheduler 中的 OpenMLDB 组件（可以修改（参考 Apache DolphinScheduler OpenMLDB Task: Creating an End-to-End MLOps Workflow）。

训练模型

下图为工作流training_model各任务的DAG图，主要执行以下任务：

• prepare_data ：准备模型训练数据和模型评估数据

• 训练：训练模型

• 部署：部署刚刚训练好的模型用户推理评估数据

• 评估：评估模型。由于评价指标为胜率和盈亏比，因此进行了额外的计算评价。

• close_service：评估完成后关闭刚刚部署的服务

准备数据

任务类型为Shell任务类型，为训练模型准备训练数据股票条件单数据库设计，为评估准备测试数据。

export PYTHONPATH=${project}
source ${project}/env/bin/activate
save_data_path=${project}/data/training

# 生成数据集

python -m dmsa.data_processing.build_datas \
    --task_type train \
    --config ${project}/feature_signal.txt \
    --save_path ${save_data_path} \
    --data_path ${project}/data/daily

#将生成的数据集save_data_path对应的路径赋值给变量data_path，通过自定义参数设置为OUT，传递给下游任务

echo "#{setValue(data_path=${save_data_path})}"

训练

任务类型为MLflow任务类型，由Apache DolphinScheduler提供，可用于输入数据集，自动训练模型。

上图显示，AutoML 用于基于传入的数据（csv 格式）进行 120 秒的训练。 AutoML 的工具使用 flaml。设置模型实验名称为baseline，模型将上传至MLflow服务中心，注册模型名称为baseline。

有关更多功能，请参阅 Apache DolphinScheduler MLflow Components()。

部署

任务类型为MLflow任务类型，由Apache DolphinScheduler提供，可以部署MLflow服务中心的模型。

上图以模型名称为baseline，版本号为Production的模型以Docker的形式部署，保留7070端口。

有关更多功能，请参阅 Apache DolphinScheduler MLflow Components()。

评估

任务类型为Shell任务类型，用户请求服务获取评估数据的结果并进行评估。

#表示每天选取得分最高的3只股票来测试效果

python -m dmsa.evaluate.calc_evaluate \
    --source_data_path ${data_path}/source.csv \
    --evaluate_data ${data_path}/test.csv \
    --api 'http://127.0.0.1:7070/invocations' \
    --top_n 3

关闭服务

任务类型为Shell任务类型，用于关闭刚刚启动的模型服务

# 刚才的模型服务会运行一个docker容器来提供服务，容器名称统一命名为ds-mlflow-{model_name}-{model_version}

# 这样就可以通过简单的一行命令关闭服务

docker rm -f ds-mlflow-baseline-Production

部署

和上面的Deployment差不多，多了一个7000端口用于推理服务。

batch_inference

Workflow batch_inference 用于推理需要在线运行的任务，包括两个任务：

• build_inference_data：为推理生成数据

• 推理：调用接口进行推理

build_inference_data

任务类型为Shell任务类型，用于生成需要推理的数据。

export PYTHONPATH=${project}
source ${project}/env/bin/activate
save_data_path=${project}/data/inference.csv

#用于生成推理数据

python -m dmsa.data_processing.build_datas \
    --task_type inference \
    --config ${project}/feature_signal.txt \
    --save_path $save_data_path
echo "#{setValue(data_path=${save_data_path})}"

推理

任务类型为Shell任务类型，根据上一个任务生成的推理数据进行推理，数据库存储前10名股票。

export PYTHONPATH=${project}
source ${project}/env/bin/activate

#推断股票条件单数据库设计，服务地址为：7000/invocations，7000为刚刚配置的端口，invocations为MLflow服务的默认路径

python -m dmsa.evaluate.inference \
    --evaluate_data ${data_path} \
    --api 'http://127.0.0.1:7000/invocations' \
    --top_n 10

监控

还有实时监控工作流监控，用于实时计算监控数据并存入数据库，实时监控选股效果。

当前包含两个任务，由 Apache DolphinScheduler 每 10 秒启动一次：

• 现货：持续更新整体市场数据，包括涨跌幅、增长率等。

• kline：从K线层面持续跟踪模型精选个股的表现

现货

定期监控全图的整体数据，包括获取当前时间点所有股票的涨跌情况、增长率等。

克莱恩

定期监测昨天所选股票在最近一个交易日的实时收益走势。

前端模块介绍

前端目前使用 Observable 进行展示，但可以使用 Juptyer 代替。

可观察

Observable 是一个基于 JavaScript 扩展来进行计算的笔记本。它使许多不懂代码的数据分析和可视化人员能够简单、低成本地使用强大的可视化工具。个人版免费。

目前主要显示三个图标来监控模型效果：

• 每日实时股票推荐

• 每日推荐股票的实时分发

• 整体市场图景实时上下分布

每日推荐股票的实时状态

通过加入每日推荐股票标记和市场概览标记显示，每5秒刷新一次。以下两段代码可以直接复制到 Observable 的 Notebook 中：

//获取数据

stockDatas = {
  let i = 0;
  while (true) {
    try {
      const data = await db.query("SELECT a.date, a.y_pred, a.score, b.* FROM candidate a LEFT JOIN spot b ON a.code = b.code WHERE a.y_pred=1;")
      yield Promises.delay(5000, data);
    } catch(e){
    console.log(e.name + ":" + e.message);
    }
  }
}

// 转换为表格

stocks = Inputs.table(stockDatas)

每日推荐股票的实时收益率曲线

用Observable自带的图来表示

Plot.plot({
  marks: [
    Plot.ruleY([0]),
    Plot.lineY(changesDatas, {x: "time", y: "changes"})
  ]
})

每日推荐股实时分布

使用 Observable 自带的条形图来表示：

Plot.plot({
  marks: [
    Plot.rectY(stockDatas, Plot.binX({y: "count"}, {x: "涨跌幅"})),
    Plot.ruleY([0])
  ],
})

全盘实时涨跌分布

使用 Observable 自带的条形图来表示：

// 从数据库中读取数据
spotDatas = db.query("SELECT * FROM spot;")
复制
// 画图
Plot.plot({
  marks: [
    Plot.rectY(spotDatas, Plot.binX({y: "count"}, {x: "涨跌幅"})),
    Plot.ruleY([0])
  ],
  x: {ticks: 20}
})

本文展示了使用 Apache DolphinScheduler 来调度机器学习系统中的各个模块。 Observable 是一个智能选股系统，具有前端展示和可视化能力。希望能给您带来以下好处：

了解如何使用 Apache DolphinScheduler 构建选股系统。了解 Apache DolphinScheduler 如何灵活且简单地编排 MLOps 场景中的每个模块。了解 Apache DolphinScheduler MLflow 组件如何以零成本为用户提供机器学习功能。了解强大的可视化产品 Observable 的功能及其与 Apache DolphinScheduler 的使用。