简介面向量化投资初学者与强化学习实践者的股市择时策略实验项目解压后可直接运行完整流程。项目基于深度Q网络DQN涵盖从历史数据收集、缺失值处理、状态表示构建到市场环境建模、策略网络训练、回测评估与结果可视化的全链路适合理解智能体如何根据价格、成交量、技术指标等市场状态做出买入、持有或卖出决策并逐步优化长期收益。压缩包共82个文件以14个Python源码文件为核心辅以数据库文件、模型权重、输出结果及配置说明等整体仅3.53MB结构清晰便于快速上手。已有210人学习资源内含可复现的训练脚本、评估指标与可视化代码可对照运行观察学习过程也可作为进一步探索双Q学习、Actor-Critic等进阶算法的起点适合课程设计、毕业设计或个人研究参考。1. 强化学习择时不是预测涨跌而是学习买卖节奏很多把强化学习用在股市上的项目最后都会变成用 LSTM 预测明天涨跌那就是监督学习而不是决策学习。这个项目的关键区别在于它把择时形式化成 MDP智能体看到状态、选择动作拿到与收益挂钩的奖励目标直接对齐累计回报。它适合看不懂“Q 网络和交易到底怎么接”的人也适合人工智能大作业或强化学习入门实验。解压 zip 后是一个完整工程数据、环境、训练、评估、可视化都在能对照着跑通。2. 从股票数据到强化学习状态preprocess 与特征构造2.1 数据从哪来存成什么样打开项目根目录能看到 collector.py、data_work.py、preprocess.py还有 stock.db 和 rew.db 两个 SQLite 数据库。常见做法是先用 collector.py 从行情源拉取日线数据写入 stock.db字段至少包括 date、open、high、low、close、volume。rew.db 则用来记录训练过程中的奖励曲线方便后期对着可视化页面复盘。用 SQLite 而不是 CSV 的好处是随机查询和增量更新都更简单回测时按日期切分也快。第一次跑通时我一般会建议先不换数据源直接沿用项目里已经生成好的 stock.db。这样做可以跳过网络请求和配额限制先确认环境能跑起来。等到训练和评估都正常了再换成自己感兴趣的美股或 A 股数据。2.2 状态向量不是原始 K 线而是归一化特征如果直接把过去 N 天的收盘价拼成一个向量喂给网络模型会极其不稳定因为价格绝对水平随时间漂移训练前期看到的 100 元和训练后期看到的 100 元含义不同。这个项目里的 preprocess.py 和 data_work.py 解决的问题就是把 K 线加工成相对稳定的特征。我拆过一遍之后认为最少要包含下面几类特征特征窗口/计算方式作用收益率当日 close / prev_close - 1去除绝对价格影响均线偏离close / MA(5) - 1 和 close / MA(20) - 1捕捉短期与中期趋势偏离RSI14 日相对强弱指标判断超买超卖区域布林带位置(close - lower) / (upper - lower)识别价格在通道中的相对位置成交量变化volume / MA(volume, 5) - 1观察放量缩量这些特征统一做 z-score 或 min-max 归一化最后拼成一维向量。注意归一化的统计量只能从训练集计算再应用到测试集否则会把未来信息泄漏进状态表示。具体到项目里globals.py 通常存放全局变量和特征列名preprocess.py 调用 data_work.py 计算指标再把结果按日期顺序切成时间窗口。2.3 preprocess.py 的片段与参数说明下面是一种常见的状态构造代码写法不唯一但逻辑和这个项目结构是对得上的import numpy as np import pandas as pd def build_state(df, idx, lookback10): df: 已按时间排序的行情 DataFrame idx: 当前时间点 lookback: 回溯窗口长度 window df.iloc[idx - lookback 1: idx 1] features [] # 1. 当前收益率与均线偏离 close window[close].values ret close[-1] / close[-2] - 1 ma5 window[close].rolling(5).mean().iloc[-1] ma20 df[close].rolling(20).mean().iloc[idx] if idx 20 else close.mean() features.extend([ret, close[-1] / ma5 - 1, close[-1] / ma20 - 1]) # 2. 布林带位置 std20 window[close].rolling(20).std().iloc[-1] upper ma20 2 * std20 lower ma20 - 2 * std20 features.append((close[-1] - lower) / (upper - lower) if upper ! lower else 0.0) # 3. 归一化后的成交量变化 vol window[volume].values vol_ma5 window[volume].rolling(5).mean().iloc[-1] features.append(vol[-1] / vol_ma5 - 1 if vol_ma5 0 else 0.0) return np.array(features, dtypenp.float32)这段代码的核心是用最近 lookback 根 K 线计算一个固定长度的特征向量。需要注意两点第一rolling(20) 在窗口左侧可能取不到值工程里要么放弃序列开头要么像我上面那样回退到局部均值第二特征的拼接顺序必须固定训练、验证、预测三套流程要用同一个 build_state 函数否则模型看到的“同一状态”数值分布完全对不上。实际跑的时候还要在外部做一个滑窗生成器从 idxlookback 开始每次生成一个 (state, next_state, action, reward) 的样本序列供强化学习环境调用。这个环节最容易出现的错误是索引错位我通常会在生成器里加一条断言确保 idx 1 不超过总长度。3. 环境与奖励函数决定策略收敛方向的关键3.1 gym 风格交易环境的状态流项目里的 env.py 是一个典型的 gym 环境。reset 方法把持仓清零、日期归位返回初始状态step 方法接收智能体的动作更新持仓和现金计算下一个状态和奖励。动作空间通常是离散的0 表示持有1 表示买入2 表示卖出。相比连续动作离散动作更适合日线级别的择时因为调仓频率低动作意义更明确DQN 这类价值算法直接就能用。把环境单独抽象出来还有一个好处训练和回测共用同一套交互逻辑不会出现“训练时用 A 规则评估时用 B 规则”的偏差。常见做法是把手续费、印花税、最小交易单位都固定成全局常量写在 globals.py 里。3.2 奖励函数用收益还不够要惩罚回撤这个项目叫“择时策略实验”说明奖励函数设计比网络结构更值得关注。最朴素的奖励是 r_t 持仓收益 - 交易成本但这样训练出来的策略容易变得过于激进只要预期收益为正就一直持仓回撤大得吓人。稍微成熟一点的做法是加入风险惩罚项比如r_t 持仓收益率 - transaction_cost - lam * abs(drawdown_delta)其中 drawdown_delta 表示当前回撤相对上一时刻的变化量lam 是惩罚系数。这样策略在接近回撤低点时会更谨慎。更进一步的方案是把一段时间内的夏普比率折算成奖励也就是用“收益 / 波动”替换“绝对收益”让策略主动追求稳健。不过奖励函数越复杂训练越难收敛。我的经验是先用最简单的“收益减手续费”跑通整个流程确认 DQN 能学到基本规律后再加回撤惩罚。否则一上来就加一堆正则项出了问题很难定位是网络问题还是奖励设计问题。3.3 环境实现片段与参数表一个精简版的环境核心逻辑如下class TradingEnv: def __init__(self, states, max_holding10, transaction_cost0.001, reward_scale1.0): self.states states # shape: [T, state_dim] self.max_holding max_holding self.transaction_cost transaction_cost self.reward_scale reward_scale self.reset() def reset(self): self.idx 0 self.holding 0 self.cash 0.0 return self.states[self.idx] def step(self, action): prev_price self.states[self.idx, 4] # 假设价格放在特征第 4 列 self.idx 1 done self.idx len(self.states) - 1 if not done: price self.states[self.idx, 4] else: price prev_price # 执行动作 if action 1 and self.holding self.max_holding: cost price * self.transaction_cost self.holding 1 self.cash - price cost elif action 2 and self.holding 0: cost price * self.transaction_cost self.holding - 1 self.cash price - cost # 以总资产变化率作为奖励 net_asset self.cash self.holding * price reward (net_asset - self.last_asset) / self.last_asset self.last_asset net_asset return self.states[self.idx], reward, done, {}这个环境把“买入”简化成每次买 1 股max_holding控制最大仓位transaction_cost同时作用于买入和卖出。奖励用的是每股等价资产的变化率所以不会因为持仓数量不同而出现量纲问题。实际项目中你还需要把done之后的最终持仓按最后价格强制平仓并在step里加一个状态超出边界的守卫否则最后几步的奖励会被低估。参数可以这样初始化参数建议值说明max_holding10每笔最多买 10 个单位相当于仓位控制transaction_cost0.001双边千分之一约等于 A 股单边佣金印花费的感觉reward_scale1.0奖励缩放系数太大容易震荡太小收敛慢需要说明的一点是这个环境里我用states[self.idx, 4]假设价格在特征列索引 4但你从 preprocess 输出的特征向量不一定有这个约定。项目里多半会用一个全局常量PRICE_IDX来维护而不是硬编码。类似于这样的小约定往往是调参时最容易被忽略的坑。4. 模型、训练与评估从 DQN 到落地验证4.1 DQN 主网络与 VAE 辅助表征model.py 中通常定义两个类一个是 Q 网络输入状态向量输出每个动作的 Q 值另一个是目标网络结构和 Q 网络一致但参数滞后更新。项目里还有 VAE 目录这不是用来做图像生成而是对原始市场特征做降维和重建。它把高维、冗余的行情特征压缩成一个隐向量再用隐向量作为 DQN 的输入或者作为状态表示学习器。这样做的效果是去除相关性强的特征之间的噪声让 Q 网络更专注于真正影响收益的维度。深度强化学习在这里的定位是逼近 Q 函数。因为股市状态是连续向量动作是离散的所以 Q 网络输出为 [batch, 3] 的矩阵3 对应持有、买入、卖出。训练时使用 MSE 损失比较predicted_q和target_q目标值用贝尔曼方程计算target_q reward gamma * max_a Q_target(next_state, a)4.2 主训练循环与经验回放参数main.py 是训练入口它的结构一般是初始化环境、初始化两个网络、初始化经验回放池然后进入 episode 循环。每个 episode 从 reset 开始智能体按 epsilon-greedy 策略选择动作即 epsilon 概率下随机探索否则取 Q 值最大的动作。交互产生的 (state, action, reward, next_state, done) 存入经验池再随机采样一个小批量更新 Q 网络。目标网络每隔若干步同步一次参数。for episode in range(num_episodes): state env.reset() total_reward 0 while True: action select_action(state, epsilon) next_state, reward, done, _ env.step(action) replay_buffer.push(state, action, reward, next_state, done) if len(replay_buffer) batch_size: batch replay_buffer.sample(batch_size) loss update_q_network(batch) state next_state total_reward reward if done: break if episode % target_update 0: target_net.load_state_dict(q_net.state_dict())这个循环里有几个参数直接决定训练效果。epsilon从 1.0 线性衰减到 0.05控制探索程度gamma设置为 0.99 会让智能体看重长期收益适合择时这种需要持仓多天的任务batch_size常用 64target_update一般 50 到 100 步同步一次。如果发现训练不稳定优先检查reward的尺度其次检查gamma是否过大导致回报方差太高。4.3 训练完成后怎么评估夏普比率与最大回撤模型训练完不代表策略就能用。evaluate.py 的作用是把训练好的 Q 网络对历史数据做滚动决策计算策略的每日收益序列然后统计夏普比率、最大回撤、胜率和累计收益。夏普比率定义为年化收益除以年化波动数值越高说明风险调整后收益越好最大回撤则是净值从峰值跌到谷底的最大幅度能直接反映策略最差体验。下面是评估指标的计算代码通常写在 evaluate.py 里def sharpe_ratio(returns, periods252): return np.sqrt(periods) * returns.mean() / (returns.std() 1e-8) def max_drawdown(nav): peak np.maximum.accumulate(nav) return np.min((nav - peak) / peak)先解释sharpe_ratio输入是日收益率序列periods252表示按一年 252 个交易日年化。分母加1e-8是为了避免全涨或全跌时标准差为零。max_drawdown则是先算累计净值的历史峰值再用当前净值减峰值得到一个负值取最小值就是最大回撤。不要只看夏普比率一个策略如果 3 年赚 50% 但中间回撤 30%实盘很难拿得住。我一般会把回撤阈值写进评估逻辑超过设置值就认为策略失效。在评估阶段还要注意状态构造不能使用未来数据。比如用全样本的均值做归一化等于让模型在测试时刻偷看到了平均值回测曲线会虚高。项目里如果把评估和训练放在同一个处理流程里这一点尤其危险。常见的修正方式是在训练前先按时间切分只在训练段计算归一化统计量评估段单独加载。5. 可视化与回测排错最容易忽略的三个细节可视化是这项目里最容易被忽略的部分。visualize.py 和 streamlit.css 组成的页面能同时看到净值曲线、持仓变化、动作分布和奖励曲线。跑通 main.py 之后我一般会直接把 rew.db 里的训练奖励和 evaluate.py 输出的净值加载到 Streamlit 里用同一个页面检查“训练是否收敛”和“回测是否平滑”而不是在终端里看一堆数字。排错时最容易踩的三个坑我按重要性排序第一归一化统计量泄漏。如果你在 preprocess 里用全样本的均值方差归一化看起来训练和评估都正常但放到未来数据上会立刻失效。修复方法是把 scaler 保存下来测试集单独使用训练集的统计量。检查方法也很简单在 predict.py 里随机打印几个测试样本的状态看看数值是否和训练状态分布一致。第二环境最后一步没有强制平仓。许多简化版环境在done时直接返回奖励没有把剩余持仓按最后价格清掉。这会导致策略在接近末尾时故意不卖出因为未平仓的浮盈没有被结算进奖励。验证方法是在 evaluate.py 中对比“包含强平”和“不包含强平”的累计收益差异超过 0.5% 就要警惕。第三动作重复导致交易成本失真。一个 episode 里如果连续买入多次而没有检查max_holding手续费会被夸张放大。可以在可视化页面里加一个“每 episode 交易次数”柱状图正常情况下每天调仓比例应该很低。如果发现平均每次决策都在买或卖问题大概率出在奖励函数中交易成本权重不够。还有一个小技巧适合做强化学习实例时验证策略是否真的学到了规律把训练好的模型在随机扰动后的数据上跑一遍如果性能骤降说明它可能只是记住了价格序列而不是学到了状态到动作的映射。你会更愿意把功夫花在特征和奖励上。本文还有配套的精品资源点击获取