简介面向金融量化研究与深度学习开发者这份压缩包完整实现了基于DQN算法的沪深300指数择时交易策略并与OpenAI Gym框架构建的智能体环境交互系统配套使用。包内共10个文件包含5个Python脚本分别对应智能体、环境、记忆回放、模型与功能模块、Jupyter Notebook运行入口、沪深300指数数据表格以及说明文档整体仅386KB结构紧凑便于快速部署与扩展。目前已有127人学习适合希望从代码层面理解深度强化学习在金融择时中应用的读者参考。借助该资源可清晰把握从环境定义、状态特征、动作奖励到DQN训练交互的完整链路同时通过附带的数据与文档缩短复现成本是入门量化择时与强化学习结合方向的实用工具包。1. 深度强化学习择时先看清决策问题再谈DQN与Gym做沪深300择时最常见的路径是训练一个分类器预测“明天涨还是跌”预测对了做多错了空仓。这类方法有一个结构性问题预测误差和交易损失不是一一对应的。指数涨0.5%但模型预测涨1%信号“错了”但仓位赚钱预测对方向次日跳空加上手续费净值反而亏损。把择时当成单步预测等于不承认交易是“先决策、后观察、再调整”的序列过程。深度强化学习换了问法不预测涨跌而是直接学一条让累计收益最大的策略。你给智能体一个状态它输出仓位动作环境把收益和成本的反馈还给它DQN算法在这个 Agent 与 Environment 的交互循环里更新Q网络。这套方案不要求你对市场方向有先验判断只要求把状态、动作、奖励定义合理再用 OpenAI Gym 的接口把模拟盘搭出来。适合已经写过回测、熟悉Python但不确定深度强化学习怎么落到量化策略的开发者。2. 把沪深300择时重写成MDP状态、动作、奖励的设计决定DQN能否收敛2.1 状态空间给智能体看的不是行情快照而是归一化后的特征窗口择时任务里状态就是Agent在t时刻能观察到的全部信息。直接给收盘价序列不行DQN用神经网络做函数逼近价格的绝对水平和网络权重初始化尺度完全不匹配训练时会出现“loss数值正常但Q值不增长”的假象。常见做法是先构造特征再滚动归一化。import numpy as np import pandas as pd def build_features(df: pd.DataFrame, window: int 20) - np.ndarray: close df[close] volume df[volume] feat pd.DataFrame(indexdf.index) feat[mom] close.pct_change(1) # 当日收益率 feat[ma_bias] (close - close.rolling(10).mean()) / close # 均线偏离 feat[vol] close.pct_change().rolling(window).std() # 20日波动率 feat[rsi] _rsi(close, 14) / 100.0 # RSI映射到[0,1] feat[vol_ratio] volume / volume.rolling(window).mean() - 1.0 feat feat.replace([np.inf, -np.inf], 0.0).fillna(0.0) # 滚动z-score只用过去60天统计量避免未来数据泄露 mean feat.rolling(60).mean() std feat.rolling(60).std().replace(0, 1e-6) feat (feat - mean) / std return np.nan_to_num(feat.values, nan0.0)动量取1日收益看的是短期惯性均线偏离反映价格离MA10的距离属于反转信息RSI除以100是为了让输入量纲进入0到1区间否则网络第一层的梯度会被大数值特征主导。归一化使用过去60天的滚动均值和标准差训练和回测共用同一套逻辑这是防止状态泄露的关键如果对整个数据集一次性标准化测试期的统计信息在训练期就被Agent偷偷看到了学出来的策略在回测里好看实盘必然变形。单独某个时刻的特征还不足以支撑决策DQN的Q网络是个静态映射同一个状态今天的收益和明天的收益含义完全不同。实际操作中会把最近10个交易日的特征拼接成一个向量环境里每个step返回的是(history_len, n_features)展平后的长向量网络自行从这段序列里提取“连续下跌三天”这类时序模式。2.2 动作空间DQN只能输出离散动作三档仓位比“买卖二值”更接近实操DQN的价值网络Q(s,a)要求动作是有限离散集。沪深300择时常见的动作设计是空仓、半仓、满仓三档对应整数0、1、2。中间档的作用是让Agent在不确定时降低暴露避免全仓进出被震荡行情来回打脸。如果只有0和1两个动作信号稍不稳定就会产生频繁换仓交易成本会吃掉大部分alpha。动作设计动作集合适配算法主要问题二值空仓/满仓DQN换手率高成本拖累明显三档0/半仓/满仓DQN中间档吸收噪声本文采用连续[-1, 1]DDPG/TD3表达力强但调参成本高、现金约束难处理动作空间不只影响语义还直接决定网络输出维度三档动作让输出层正好是3个节点每个节点代表状态s下执行该动作的期望回报。DQN为什么不做成连续动作因为argmax操作在连续空间上无法直接计算一定要做连续控制只能用actor-critic一类算法训练稳定性和收敛速度都比离散DQN差一截对日频指数择时来说性价比不高。2.3 奖励函数把收益率和交易成本揉进同一个标量奖励是Agent唯一能感知的“对错信号”设计目标只有一个单步收益减去单步交易成本让所有动作在统一的价值尺度下比较。def compute_reward(close_t: float, close_t_1: float, action_t: int, action_t_1: int, cost_rate: float 0.0005) - float: # 仓位以日初持仓结算当日收益 昨日仓位 * 当日涨跌幅 position_t action_t * 0.5 position_t_1 action_t_1 * 0.5 ret position_t_1 * (close_t - close_t_1) / close_t_1 turnover abs(position_t - position_t_1) cost turnover * cost_rate return ret - cost当日收益按日初的旧仓位结算因为当天调仓发生在开盘后新仓位面对的是下一个交易日的涨跌这个细节不处理对Agent会学会“当日追涨、次日被闷杀”的虚假收益。交易成本按换手比例单边万5计算比A股常规的佣金加滑点略保守实际交易费用会收在万2到万8之间取万5能做压力测试。成本必须直接减进收益标量里不能单独设一个“频繁交易惩罚项”。DQN优化的是累计收益期望你单独惩罚换手Agent很快会发现规则漏洞不动也能获得与动接近的收益策略退化成“永远满仓”或“永远空仓”而不是学会择时。奖励绝对值大约在0.001量级Q值方差很小训练曲线会平得像一条直线常见处理是把reward放大100倍再进网络或者改用“策略日收益 - 基准日收益”的超额收益形式后者更稳因为它顺带规避了“长期持有就是最优解”的退化情况。2.4 回合边界片段长度本身也是约束择时是继续性任务但DQN需要明确的回合边界来算折扣回报。常见做法是把每120个交易日设为一个episode相当于半个季度到期强制doneTrue。这样做有两个好处一是沪深300的趋势结构大体在季度级别更替短片段能让Agent反复经历上涨和下跌两种市场状态二是防止单个episode过长导致Q值方差爆炸。起点从第60个交易日之后随机抽取保证滚动归一化窗口始终有完整数据。3. 用OpenAI Gym搭Environment状态转移、交易成本与数据切分3.1 先踩平Gym接口的版本差异OpenAI Gym的接口在0.26版本统一改为新写法reset()返回(obs, info)step()返回(obs, reward, terminated, truncated, info)。网上大量旧教程写的是obs env.reset()、obs, reward, done, info env.step(action)那是0.21及以前的老用法。现在新装的环境基本都按Gymnasium兼容接口实现代码里写成新API并在step()返回5元组就不会遇到“cannot unpack”这类低级错误。3.2 实现Environment类数据、状态转移和成本都在一个类里管做agent开发时最花时间的不是网络而是环境写得不对。状态索引差一个、成本算漏一次换手这些错误不会直接报错只会让训练出的策略在回测里表现出莫名的“能力”。下面是一个可以直接跑的IndexTradingEnv按Gymnasium接口实现import gymnasium as gym from gymnasium import spaces import numpy as np class IndexTradingEnv(gym.Env): def __init__(self, features: np.ndarray, prices: np.ndarray, history_len: int 10, cost_rate: float 0.0005, seed: int 42): super().__init__() self.features features self.prices prices self.history_len history_len self.cost_rate cost_rate self.action_space spaces.Discrete(3) # 0空仓 / 1半仓 / 2满仓 self.observation_space spaces.Box( low-np.inf, highnp.inf, shape(history_len * features.shape[1],), # 10日特征展平 dtypenp.float32, ) self._rng np.random.default_rng(seed) self.reset() def reset(self, *, seedNone, optionsNone): super().reset(seedseed) if seed is not None: self._rng np.random.default_rng(seed) # 起点从归一化窗口之后随机抽取保证有足够历史数据 self._step_idx int(self._rng.integers(60, len(self.features) - 120)) self._pos 0 # 当前动作0/1/2 self._done False return self._get_obs(), {} def step(self, action: int): pos action * 0.5 # 动作映射为仓位比例 prev_pos self._pos * 0.5 prev_close self.prices[self._step_idx - 1] close self.prices[self._step_idx] ret prev_pos * (close - prev_close) / prev_close # 用旧仓位结算当日收益 cost abs(pos - prev_pos) * self.cost_rate # 换手成本 reward float(ret - cost) self._pos action self._step_idx 1 self._done self._step_idx len(self.features) - 1 return self._get_obs(), reward, self._done, False, {} def _get_obs(self) - np.ndarray: start self._step_idx - self.history_len end self._step_idx return self.features[start:end].flatten().astype(np.float32) def render(self): pass状态向量由最近10个交易日的特征展平而来维度是10 * 特征数Agent能看到“连跌三天”这类跨步模式。注意观察空间的shape要在__init__里定死否则gymnasium内部校验会报错。step()里的结算顺序是先用旧仓位算当日收益再按新动作算换手成本这对应真实交易中“收盘后决策、次日执行”的节奏。若把顺序写反Agent会得到一天“低买高卖”的虚假奖励。3.3 数据切分与归一化窗口时序数据的隔离不能省数据用pandas读入后按时间升序排列前80%做训练集后20%做回测集。切分必须按时间顺序不能随机打乱。代码上就是取索引分片import pandas as pd df pd.read_csv(hs300_daily.csv, parse_dates[date]).sort_values(date) split int(len(df) * 0.8) train_df, test_df df.iloc[:split], df.iloc[split:] train_feat build_features(train_df) # 滚动归一化只用train段 train_env IndexTradingEnv(train_feat, train_df[close].values, seed0) # 回测环境用回测段数据重新构造状态统计量不跨段 test_feat build_features(test_df) test_env IndexTradingEnv(test_feat, test_df[close].values, seed1)build_features内部用的是滚动窗口理论上单独对test_df构造特征不会使用train段信息但两个DataFrame在索引边界上会有窗口值缺失需要在构造环境前确认数据长度足够。3.4 安装Gym的常见环境错误在Python 3.11及以上的系统自带环境里直接pip install gymnasium大概率遇到error: externally-managed-environment这是PEP 668的机制系统Python禁止pip往外部环境写包不是包本身装不上。正确做法是先建虚拟环境python3 -m venv .venv source .venv/bin/activate pip install gymnasium numpy pandas torch提示不要图省事加--break-system-packages那会让系统Python的包管理失控。装完检查一下gymnasium.__version__确认用的是0.26以上的接口再跑一节的环境类。4. Agent端实现DQN算法的经验回放、目标网络与训练循环4.1 为什么标准Q-learning做不了这件事表格型Q-learning需要把状态离散成有限格子沪深300的状态是连续向量哪怕只保留5个特征、每个量化成10档组合数也是天文数字。DQN的核心是把Q表换成神经网络输入状态输出每个离散动作的Q值用TD误差做梯度更新。此时“Agent的记忆”不再是一张表而是网络权重加经验回放缓冲池后者正好对应强化学习里的经验回放机制。4.2 网络结构全连接网络对日频择时够用import torch import torch.nn as nn class DQN(nn.Module): def __init__(self, obs_dim: int, n_actions: int, hidden: int 128): super().__init__() self.net nn.Sequential( nn.Linear(obs_dim, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU(), nn.Linear(hidden, n_actions), ) def forward(self, x: torch.Tensor) - torch.Tensor: return self.net(x)obs_dim取环境里observation_space.shape[0]隐藏层128一般够用。指数择时没有图像级高维输入网络太大反而把历史上某段行情的形态“背”下来回测好看、实盘失效。两个隐藏层是经验值加第三层对收益率的提升通常不如调奖励函数明显。4.3 经验回放打破样本相关性环境产生的样本在时间上高度相关——今天的价格和明天强相关直接用连续样本更新网络会让梯度方向偏差很大。经验回放的思路是把交互产生的转移元组存进缓冲区训练时随机采样破坏时间相关性。缓冲区容量设50000相当于400多个episode的总步数容量过小多样性不足过大会让新样本占比太低、学习变慢。from collections import deque import random class ReplayBuffer: def __init__(self, capacity: int 50000): self.buf deque(maxlencapacity) def push(self, s, a, r, s_, done): self.buf.append((s, a, r, s_, done)) def sample(self, batch_size: int): batch random.sample(self.buf, batch_size) s torch.tensor([x[0] for x in batch], dtypetorch.float32) a torch.tensor([x[1] for x in batch], dtypetorch.long) r torch.tensor([x[2] for x in batch], dtypetorch.float32) s_ torch.tensor([x[3] for x in batch], dtypetorch.float32) done torch.tensor([x[4] for x in batch], dtypetorch.float32) return s, a, r, s_, done4.4 训练主循环目标网络与衰减探索目标网络的作用是给TD目标提供一个相对稳定的参考。如果直接用在线网络同时计算Q值和目标值每一轮更新目标也在变训练容易震荡。每200步把在线网络的权重复制给目标网络让目标值的更新频率降下来。obs_dim train_env.observation_space.shape[0] agent DQN(obs_dim, 3) target DQN(obs_dim, 3) target.load_state_dict(agent.state_dict()) buffer ReplayBuffer(50000) optimizer torch.optim.Adam(agent.parameters(), lr3e-4) gamma 0.99 epsilon 1.0 epsilon_min 0.05 batch_size 64 target_update_steps 200 step_count 0 for episode in range(300): obs, _ train_env.reset() done False while not done: if random.random() epsilon: action train_env.action_space.sample() else: with torch.no_grad(): q agent(torch.tensor(obs, dtypetorch.float32).unsqueeze(0)) action int(q.argmax().item()) obs_, reward, terminated, truncated, _ train_env.step(action) done terminated or truncated buffer.push(obs, action, reward, obs_, done) obs obs_ if len(buffer.buf) batch_size: s, a, r, s_, d buffer.sample(batch_size) q_current agent(s).gather(1, a.unsqueeze(1)).squeeze(1) with torch.no_grad(): q_next target(s_).max(1).values y r gamma * q_next * (1 - d) loss nn.MSELoss()(q_current, y) optimizer.zero_grad() loss.backward() optimizer.step() step_count 1 if step_count % target_update_steps 0: target.load_state_dict(agent.state_dict()) epsilon max(epsilon_min, epsilon * 0.995)q_current用gather取出当前状态下实际执行动作的Q值a.unsqueeze(1)把动作张量从[64]变成[64,1]以匹配Q值的二维形状。TD目标y在torch.no_grad()下计算避免梯度穿过目标网络。(1 - d)的目的如果回合在下一状态结束后面的累计回报就不该算进去。epsilon从1.0起步让Agent前期充分随机探索300个episode后衰减到接近0.05进入纯利用阶段。4.5 超参数表与调参顺序参数取值调整方向学习率3e-4loss发散降到1e-4收敛过慢先试5e-4gamma0.99日频数据建议不低于0.99反映长投资周期batch_size64数据量小用32样本充足可128回放容量50000样本多样性不足时加大过渡拟合时减小目标网络更新周期200步训练震荡时缩短到100步epsilon衰减0.995/episode衰减过慢浪费训练过快过早陷入局部最优调参顺序有讲究先看奖励的数值范围再动gamma最后才碰学习率。Q值发散时优先把奖励缩放回0.01以内而不是盲目调学习率。如果loss震荡但不发散把目标网络更新周期缩短如果Q值整体漂移检查奖励函数是不是算错了一步比如用了当日新仓位结算收益。5. 沪深300择时回测验证DQN训练结果的三个指标与一种必做的对照实验5.1 从回测Environment里取出交易序列训练完成后用测试段数据重新构造环境把Agent的输出动作存成数组再拿真实价格算净值曲线。注意动作要保存决策当天的仓位回测时用前一天的仓位结算当日收益这和训练环境里的逻辑必须严格一致。positions [] obs, _ test_env.reset() done False while not done: with torch.no_grad(): q agent(torch.tensor(obs, dtypetorch.float32).unsqueeze(0)) action int(q.argmax().item()) positions.append(action) obs, _, terminated, truncated, _ test_env.step(action) done terminated or truncated positions np.array(positions) * 0.5 def backtest_metrics(pos_ratio: np.ndarray, prices: np.ndarray, cost_rate: float 0.0005): rets np.diff(prices) / prices[:-1] strategy_ret pos_ratio[:-1] * rets # 前一日仓位结算 cost np.abs(np.diff(pos_ratio)) * cost_rate net strategy_ret - cost cum np.cumprod(1 net) total_return cum[-1] - 1 ann_return (1 total_return) ** (252 / len(net)) - 1 ann_vol np.std(net) * np.sqrt(252) sharpe ann_return / ann_vol if ann_vol 0 else 0.0 max_dd (cum / np.maximum.accumulate(cum) - 1).min() return {total_return: total_return, ann_return: ann_return, sharpe: sharpe, max_drawdown: max_dd} metrics backtest_metrics(positions, test_df[close].values) print(metrics)重点看三个数字年化收益是否跑赢同期沪深300本身的买入持有夏普比率在日频策略里0.6以上算可用0.8到1.2算优秀最大回撤有没有比指数本身的回撤明显收窄。再配合两个辅助统计总交易次数和年换手率。如果一年交易不到10次策略大概率退化成了“买入持有”DQN学的不是择时是“满仓最赚”这个价投结论。5.2 多随机种子重复实验一次回测说明不了问题训练过程有随机性初始化权重、epsilon探索、回合起点抽样都会影响最终策略。跑一次实验看到好结果可能是运气也可能是过拟合到测试段的某段行情。判断策略稳定性的最低成本做法是固定不同seed重复跑5遍用相同的超参数和相同的训练/测试切分记录每次的年化收益与最大回撤看均值和离散程度。如果5个seed的年化收益标准差超过均值的一半说明Q网络没有收敛到稳定的策略面问题出在奖励函数或状态特征换网络结构解决不了。操作上把seed同时注入IndexTradingEnv、ReplayBuffer随机采样和PyTorch的manual_seed训练脚本里统一用循环控制。这个“多seed对照”实验比任何单次回测数字都更能说明策略的真实水平。本文还有配套的精品资源点击获取