基于机器学习的实时股市预测服务器架构

在这个仓库中，我开发了完整的基于机器学习的实时股市预测服务器端主架构。我使用 TensorFlow.js 构建了机器学习模型架构，并利用 Kafka 实现实时数据流和管道处理。

使用的技术：

1. Kafka。
   
   • 将日志从数据源高效地传输到各个主题。
   • 消费者订阅这些主题，以实现实时机器学习预测与模型训练的并行操作。
2. TensorFlow.js。
   
   • 在 Node.js 环境中构建 TensorFlow 模型。
   • 利用时间序列的股市数据对模型进行训练。
   • 使用 TensorFlow.js 模型进行实时预测。
3. MongoDB。
   
   • 将传入的股市日志更新至数据库。
   • 利用存储的日志进行分析及模型训练。
   • 存储机器学习模型的性能指标，以便监控。
4. Node.js
   
   • 整个服务器架构均采用 Node.js 开发。

实现步骤

启动 Kafka 环境

首先启动 ZooKeeper 服务：

$ bin/zookeeper-server-start.sh config/zookeeper.properties

然后启动 Kafka Broker 服务：

$ bin/kafka-server-start.sh config/server.properties

开始日志流式传输

在此项目中，我使用 dataset 文件夹下的 CSV 文件作为数据源。数据通过 Kafka 主题进行管道化处理：第一个主题将日志传输至 MongoDB，而第二个主题则将日志发送至 TensorFlow 模型，用于实时预测。
通过生产者和消费者对数据源进行流式传输，该架构非常适合同时进行实时分析、机器学习模型训练以及模型预测。
生产者可以通过以下命令启动：

$ node producer.js
# 或
$ start.sh

生产者正在流式传输日志。

$ node consumer.js

消费者正在流式传输日志。

在消费者（consumer.js）中，接收到的日志会被更新到 MongoDB 数据库，以供后续的模型训练和分析使用。

机器学习模型

机器学习模型架构是使用 TensorFlow.js 开发的。模型使用 80% 的存储数据进行训练，并用剩余的 20% 数据进行验证。由于本问题属于时间序列问题，因此在训练前需要对数据进行预处理。数据预处理采用了 MinMax-Scalar 算法。
训练机器学习模型：

$ node tf_train.js
# 或
$ server.sh

模型验证： ```sh $ node tf_validate.js ```
验证完成后，模型的真实值、预测值以及对应的时间序列数据的日期和属性都会被更新到 MongoDB 数据库中。

从 MongoDB 获取的模型性能图表

实时预测

训练好的模型权重会被保存并在消费者端加载。消费者订阅 Kafka 流中的第二个主题，从而能够实时预测时间序列事件的结果。由于 Kafka 管道中的两个主题可以并行工作，且日志以实时方式流经 Kafka，因此机器学习模型能够在实时环境中完成训练和预测任务。

$ node ml_consumer.js

``` 预测 [属性] [预测值]。 例如 - 预测开盘价 0.12453 ``` 上图中显示的这行内容即为模型的实时预测结果。模型以过去 7 个时间序列日志作为输入，预测第 8 个时间序列事件。

未来展望

1. 目前已完成模型的服务器端架构，接下来还需要开发客户端部分。
2. 我曾尝试使用 React.js 构建客户端界面，但由于某些技术难点未能成功，因此暂时搁置了客户端的开发，留待未来完善。
3. 当前编写的代码还可以进一步优化，以提升架构设计和模型性能。

Real-time-stock-market-prediction 快速上手指南

本指南帮助开发者快速搭建基于 Kafka、TensorFlow.js 和 MongoDB 的实时股市预测服务端架构。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统: Linux 或 macOS (Windows 用户建议使用 WSL2)
• Node.js: 已安装 Node.js 运行环境 (推荐 LTS 版本)
• Apache Kafka: 已下载并配置 Kafka 安装包 (包含 ZooKeeper)
• MongoDB: 已安装并启动 MongoDB 服务
• 依赖安装: 在项目根目录下运行 npm install 安装所需节点模块

安装与启动步骤

1. 启动 Kafka 环境

首先需要在终端中依次启动 ZooKeeper 和 Kafka Broker 服务。请进入 Kafka 安装目录执行以下命令：

$ bin/zookeeper-server-start.sh config/zookeeper.properties

$ bin/kafka-server-start.sh config/server.properties

2. 启动数据流 (Streaming)

本项目使用 dataset 文件夹中的 .csv 文件作为数据源。生产者将数据推送至 Kafka Topic，消费者订阅 Topic 并将数据存入 MongoDB 或用于实时预测。

启动生产者 (Producer):

$ node producer.js
# 或
$ start.sh

启动基础消费者 (Consumer): 该消费者负责接收日志并更新到 MongoDB，用于后续模型训练和分析。

$ node consumer.js

3. 机器学习模型训练与验证

模型基于 TensorFlow.js 构建，使用 MinMax-Scaler 算法对时间序列数据进行预处理。默认使用 80% 的数据进行训练，20% 用于验证。

训练模型:

$ node tf_train.js
# 或
$ server.sh

验证模型: 验证完成后，真实值、预测值及相关属性将自动更新至 MongoDB。

$ node tf_validate.js

基本使用

完成上述步骤后，即可进行实时预测。加载已训练好的模型权重，订阅 Kafka 的第二个 Topic，系统将根据前 7 个时间序列日志实时预测第 8 个事件。

启动实时预测消费者:

$ node ml_consumer.js

输出示例: 终端将实时输出预测结果，格式如下：

Prediction [attribute] [predicted value].
example - Prediction Open 0.12453 

此时，系统已实现数据流采集、存储、模型训练与实时推理的并行处理。您可以查看 MongoDB 中的 MLCharts 相关集合来监控模型性能图表。