- 示例工程
【免费下载链接】Stock-Prediction-Models
Gathers machine learning and deep learning models for Stock forecasting including trading bots and simulations
本篇技术指南聚焦 Stock-Prediction-Models 仓库中的realtime-agent模块，讲解如何将 agent 目录下的 Evolution Strategy 交易 Agent 训练成一个"实时数据进、交易决策出"的在线交易服务：先在多只股票上离线训练一个不依赖反向传播的神经网络交易策略，再把训练好的模型序列化为model.pkl，最终通过 Flask 暴露成 HTTP 接口，供 websocket、行情 API 等任意实时数据源逐笔调用。读完本文，你将掌握该模块从数据准备、多股票轮训、模型导出到 Flask 服务启动与客户端接入的完整实战链路，并理解其窗口状态构造、三动作决策与收益奖励函数等底层实现细节。
一、realtime-agent 模块概览与设计思路
realtime-agent是仓库中面向"实时交易"的落地模块，其核心思路一句话可以概括为：把离线强化学习训练出的交易策略，封装成一个无状态依赖的 HTTP 服务。
整个模块的文件组成如下（均位于 realtime-agent 目录）：
| 文件 | 作用 |
|---|---|
| realtime-evolution-strategy.ipynb | 训练 notebook：多股票轮训 Evolution Strategy 交易模型，并导出model.pkl |
| app.py | Flask 服务端：加载model.pkl，对外提供/trade、/balance等接口 |
| request.ipynb | 客户端示例：用requests库逐条喂入实时行情并读取交易指令 |
model.pkl | 训练好的模型权重（Model类实例） |
TWTR.csv、GOOG.csv、FB.csv、LB.csv、MTDR.csv、CPRT.csv、FSV.csv、TSLA.csv、SINA.csv、GWR.csv、AMD.csv、LYFT.csv | 训练与回测用的股票历史数据（含Date / Open / High / Low / Close / Adj Close / Volume字段） |
1.1 与 Evolution Strategy Agent 的关系
如 realtime-agent/README.md 所述，"this model based on evolution-strategy"，即本模块是对 agent/6.evolution-strategy-agent.ipynb 的实时化改造。二者的差别在于：
- 离线版本：在单只股票上完成训练与回测，输出买卖信号图；
- 实时版本：用同一套
Deep_Evolution_Strategy+Model+Agent代码，在多只股票上轮流训练同一个模型以覆盖更多随机形态，然后把模型权重存盘，交给 Flask 进程常驻运行。
这一设计使得训练（notebook 一次性完成）与推理（app.py常驻）彻底解耦，服务进程不需要任何训练依赖，只需加载权重即可逐笔决策。
二、训练阶段：多股票轮训一个 Evolution Strategy 交易模型
2.1 训练数据与股票清单
README 明确给出了作者训练时使用的股票清单：
['TWTR.csv', 'GOOG.csv', 'FB.csv', 'LB.csv', 'MTDR.csv', 'CPRT.csv', 'FSV.csv', 'TSLA.csv', 'SINA.csv', 'GWR.csv']而在 realtime-evolution-strategy.ipynb 的实际训练代码中，是通过os.listdir自动收集目录下所有 CSV（排除用于测试的TWTR.csv）得到的，因此实际参与轮训的还包括AMD.csv与LYFT.csv：
stocks = [i for i in os.listdir(os.getcwd()) if '.csv' in i and not 'TWTR' in i] stocks # ['CPRT.csv', 'AMD.csv', 'GWR.csv', 'MTDR.csv', 'GOOG.csv', 'TSLA.csv', # 'FSV.csv', 'LB.csv', 'FB.csv', 'SINA.csv', 'LYFT.csv']关键设计：TWTR.csv被特意排除在训练集外，只用作测试/实时演示数据，以此检验模型对"未见过的股票"的泛化能力。README 同时建议："You might want to add more to cover more stochastic patterns"——即可以继续补充更多股票 CSV 来覆盖更丰富的随机形态，提升模型的鲁棒性。
2.2 核心超参数
训练代码中先集中定义了一组超参数（realtime-evolution-strategy.ipynb 第 11 个 cell）：
skip = 1 layer_size = 500 output_size = 3 window_size = 20同时Agent类内部还定义了 Evolution Strategy 的优化超参数（与 app.py 中一致）：
class Agent: POPULATION_SIZE = 15 SIGMA = 0.1 LEARNING_RATE = 0.03各参数含义与取值说明：
| 参数 | 取值 | 含义 |
|---|---|---|
window_size | 20 | 决策时回看的滑动窗口长度（天），只有队列攒满 20 个数据点后才开始交易 |
skip | 1 | 遍历时间序列时的步长，取 1 表示逐日评估 |
layer_size | 500 | 隐藏层神经元数量，模型结构为单隐藏层全连接网络 |
output_size | 3 | 输出动作数：0持有不动、1买入、2卖出 |
POPULATION_SIZE | 15 | 每代随机扰动的种群个体数 |
SIGMA | 0.1 | 权重扰动噪声标准差，控制探索强度 |
LEARNING_RATE | 0.03 | 权重更新步长 |
2.3 状态空间构造：从行情到特征向量
get_state函数负责把最近window_size个数据点转换成模型输入。以parameters = [Close 序列, Volume 序列]为例，每个序列在窗口内生成两类特征（realtime-evolution-strategy.ipynb）：
- 一阶差分：
block[i+1] - block[i]，共window_size - 1 = 19个，刻画序列的短时变化； - 相对窗口首值：
block[i] - block[0]，共 19 个，刻画序列相对窗口起点的累积偏移。
因此每个序列输出 38 维，两个序列合计 76 维。notebook 中 "Our output size fromget_stateis 38" 指的是单个序列的状态维数。随后Agent.get_state还会拼接 3 个账户维度（app.py 第 241-253 行）：
len_inventory：当前持仓数量；z_inventory：持仓均价的 z-score（相对 Close 序列的均值与标准差）；z_capital：剩余资金（缩放后）的 z-score。
最终输入维度为76 + 3 = 79，与 notebook 执行结果input_size = 79完全一致。若想加入更多列（如 Open、High、Low），只需把对应序列加进parameters列表并同步调整input_size，这一点在 notebook 中也有明确说明："you can choose any parameters you want from your DataFrame"。
2.4 多股票轮训循环
训练循环的核心逻辑（realtime-evolution-strategy.ipynb）：
model = Model(input_size = input_size, layer_size = layer_size, output_size = output_size) agent = None for no, stock in enumerate(stocks): print('training stock %s'%(stock)) df = pd.read_csv(stock) real_trend = df['Close'].tolist() parameters = [df['Close'].tolist(), df['Volume'].tolist()] minmax = MinMaxScaler(feature_range = (100, 200)).fit(np.array(parameters).T) scaled_parameters = minmax.transform(np.array(parameters).T).T.tolist() initial_money = np.max(parameters[0]) * 2 if no == 0: agent = Agent(model = model, timeseries = scaled_parameters, skip = skip, initial_money = initial_money, real_trend = real_trend, minmax = minmax) else: agent.change_data(timeseries = scaled_parameters, skip = skip, initial_money = initial_money, real_trend = real_trend, minmax = minmax) agent.fit(iterations = 100, checkpoint = 10) print()要点：
- 数据预处理：使用
MinMaxScaler(feature_range=(100, 200))将Close与Volume同时缩放到[100, 200]区间（这样资金与价格处于同一量纲）；initial_money = 最大收盘价 × 2，保证初始资金足以买至少 2 个单位。 - 同一模型、不同数据：第一只股票创建
Agent，后续股票通过agent.change_data(...)切换数据（内部调用_initiate()重置inventory、capital、queue），但模型权重持续保留并在上一只股票的基础上继续进化——这就是"用一个模型覆盖多只股票随机形态"的机制。 - 训练代数：每只股票训练
iterations = 100，每 10 代打印一次 reward。notebook 中记录的实测日志显示，单只股票训练约耗时 16～19 秒，例如 CPRT 从-1.25进化到30.14，AMD 最高到55.36，而 FB 全程为负（-28.70→-17.51）。notebook 对此给出了明确提示："If you saw the whole training session on certain stocks are negatives (like FB), means that, that stock markets are losing very bad"——即该股票区间整体下跌，策略自然难以盈利，这是市场环境所致，并非训练错误。
2.5 奖励函数：交易收益率与资金增长率
Agent.get_reward（app.py 第 255-282 行）定义了个体的适应度，用于指导权重进化：
invests = np.mean(invests) # 每笔卖出的平均收益率（%） if np.isnan(invests): invests = 0 score = (starting_money - initial_money) / initial_money * 100 # 资金增长率（%） return invests * 0.7 + score * 0.3即奖励 = 平均单笔收益率 × 0.7 + 总资金增长率 × 0.3，更看重每笔交易的质量而非单纯的总收益。Deep_Evolution_Strategy.train则通过(rewards - mean) / (std + 1e-7)对群体奖励做标准化后，按如下公式更新权重：
weights = weights + learning_rate / (population_size * sigma) * dot(A.T, rewards).T其中A为群体扰动矩阵，本质是 OpenAI ES 风格的"无梯度策略进化"。
2.6 导出模型
训练与测试（agent.buy()会在测试集上打印逐日买卖指令与 softmax 概率，并画出买入/卖出信号图）完成后，用深拷贝 + pickle 导出权重：
import copy import pickle copy_model = copy.deepcopy(agent.model) with open('model.pkl', 'wb') as fopen: pickle.dump(copy_model, fopen)注意使用copy.deepcopy隔离训练中的模型对象，确保导出的权重是稳定的快照。
三、部署阶段：用 Flask 将模型做成实时交易服务
3.1 启动服务
训练并导出model.pkl后，直接在 realtime-agent 目录下运行（README 明确给出的命令）：
python3 app.py预期输出：
* Serving Flask app "app" (lazy loading) * Environment: production WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Debug mode: off * Running on http://0.0.0.0:8005/ (Press CTRL+C to quit)服务默认监听0.0.0.0:8005。README 特别提示了 Flask 开发服务器的警告——该服务适合作为原型/内部实时策略服务，若用于生产环境，应替换为 gunicorn 等生产级 WSGI 服务器。
3.2 启动时的全局初始化
app.py在加载模型后会做一次初始化（realtime-agent/app.py 第 342-357 行）：
with open('model.pkl', 'rb') as fopen: model = pickle.load(fopen) df = pd.read_csv('TWTR.csv') real_trend = df['Close'].tolist() parameters = [df['Close'].tolist(), df['Volume'].tolist()] minmax = MinMaxScaler(feature_range = (100, 200)).fit(np.array(parameters).T) scaled_parameters = minmax.transform(np.array(parameters).T).T.tolist() initial_money = np.max(parameters[0]) * 2 agent = Agent(model = model, timeseries = scaled_parameters, skip = skip, initial_money = initial_money, real_trend = real_trend, minmax = minmax)注意app.py用TWTR.csv来拟合MinMaxScaler并确定初始资金，因此训练与部署时的特征缩放必须保持一致（同样的feature_range=(100, 200)与同样的特征列）。如需切换交易标的，应同步修改这里的参考数据文件，并使用change_data或直接替换 CSV。
3.3 路由一览
app.py共注册了 6 个 GET 接口：
| 路由 | 参数 | 返回内容 |
|---|---|---|
/ | 无 | {'status': 'OK'}，健康检查 |
/balance | 无 | 当前剩余资金（agent._capital） |
/inventory | 无 | 当前持仓列表（缩放后的买入价） |
/queue | 无 | 当前滑动窗口队列（缩放后的[close, volume]序列） |
/trade | data（JSON 数组字符串） | 本次交易决策结果 |
/reset | money（JSON 数字） | 重置资金与持仓，返回true |
3.4/trade的决策流程
/trade是核心接口，其处理逻辑为（realtime-agent/app.py 第 379-382 行）：
@app.route('/trade', methods = ['GET']) def trade(): data = json.loads(request.args.get('data')) return jsonify(agent.trade(data))Agent.trade的内部流程（第 159-221 行）：
- 输入约束：
data必须形如[close, volume]（源码 docstring 明确要求 "you need to make sure the data is [close, volume]"）； - 缩放：用
minmax.transform将原始行情缩放到[100, 200]，同时保留real_close用于真实资金结算； - 维护队列：把缩放后的数据推入
_queue，若队列长度已达window_size = 20则先弹出最旧的数据（FIFO）； - 数据不足判定：若
len(_queue) < window_size，直接返回{'status': 'data not enough to trade', 'action': 'fail', ...}——这是首次调用最常见的返回，需要连续喂入至少 20 个数据点后才会开始真正决策； - 构造状态并推理：用
get_state构造 79 维输入，经model.predict得到 3 维 logits，act_softmax返回argmax动作与 softmax 概率； - 执行动作：
action == 1且缩放后资金充足 → 买入 1 单位，返回{'action': 'buy', 'status': 'buy 1 unit, cost %f', 'balance': ...}；action == 2且有持仓 → 卖出 1 单位（FIFO 出仓），计算investment（收益率百分比）与gain（盈亏额），返回{'action': 'sell', ...}；- 其余情况 → 返回
{'action': 'nothing', 'status': 'do nothing', ...}。
每次调用都会附带timestamp字段（datetime.now()字符串），便于审计与回放。README 中给出的响应示例：
{'action': 'sell', 'balance': 971.1199990000001, 'investment': '10.224268 %', 'status': 'sell 1 unit, price 16.709999', 'timestamp': '2019-05-26 01:12:10.370206'} {'action': 'nothing', 'balance': 971.1199990000001, 'status': 'do nothing', 'timestamp': '2019-05-26 01:12:10.376245'}3.5 其他辅助接口
/balance：返回agent._capital（真实资金），用于监控账户状态；/inventory：返回缩放后的持仓价列表，可据此判断当前仓位；/queue：返回当前滑动窗口内的数据点，可用于调试状态构造；/reset：requests.get('http://localhost:8005/reset?money=100')可重置资金并把_queue、_inventory清空，开启新一轮交易会话。
四、客户端接入：curl 与 request.ipynb
4.1 一行 curl 发起交易
README 给出的最小示例：
curl http://localhost:8005/trade?data=[13.1, 13407500]其中data即[close, volume]。若服务端队列尚未攒满 20 个点，会返回data not enough to trade；攒满后则返回buy / sell / nothing之一的决策 JSON。
4.2 request.ipynb 完整接入流程
request.ipynb 演示了用 Python 客户端模拟"每日新增一条行情"的接入方式，整体分三步：
第 1 步：准备实时数据。以TWTR.csv作为实时数据源（README 注明 "every request means new daily data"，即每个请求代表一天的新数据）：
df = pd.read_csv('TWTR.csv') close = df['Close'].tolist() volume = df['Volume'].tolist()第 2 步：查询余额。用/balance获取当前资金，确认初始资金定义（源码中为initial_money = max(Close) * 2，也可通过/reset?money=...覆盖）：
requests.get('http://localhost:8005/balance').json() # 93.51999599999999第 3 步：逐条喂入数据并读取决策。将[close, volume]序列化为 JSON 字符串后拼进 URL：
import json data = json.dumps([close[0], volume[0]]) # '[33.419998, 13407500]' requests.get('http://localhost:8005/trade?data=' + data).json() # {'action': 'fail', 'balance': 93.51999599999999, # 'status': 'data not enough to trade', 'timestamp': '2019-08-31 02:40:10.625022'}批量循环喂入 200 个数据点：
for i in range(200): data = json.dumps([close[i], volume[i]]) requested = requests.get('http://localhost:8005/trade?data=' + data).json() print(requested)从输出可以看到：前 19 个请求全部返回'action': 'fail'（队列未满），第 20 个数据点开始出现真正的买卖指令（如buy 1 unit, cost 44.950001、sell 1 unit, price 44.259998等），之后模型依据窗口内行情持续给出buy / sell / nothing决策。这就是window_size = 20语义的直接体现：回看 20 个历史数据点后才开始交易（request.ipynb 中有明确说明 "atleast same as window_size size ... it only look back 20 historical data to trade"）。
4.3 自定义参数接入的注意事项
request.ipynb 特别强调了两点：
- 当前训练/部署只用到了
Close与Volume两个特征，所以请求数据必须严格为[close, volume]；若要使用其他特征（如 Open、High、Low），需要同步修改训练代码中的parameters列表、app.py中的缩放器拟合逻辑以及请求格式； - 本模块把 CSV 当作"逐日新增的实时数据"来驱动，README 的 Notes 进一步指出可以把这个代码与实时 socket 或任意行情 API 集成，让数据源从"读文件"变成"监听推送"，即可落地成真正的实时交易流水线。
五、集成扩展：接入实时行情源
README 的 Notes 给出了明确的扩展方向：
You can use this code to integrate with realtime socket, or any APIs you wanted, imagination is your limit now.
具体来说，将/trade调用嵌入到数据消费循环中即可完成实时化：
- WebSocket / FIX 行情流：在行情回调中把
[close, volume]组装成 JSON，调用/trade，拿到action后对接券商下单 API； - 定时轮询行情 API：定时拉取最新 K 线并喂给
/trade； - 多标的轮询：对每个标的维护独立的
Agent实例（或复用change_data切换参考数据），实现多股票实时监控。
由于服务端状态（_queue、_inventory、_capital）完全保存在Agent实例内存中，客户端只需保证按时间顺序、逐条提交即可，接入成本极低。
六、关键源码速查
以下文件路径供深入阅读时直接定位：
- 服务端实现：realtime-agent/app.py（
get_state状态构造见第 25-40 行，Agent.trade见第 159-221 行，get_reward见第 255-282 行，路由注册见第 359-389 行） - 训练 notebook：realtime-agent/realtime-evolution-strategy.ipynb（多股票轮训见第 12 个 cell，模型导出见第 19 个 cell）
- 客户端示例：realtime-agent/request.ipynb（
/trade循环喂数据见第 7 个 cell） - 离线版算法出处：agent/6.evolution-strategy-agent.ipynb
- 数据文件：TWTR、GOOG、FB、LB、MTDR、CPRT、FSV、TSLA、SINA、GWR、AMD、LYFT 等 CSV 均位于 realtime-agent 目录
七、使用前提与注意事项
- 运行环境：需要 Python 3 与 Flask、numpy、pandas、scikit-learn（
MinMaxScaler）；notebook 元数据显示开发环境为 Python 3.6.8； - 特征一致：训练、缩放器拟合、请求数据三者必须保持
[close, volume]的两特征格式一致，否则推理结果会失真； - 冷启动窗口：服务启动后需喂满 20 个数据点才会出决策，这是
window_size=20的设计使然，并非故障； - 单笔交易约束：模型每次最多买卖 1 个单位（与仓库 README "This agent only able to buy or sell 1 unit per transaction" 一致），资金不足时会静默转为
nothing，因此初始资金应覆盖至少 2 个单位的买入成本； - 回测结果仅代表历史区间：notebook 中出现的负收益（如 FB、LB）是相应股票在训练区间内走势疲弱所致，不代表策略在任意市场环境中都能盈利，实盘前应充分回测与风控。
- 示例工程
【免费下载链接】Stock-Prediction-Models
Gathers machine learning and deep learning models for Stock forecasting including trading bots and simulations
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
