简介这是一份基于深度强化学习DRL的时间序列预测项目源码包面向想动手实践DRL的算法工程师、研究者和高年级学生。项目以正弦信号为预测对象完整演示了强化学习环境定义、智能体交互与策略训练的全流程可用于理解DQN等经典模型在连续值预测中的应用。全包共35个文件压缩后约760KB核心代码由10个Python脚本组成涵盖智能体、环境模拟等主要模块配套的yml文件描述依赖环境json和pickle分别存储实验配置与序列数据README提供项目说明与使用指引。目前已有525人学习下载属于轻量级且便于入门的实践资源。借助该包读者能快速掌握DRL在时间序列预测中的完整工程实现包括数据采样器、经验回放、模型训练循环等关键写法并可通过具体案例体会动态策略在非平稳时间序列上的优势。1. 强化学习预测时间序列先按下对“预测准”的执念看看它到底在解决什么问题如果你第一次打开deep-RL-time-series.zip这套代码最容易被三件事劝退训练半天 loss 不降、同一份代码跑两次结果差很多、以及把预测结果画出来发现比 ARIMA 还难看。先别急着删库这套基于深度强化学习的时间序列方案解决的问题不是你熟悉的“拟合历史曲线”而是“在不确定环境里做序列决策”。它适合的场景有三个特征数据有可观测的外部状态、预测结果会影响后续动作、以及你需要模型具备一定的在线适应能力。换句话说传统监督学习负责“看到什么就预测什么”强化学习负责“边看边学、边预测边调整策略”。这篇文章会从问题定义开始逐步拆到网络设计、训练代码和落地坑点让你能真正把这套代码跑起来并判断值不值得投入。2. 为什么传统模型逼近极限时RL 反而能接住先把预测问题重写成马尔可夫决策过程2.1 从监督学习到决策问题变化的核心不是模型是目标函数传统时间序列预测比如 LSTM、Transformer、ARIMA无论结构多复杂目标函数都是同一个套路最小化预测值和真实值之间的误差比如 MSE 或 MAE。这个框架在数据平稳、分布不变时很好用但一旦预测结果会被拿去指导动作——比如根据流量预测调整带宽、根据库存预测决定补货量——问题就变了预测错了 5% 和错了 20%代价不是对称的。预测偏低导致缺货的损失可能比预测偏高导致积压的损失大十倍。强化学习换了一个优化目标最大化累积回报。模型不再直接输出“未来 12 步的值”而是输出一个策略——在某个状态下该采取什么动作。这时候预测变成了策略的一部分智能体内部可能有一个预测模块但对外表现是“动作”。这正是deep-RL-time-series这类项目最值得研究的点它把预测建模成一个序贯决策问题而不是一步到位的回归问题。我在实际做库存预测和带宽预测时对这种差异体会很深。传统监督学习给你一个预测值就结束了但业务方真正要的是一个决策今天备多少货、要不要扩容。这两个问题之间的缝隙就是强化学习能插进去的地方。2.2 状态、动作、奖励怎么把股价或流量预测套进 RL 框架强化学习的三个核心要素放到时间序列预测里要重新定义状态State不是原始序列本身而是能描述“当前环境处境”的特征集。比如用过去 N 个时间步的观测值、衍生指标均值、方差、动量、以及外部变量星期几、是否节假日拼接成一个向量。动作Action有两种常见设计。一种是直接输出未来值此时 action 是连续的适合 DDPG、SAC 这类连续控制算法另一种是输出“调整量”——比如预测值 基线预测 动作偏移量这更像是校正机制。奖励Reward这是整个方案里最灵活也最坑的部分。最简单的设计是-MSE也就是预测越准奖励越高。但更好的设计要加入业务约束比如“预测偏高但不超过 10% 可以接受偏低超过 5% 就重罚”这种不对称的奖励函数正是 RL 对比监督学习的核心优势。回到deep-RL-time-series这个项目它通常会把原始序列滑窗成一批状态-动作-奖励四元组存入经验回放池再用 DQN 或其变种训练。虽然名字里带 deep但核心思路没有跳出 Q-learning 的框架学习一个 Q 函数评估在某个状态下采取某个动作的长期回报。2.3 连续值输出与基于模型的预测控制有一个 DQN 落地时必须解决的矛盾Q-learning 擅长离散动作而时间序列预测的输出通常是连续值。如果不做任何处理DQN 没法直接输出“下一时刻价格是 12.5 还是 12.7”。常见的是三种做法第一种动作离散化。把预测区间切成 N 个桶每个桶代表一个值域范围模型输出的是桶的编号。这种做法实现简单、训练稳定但精度受桶大小限制。第二种混合架构。用监督学习输出一个基线预测然后用强化学习输出一个修正系数比如final_pred base_pred * (1 action)action 被约束在 [-0.1, 0.1] 之间。第三种改用连续动作算法比如 DDPG 或 SAC但这类算法对超参数更敏感训练也更慢。这三条路的选型标准很简单数据量少、想快速验证选动作离散化数据量大、算力充裕选 SAC 或 DDPG业务场景里已经有一套能用的基线预测模型选修正机制最稳。deep-RL-time-series这类项目通常会给你一个 DQN 实现和一个连续动作实现方便对比。3. 跑通 deep-RL-time-series 的最小闭环环境准备、数据切分与 DQN 训练3.1 环境依赖与目录结构先花十分钟确认版本别等报错再回头我建议先建一个干净的虚拟环境Python 版本锁在 3.9 或 3.10。深度学习框架用 PyTorch强化学习环境用 Gymnasium。这里有一个常见坑老项目用的是gym新版本已经改名为gymnasiumAPI 也有变化。如果你是第一次跑这套代码看到ImportError: cannot import name Envelope from gym这类报错八成就是版本问题。先把依赖装好conda create -n rl_ts python3.9 conda activate rl_ts pip install torch2.1.2 pip install gymnasium0.29.1 pandas numpy matplotlib提示如果你用的是 Apple Silicon 的 MacPyTorch 建议装pip install torch2.1.2的 ARM 版本跑小规模实验足够如果是 Linux 服务器且有 NVIDIA GPU把 pytorch 换成 CUDA 对应的版本训练速度会快五到十倍。deep-RL-time-series.zip解压后最需要关注的文件是train.py、env.py和model.py。env.py负责把时间序列包装成强化学习环境的交互逻辑model.py定义网络结构train.py是训练入口。建议你先不要改任何参数直接用小数据集跑通一次再去调网络结构和奖励函数。3.2 数据切分时间序列不能用随机打乱必须按时间切监督学习里随机切分训练集和测试集是常规操作但时间序列不行。用未来数据训练、过去数据测试等于作弊——模型见过答案再去考试。正确做法是严格按时间顺序切分并且要考虑数据泄漏问题。import pandas as pd import numpy as np def split_ts_data(df, train_ratio0.7, val_ratio0.15): n len(df) train_end int(n * train_ratio) val_end int(n * (train_ratio val_ratio)) train df.iloc[:train_end].copy() val df.iloc[train_end:val_end].copy() test df.iloc[val_end:].copy() # Z-score 标准化参数只从训练集统计 mean train[value].mean() std train[value].std() for split in (train, val, test): split[value_scaled] (split[value] - mean) / std return train, val, test, mean, std这段代码有两个关键点一是mean和std只从训练集计算不能包含验证集和测试集的信息二是返回了mean和std预测结束之后要反标准化还原真实数值。很多初学者在这里翻车标准化时用了全量数据导致测试集的分布信息间接渗入训练过程离线评估指标虚高上线后立刻现原形。3.3 环境类实现把滑窗数据变成 Gymnasium 交互接口强化学习训练需要的不是数据集而是一个环境。所谓环境就是一组规则你给它一个动作它返回下一个状态和奖励。在时间序列场景里环境内部维护一个滑窗每次 step 就向后滑动一步。import gymnasium as gym import numpy as np class TimeSeriesEnv(gym.Env): def __init__(self, series, window_size24, horizon1, action_bins20): super().__init__() self.series series self.window_size window_size self.horizon horizon self.action_bins action_bins # 动作空间离散化预测偏差系数从 -0.1 到 0.1 self.action_space gym.spaces.Discrete(action_bins) # 状态空间: 过去 window_size 步 当前步 self.observation_space gym.spaces.Box( low-np.inf, highnp.inf, shape(window_size 1,), dtypenp.float32 ) self.current_idx window_size def reset(self, seedNone): self.current_idx self.window_size state self.series[self.current_idx - self.window_size: self.current_idx] return state.astype(np.float32), {} def step(self, action): # 动作索引映射到偏差系数 bias (action / (self.action_bins - 1) - 0.5) * 0.2 # 基线预测最近一步的值 base_pred self.series[self.current_idx] pred base_pred * (1 bias) true_val self.series[self.current_idx self.horizon] # 不对称奖励低估惩罚更重 error pred - true_val if error 0: reward -np.abs(error) * 2.0 else: reward -np.abs(error) * 0.5 self.current_idx 1 done self.current_idx self.horizon len(self.series) - 1 next_state self.series[ self.current_idx - self.window_size: self.current_idx ] return next_state.astype(np.float32), reward, done, False, {}这个环境类是一个完整的最小实现拆开来看逻辑顺序是先从动作索引映射出偏差系数然后乘到基线预测上得到最终预测值再和真实值比对算出奖励。奖励函数的不对称设计是有意的在库存场景里预测偏低的代价更高因为缺货直接损失销售额而积压只是占用资金这个比例可以根据业务调整。done标志用来判断序列是否耗尽reset在每次 episode 开始时把指针拉回滑窗起点让模型反复学习不同周期的模式。3.4 DQN 训练主循环经验回放、目标网络和 epsilon 衰减有了环境接下来就是训练主循环。这里我直接给一套能跑通的 DQN 代码核心是经验回放池、目标网络和 epsilon 贪心策略。这三样缺一不可经验回放打破样本之间的相关性目标网络稳定 Q 值更新目标epsilon 贪心负责探索。import torch import torch.nn as nn import torch.optim as optim import random from collections import deque class DQN(nn.Module): def __init__(self, state_dim, n_actions): super().__init__() self.net nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU(), nn.Linear(128, n_actions) ) def forward(self, x): return self.net(x) def train_dqn(env, episodes500, batch_size64, gamma0.99, lr1e-3, epsilon_start1.0, epsilon_end0.05): state_dim env.observation_space.shape[0] n_actions env.action_space.n online_net DQN(state_dim, n_actions) target_net DQN(state_dim, n_actions) target_net.load_state_dict(online_net.state_dict()) target_net.eval() optimizer optim.Adam(online_net.parameters(), lrlr) replay_buffer deque(maxlen20000) epsilon epsilon_start for episode in range(episodes): state, _ env.reset() done False total_reward 0 while not done: # epsilon 贪心随机探索 vs 利用当前策略 if random.random() epsilon: action env.action_space.sample() else: with torch.no_grad(): q_values online_net(torch.FloatTensor(state).unsqueeze(0)) action q_values.argmax(dim1).item() next_state, reward, done, truncated, _ env.step(action) replay_buffer.append((state, action, reward, next_state, done)) state next_state total_reward reward if len(replay_buffer) batch_size: batch random.sample(replay_buffer, batch_size) states, actions, rewards, next_states, dones zip(*batch) states torch.FloatTensor(np.array(states)) actions torch.LongTensor(actions).unsqueeze(1) rewards torch.FloatTensor(rewards) next_states torch.FloatTensor(np.array(next_states)) dones torch.FloatTensor(dones) q_values online_net(states).gather(1, actions).squeeze(1) with torch.no_grad(): next_q_values target_net(next_states).max(1)[0] targets rewards gamma * next_q_values * (1 - dones) loss nn.MSELoss()(q_values, targets) optimizer.zero_grad() loss.backward() optimizer.step() # 每 20 个 episode 同步一次目标网络 if episode % 20 0: target_net.load_state_dict(online_net.state_dict()) epsilon max(epsilon_end, epsilon * 0.995) if episode % 50 0: print(fEpisode {episode}, Total Reward: {total_reward:.4f}, Epsilon: {epsilon:.3f}) return online_net这里有几个参数值得解释gamma0.99是折扣因子数值越大代表模型越看重长期回报。在时间序列场景里gamma 不建议设成 1.0因为未来太远的事件不确定性太高过于看重远期反而会让训练不稳定。epsilon_start1.0意味着初始阶段完全随机探索然后以 0.995 的速率衰减到故事中后期模型基本靠策略行动。经验回放池容量20000对时间序列来说略小如果你训练的序列很长建议提升到50000以上防止早期样本被过早淘汰。4. 把预测精度抠上去特征工程、奖励塑形与网络结构的三个关键改动4.1 输入特征不是越多越好滞后差分、波动率与周期项怎么选模型性能的天花板往往不取决于网络结构而取决于输入特征。直接从序列上滑窗切出来的原始数值信息密度很低。我一般会在进模型之前做四类特征拼接第一类是滞后值也就是过去 p 个时刻的观测值这是基础。第二类是差分值一阶差分能消除趋势二阶差分能捕捉加速度。第三类是滚动统计量比如过去 24 小时的均值、标准差、最大值和最小值这类特征对突变点的响应更快。第四类是时间刻度特征编码星期几、小时、是否节假日用 sin/cos 编码而不是整数编码避免「星期一1、星期二2」带来虚假的线性关系。def build_features(series, window24): df pd.DataFrame({value: series}) # 滞后特征 for lag in range(1, window 1): df[flag_{lag}] df[value].shift(lag) # 差分特征 df[diff_1] df[value].diff(1) df[diff_2] df[value].diff(2) # 滚动统计量中心化处理避免未来信息 df[roll_mean] df[value].rolling(window, min_periods1).mean().shift(1) df[roll_std] df[value].rolling(window, min_periods1).std().shift(1) # 时间周期特征 df[hour_sin] np.sin(2 * np.pi * df.index.hour / 24) df[hour_cos] np.cos(2 * np.pi * df.index.hour / 24) return df.dropna().fillna(0)这里有个细节容易忽略rolling(window).mean()计算出来的当前时刻均值包含了当前时刻本身如果直接作为特征就相当于把答案的一部分泄露给了模型。所以我加了一个.shift(1)强制让特征只使用历史信息。这个细节在监督学习里影响不大因为在 batch 训练时模型可以隐式学到泄漏模式但在强化学习里状态特征的微小改变会被 Q 函数放大导致训练不稳定。如果你发现模型离线评估很好、上线就差很多优先检查这类特征是否有前视偏差。4.2 奖励塑形的边界MSE 之外的业务化奖励函数设计奖励函数是强化学习真正的“调参玄学”。设计原则只有一条奖励函数必须和业务目标严格对齐。这句话听起来是废话但实际操作中很容易走偏。常见的一个误区是直接用 MSE 取负作为奖励。这会导致模型退化成“平庸预测机”如果序列波动大模型学会预测均值附近的值因为这样整体误差最小但业务上完全没用——你真正需要的是在拐点处有敏锐反应。比较好的设计思路是把奖励拆成三个部分精度项、方向项和稳定性项。精度项还是误差的负值方向项检查“预测方向和真实变化方向是否一致”一致性给额外奖励稳定性项惩罚相邻预测之间的剧烈抖动。三个部分的权重依靠业务目标调整如果波动太剧烈加大稳定性惩罚如果更在意拐点预测加大方向项权重。4.3 网络结构的小改动从 Q 网络到分位数回归的启发DQN 的输出层是每个动作的 Q 值本质上是一个点估计。但时间序列预测天然存在不确定性——未来不可完全预知模型应该输出分布而不是点值。这里有一个折中方案把输出层从 Q 值改为分位数网络的参数。具体做法是不要只输出一个 Q 值而是输出三个分位数10%、50%、90%。训练时用分位数回归的损失函数。class QuantileDQN(nn.Module): def __init__(self, state_dim, n_actions, quantiles(0.1, 0.5, 0.9)): super().__init__() self.net nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU() ) # 每个动作输出一组分位数 self.head nn.Linear(128, n_actions * len(quantiles)) self.quantiles quantiles def forward(self, x): features self.net(x) raw self.head(features).view(-1, len(self.quantiles), x.shape[0]) return raw def quantile_loss(pred, target, quantiles): # 分位数回归损失 errors target.unsqueeze(-1) - pred loss torch.max((quantiles - 1) * errors, quantiles * errors) return loss.mean()这样改有两个直接好处一是训练时模型被迫学习不同置信水平的预测相当于天然的集成二是推断时你可以直接输出 90% 分位数作为保守预测或者在业务允许的情况下用 50% 分位数做激进预测。我在做流量预测时常用这个方案配合置信区间画图业务方和领导的接受度会高很多——他们看到的不再是一个孤零零的数字而是一个有边界的预测区间。5. 强化学习时间序列的常见问题排查四个踩坑记录5.1 训练不收敛但 loss 也不爆炸奖励信号太稀疏现象是训练日志里Total Reward一直在某个区间震荡没有明显上升趋势但 Q loss 并没有发散。最常见的原因是奖励信号太稀疏序列很长但大部分时间步的误差差不多模型区分不了哪些动作更好。排查时先打印奖励的分布。如果绝大多数奖励集中在-0.1到0.1之间说明奖励没有区分度。解决办法是奖励缩放或奖励分解提高误差的放大倍数或者在奖励里加入方向性信息让“方向对但幅度差”和“方向错”能分开。另一种做法是缩短 episode 长度拆成多个子序列提高训练频率。5.2 指标正常但上线就翻车数据泄漏发生在状态构造阶段离线评估时 RMSE 很低一上线预测结果就是一条水平线。这个现象十有八九是特征构造里用了未来信息。最常见的是标准化用了全量数据的均值和方差或滚动统计量没有shift(1)。前者的后果是测试集的分布信息泄漏到训练过程后者的后果更隐蔽——模型直接“看到”了当前时刻的真实值DQN 学会的其实是复读机遇到真实在线数据抓瞎。解决方法是严格模拟在线预测的数据流构造一个“伪在线”验证脚本遍历时间步每步只用当前时刻之前的数据生成特征再调用模型预测。这样能发现绝大多数数据泄漏问题。5.3 经验回放池只存“过去”的样本模型适应不了分布变化时间序列的数据分布是会漂移的。DQN 的经验回放池默认均匀采样早期样本和近期样本权重相同。当序列发生概念漂移时模型会被大量过期样本拖着迟迟跟不上新趋势。常见的做法是加权采样近期样本的采样概率更高或者直接把回放池大小减小让旧样本快速被淘汰。还有一种更直观的方案每隔一段时间用最近一周的数据单独微调模型然后再合并回回放池训练。这个方案牺牲了短期稳定性但在金融、流量这类高度非平稳的场景里应变速度比稳定更重要。5.4 状态空间和动作空间范围差异过大Q 值更新振荡如果状态特征是原始价格比如几千到几万而动作空间是离散索引0 到 19两者量级差异悬殊神经网络要同时拟合这两个量级的输入输出收敛会很吃力。一种做法是对状态做标准化另一种是动作不从原始值映射而是改为“比率动作”比如预测值是当前值 * (1 bias)这样动作实际影响的是一个比率而不是绝对值量级差异自然消失。deep-RL-time-series的原始代码里如果直接用原始价格做状态大概率会遇到这个问题我建议优先检查这个点。6. 从训练到可信三步验证法和置信区间可视化训练完成只代表模型拟合了历史不代表它能可靠预测未来。我的验证习惯分三步走。第一步是滚动回测把测试集按时间切成 N 段模型逐段预测每一段预测完成后将真实值加入历史重新训练或更新模拟真实在线更新的节奏。这种回测比一次性预测更能暴露模型的适应能力。第二步是基准对比至少要跑三个基线——naive 预测用最近一期值、ARIMA 或 ETS、以及监督学习 LSTM。如果强化学习模型的累计误差没有显著优于最差的基线基本上可以判断方向错了。第三步是置信区间检查用 90% 分位数预测的覆盖率来评估不确定性估计是否合理。import matplotlib.pyplot as plt def plot_prediction_with_interval(true, pred_median, pred_low, pred_high): fig, ax plt.subplots(figsize(14, 5)) x range(len(true)) ax.plot(x, true, labelTrue, colorblack, linewidth1.5) ax.plot(x, pred_median, labelPred (50%), colorblue, linewidth1.2) ax.fill_between(x, pred_low, pred_high, colorblue, alpha0.2, labelPred interval (10%-90%)) ax.legend() ax.set_xlabel(Time step) ax.set_ylabel(Value) plt.tight_layout() plt.show()覆盖率检验的方法统计真实值落在分位数区间内的比例如果 80% 的区间实际覆盖率只有 50%说明模型对不确定性估计过于乐观这时候要回查奖励函数设计——很可能是追求精度的同时压制了输出方差模型变保守了。最终建议落到一个习惯上把验证代码和训练代码放进同一个 pipeline每次修改奖励函数、网络结构或特征自动跑一遍回测和覆盖率检验用数据说话而不是靠感觉调参。这就是我对这套强化学习时间序列方案最核心的判断依据也希望这些经验能帮你少走一些弯路祝顺利。提示训练阶段的超参数调优先放在一边先确认数据泄漏和奖励对齐有没有做好这两个点出错时任何超参数调整都是在错误的地基上盖楼。本文还有配套的精品资源点击获取 
长河君 · 2026/10/5 13:34:21 · 阅读 1.2k 深度强化学习时间序列预测实战：从DQN到业务落地
简介这是一份基于深度强化学习DRL的时间序列预测项目源码包面向想动手实践DRL的算法工程师、研究者和高年级学生。项目以正弦信号为预测对象完整演示了强化学习环境定义、智能体交互与策略训练的全流程可用于理解DQN等经典模型在连续值预测中的应用。全包共35个文件压缩后约760KB核心代码由10个Python脚本组成涵盖智能体、环境模拟等主要模块配套的yml文件描述依赖环境json和pickle分别存储实验配置与序列数据README提供项目说明与使用指引。目前已有525人学习下载属于轻量级且便于入门的实践资源。借助该包读者能快速掌握DRL在时间序列预测中的完整工程实现包括数据采样器、经验回放、模型训练循环等关键写法并可通过具体案例体会动态策略在非平稳时间序列上的优势。1. 强化学习预测时间序列先按下对“预测准”的执念看看它到底在解决什么问题如果你第一次打开deep-RL-time-series.zip这套代码最容易被三件事劝退训练半天 loss 不降、同一份代码跑两次结果差很多、以及把预测结果画出来发现比 ARIMA 还难看。先别急着删库这套基于深度强化学习的时间序列方案解决的问题不是你熟悉的“拟合历史曲线”而是“在不确定环境里做序列决策”。它适合的场景有三个特征数据有可观测的外部状态、预测结果会影响后续动作、以及你需要模型具备一定的在线适应能力。换句话说传统监督学习负责“看到什么就预测什么”强化学习负责“边看边学、边预测边调整策略”。这篇文章会从问题定义开始逐步拆到网络设计、训练代码和落地坑点让你能真正把这套代码跑起来并判断值不值得投入。2. 为什么传统模型逼近极限时RL 反而能接住先把预测问题重写成马尔可夫决策过程2.1 从监督学习到决策问题变化的核心不是模型是目标函数传统时间序列预测比如 LSTM、Transformer、ARIMA无论结构多复杂目标函数都是同一个套路最小化预测值和真实值之间的误差比如 MSE 或 MAE。这个框架在数据平稳、分布不变时很好用但一旦预测结果会被拿去指导动作——比如根据流量预测调整带宽、根据库存预测决定补货量——问题就变了预测错了 5% 和错了 20%代价不是对称的。预测偏低导致缺货的损失可能比预测偏高导致积压的损失大十倍。强化学习换了一个优化目标最大化累积回报。模型不再直接输出“未来 12 步的值”而是输出一个策略——在某个状态下该采取什么动作。这时候预测变成了策略的一部分智能体内部可能有一个预测模块但对外表现是“动作”。这正是deep-RL-time-series这类项目最值得研究的点它把预测建模成一个序贯决策问题而不是一步到位的回归问题。我在实际做库存预测和带宽预测时对这种差异体会很深。传统监督学习给你一个预测值就结束了但业务方真正要的是一个决策今天备多少货、要不要扩容。这两个问题之间的缝隙就是强化学习能插进去的地方。2.2 状态、动作、奖励怎么把股价或流量预测套进 RL 框架强化学习的三个核心要素放到时间序列预测里要重新定义状态State不是原始序列本身而是能描述“当前环境处境”的特征集。比如用过去 N 个时间步的观测值、衍生指标均值、方差、动量、以及外部变量星期几、是否节假日拼接成一个向量。动作Action有两种常见设计。一种是直接输出未来值此时 action 是连续的适合 DDPG、SAC 这类连续控制算法另一种是输出“调整量”——比如预测值 基线预测 动作偏移量这更像是校正机制。奖励Reward这是整个方案里最灵活也最坑的部分。最简单的设计是-MSE也就是预测越准奖励越高。但更好的设计要加入业务约束比如“预测偏高但不超过 10% 可以接受偏低超过 5% 就重罚”这种不对称的奖励函数正是 RL 对比监督学习的核心优势。回到deep-RL-time-series这个项目它通常会把原始序列滑窗成一批状态-动作-奖励四元组存入经验回放池再用 DQN 或其变种训练。虽然名字里带 deep但核心思路没有跳出 Q-learning 的框架学习一个 Q 函数评估在某个状态下采取某个动作的长期回报。2.3 连续值输出与基于模型的预测控制有一个 DQN 落地时必须解决的矛盾Q-learning 擅长离散动作而时间序列预测的输出通常是连续值。如果不做任何处理DQN 没法直接输出“下一时刻价格是 12.5 还是 12.7”。常见的是三种做法第一种动作离散化。把预测区间切成 N 个桶每个桶代表一个值域范围模型输出的是桶的编号。这种做法实现简单、训练稳定但精度受桶大小限制。第二种混合架构。用监督学习输出一个基线预测然后用强化学习输出一个修正系数比如final_pred base_pred * (1 action)action 被约束在 [-0.1, 0.1] 之间。第三种改用连续动作算法比如 DDPG 或 SAC但这类算法对超参数更敏感训练也更慢。这三条路的选型标准很简单数据量少、想快速验证选动作离散化数据量大、算力充裕选 SAC 或 DDPG业务场景里已经有一套能用的基线预测模型选修正机制最稳。deep-RL-time-series这类项目通常会给你一个 DQN 实现和一个连续动作实现方便对比。3. 跑通 deep-RL-time-series 的最小闭环环境准备、数据切分与 DQN 训练3.1 环境依赖与目录结构先花十分钟确认版本别等报错再回头我建议先建一个干净的虚拟环境Python 版本锁在 3.9 或 3.10。深度学习框架用 PyTorch强化学习环境用 Gymnasium。这里有一个常见坑老项目用的是gym新版本已经改名为gymnasiumAPI 也有变化。如果你是第一次跑这套代码看到ImportError: cannot import name Envelope from gym这类报错八成就是版本问题。先把依赖装好conda create -n rl_ts python3.9 conda activate rl_ts pip install torch2.1.2 pip install gymnasium0.29.1 pandas numpy matplotlib提示如果你用的是 Apple Silicon 的 MacPyTorch 建议装pip install torch2.1.2的 ARM 版本跑小规模实验足够如果是 Linux 服务器且有 NVIDIA GPU把 pytorch 换成 CUDA 对应的版本训练速度会快五到十倍。deep-RL-time-series.zip解压后最需要关注的文件是train.py、env.py和model.py。env.py负责把时间序列包装成强化学习环境的交互逻辑model.py定义网络结构train.py是训练入口。建议你先不要改任何参数直接用小数据集跑通一次再去调网络结构和奖励函数。3.2 数据切分时间序列不能用随机打乱必须按时间切监督学习里随机切分训练集和测试集是常规操作但时间序列不行。用未来数据训练、过去数据测试等于作弊——模型见过答案再去考试。正确做法是严格按时间顺序切分并且要考虑数据泄漏问题。import pandas as pd import numpy as np def split_ts_data(df, train_ratio0.7, val_ratio0.15): n len(df) train_end int(n * train_ratio) val_end int(n * (train_ratio val_ratio)) train df.iloc[:train_end].copy() val df.iloc[train_end:val_end].copy() test df.iloc[val_end:].copy() # Z-score 标准化参数只从训练集统计 mean train[value].mean() std train[value].std() for split in (train, val, test): split[value_scaled] (split[value] - mean) / std return train, val, test, mean, std这段代码有两个关键点一是mean和std只从训练集计算不能包含验证集和测试集的信息二是返回了mean和std预测结束之后要反标准化还原真实数值。很多初学者在这里翻车标准化时用了全量数据导致测试集的分布信息间接渗入训练过程离线评估指标虚高上线后立刻现原形。3.3 环境类实现把滑窗数据变成 Gymnasium 交互接口强化学习训练需要的不是数据集而是一个环境。所谓环境就是一组规则你给它一个动作它返回下一个状态和奖励。在时间序列场景里环境内部维护一个滑窗每次 step 就向后滑动一步。import gymnasium as gym import numpy as np class TimeSeriesEnv(gym.Env): def __init__(self, series, window_size24, horizon1, action_bins20): super().__init__() self.series series self.window_size window_size self.horizon horizon self.action_bins action_bins # 动作空间离散化预测偏差系数从 -0.1 到 0.1 self.action_space gym.spaces.Discrete(action_bins) # 状态空间: 过去 window_size 步 当前步 self.observation_space gym.spaces.Box( low-np.inf, highnp.inf, shape(window_size 1,), dtypenp.float32 ) self.current_idx window_size def reset(self, seedNone): self.current_idx self.window_size state self.series[self.current_idx - self.window_size: self.current_idx] return state.astype(np.float32), {} def step(self, action): # 动作索引映射到偏差系数 bias (action / (self.action_bins - 1) - 0.5) * 0.2 # 基线预测最近一步的值 base_pred self.series[self.current_idx] pred base_pred * (1 bias) true_val self.series[self.current_idx self.horizon] # 不对称奖励低估惩罚更重 error pred - true_val if error 0: reward -np.abs(error) * 2.0 else: reward -np.abs(error) * 0.5 self.current_idx 1 done self.current_idx self.horizon len(self.series) - 1 next_state self.series[ self.current_idx - self.window_size: self.current_idx ] return next_state.astype(np.float32), reward, done, False, {}这个环境类是一个完整的最小实现拆开来看逻辑顺序是先从动作索引映射出偏差系数然后乘到基线预测上得到最终预测值再和真实值比对算出奖励。奖励函数的不对称设计是有意的在库存场景里预测偏低的代价更高因为缺货直接损失销售额而积压只是占用资金这个比例可以根据业务调整。done标志用来判断序列是否耗尽reset在每次 episode 开始时把指针拉回滑窗起点让模型反复学习不同周期的模式。3.4 DQN 训练主循环经验回放、目标网络和 epsilon 衰减有了环境接下来就是训练主循环。这里我直接给一套能跑通的 DQN 代码核心是经验回放池、目标网络和 epsilon 贪心策略。这三样缺一不可经验回放打破样本之间的相关性目标网络稳定 Q 值更新目标epsilon 贪心负责探索。import torch import torch.nn as nn import torch.optim as optim import random from collections import deque class DQN(nn.Module): def __init__(self, state_dim, n_actions): super().__init__() self.net nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU(), nn.Linear(128, n_actions) ) def forward(self, x): return self.net(x) def train_dqn(env, episodes500, batch_size64, gamma0.99, lr1e-3, epsilon_start1.0, epsilon_end0.05): state_dim env.observation_space.shape[0] n_actions env.action_space.n online_net DQN(state_dim, n_actions) target_net DQN(state_dim, n_actions) target_net.load_state_dict(online_net.state_dict()) target_net.eval() optimizer optim.Adam(online_net.parameters(), lrlr) replay_buffer deque(maxlen20000) epsilon epsilon_start for episode in range(episodes): state, _ env.reset() done False total_reward 0 while not done: # epsilon 贪心随机探索 vs 利用当前策略 if random.random() epsilon: action env.action_space.sample() else: with torch.no_grad(): q_values online_net(torch.FloatTensor(state).unsqueeze(0)) action q_values.argmax(dim1).item() next_state, reward, done, truncated, _ env.step(action) replay_buffer.append((state, action, reward, next_state, done)) state next_state total_reward reward if len(replay_buffer) batch_size: batch random.sample(replay_buffer, batch_size) states, actions, rewards, next_states, dones zip(*batch) states torch.FloatTensor(np.array(states)) actions torch.LongTensor(actions).unsqueeze(1) rewards torch.FloatTensor(rewards) next_states torch.FloatTensor(np.array(next_states)) dones torch.FloatTensor(dones) q_values online_net(states).gather(1, actions).squeeze(1) with torch.no_grad(): next_q_values target_net(next_states).max(1)[0] targets rewards gamma * next_q_values * (1 - dones) loss nn.MSELoss()(q_values, targets) optimizer.zero_grad() loss.backward() optimizer.step() # 每 20 个 episode 同步一次目标网络 if episode % 20 0: target_net.load_state_dict(online_net.state_dict()) epsilon max(epsilon_end, epsilon * 0.995) if episode % 50 0: print(fEpisode {episode}, Total Reward: {total_reward:.4f}, Epsilon: {epsilon:.3f}) return online_net这里有几个参数值得解释gamma0.99是折扣因子数值越大代表模型越看重长期回报。在时间序列场景里gamma 不建议设成 1.0因为未来太远的事件不确定性太高过于看重远期反而会让训练不稳定。epsilon_start1.0意味着初始阶段完全随机探索然后以 0.995 的速率衰减到故事中后期模型基本靠策略行动。经验回放池容量20000对时间序列来说略小如果你训练的序列很长建议提升到50000以上防止早期样本被过早淘汰。4. 把预测精度抠上去特征工程、奖励塑形与网络结构的三个关键改动4.1 输入特征不是越多越好滞后差分、波动率与周期项怎么选模型性能的天花板往往不取决于网络结构而取决于输入特征。直接从序列上滑窗切出来的原始数值信息密度很低。我一般会在进模型之前做四类特征拼接第一类是滞后值也就是过去 p 个时刻的观测值这是基础。第二类是差分值一阶差分能消除趋势二阶差分能捕捉加速度。第三类是滚动统计量比如过去 24 小时的均值、标准差、最大值和最小值这类特征对突变点的响应更快。第四类是时间刻度特征编码星期几、小时、是否节假日用 sin/cos 编码而不是整数编码避免「星期一1、星期二2」带来虚假的线性关系。def build_features(series, window24): df pd.DataFrame({value: series}) # 滞后特征 for lag in range(1, window 1): df[flag_{lag}] df[value].shift(lag) # 差分特征 df[diff_1] df[value].diff(1) df[diff_2] df[value].diff(2) # 滚动统计量中心化处理避免未来信息 df[roll_mean] df[value].rolling(window, min_periods1).mean().shift(1) df[roll_std] df[value].rolling(window, min_periods1).std().shift(1) # 时间周期特征 df[hour_sin] np.sin(2 * np.pi * df.index.hour / 24) df[hour_cos] np.cos(2 * np.pi * df.index.hour / 24) return df.dropna().fillna(0)这里有个细节容易忽略rolling(window).mean()计算出来的当前时刻均值包含了当前时刻本身如果直接作为特征就相当于把答案的一部分泄露给了模型。所以我加了一个.shift(1)强制让特征只使用历史信息。这个细节在监督学习里影响不大因为在 batch 训练时模型可以隐式学到泄漏模式但在强化学习里状态特征的微小改变会被 Q 函数放大导致训练不稳定。如果你发现模型离线评估很好、上线就差很多优先检查这类特征是否有前视偏差。4.2 奖励塑形的边界MSE 之外的业务化奖励函数设计奖励函数是强化学习真正的“调参玄学”。设计原则只有一条奖励函数必须和业务目标严格对齐。这句话听起来是废话但实际操作中很容易走偏。常见的一个误区是直接用 MSE 取负作为奖励。这会导致模型退化成“平庸预测机”如果序列波动大模型学会预测均值附近的值因为这样整体误差最小但业务上完全没用——你真正需要的是在拐点处有敏锐反应。比较好的设计思路是把奖励拆成三个部分精度项、方向项和稳定性项。精度项还是误差的负值方向项检查“预测方向和真实变化方向是否一致”一致性给额外奖励稳定性项惩罚相邻预测之间的剧烈抖动。三个部分的权重依靠业务目标调整如果波动太剧烈加大稳定性惩罚如果更在意拐点预测加大方向项权重。4.3 网络结构的小改动从 Q 网络到分位数回归的启发DQN 的输出层是每个动作的 Q 值本质上是一个点估计。但时间序列预测天然存在不确定性——未来不可完全预知模型应该输出分布而不是点值。这里有一个折中方案把输出层从 Q 值改为分位数网络的参数。具体做法是不要只输出一个 Q 值而是输出三个分位数10%、50%、90%。训练时用分位数回归的损失函数。class QuantileDQN(nn.Module): def __init__(self, state_dim, n_actions, quantiles(0.1, 0.5, 0.9)): super().__init__() self.net nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU() ) # 每个动作输出一组分位数 self.head nn.Linear(128, n_actions * len(quantiles)) self.quantiles quantiles def forward(self, x): features self.net(x) raw self.head(features).view(-1, len(self.quantiles), x.shape[0]) return raw def quantile_loss(pred, target, quantiles): # 分位数回归损失 errors target.unsqueeze(-1) - pred loss torch.max((quantiles - 1) * errors, quantiles * errors) return loss.mean()这样改有两个直接好处一是训练时模型被迫学习不同置信水平的预测相当于天然的集成二是推断时你可以直接输出 90% 分位数作为保守预测或者在业务允许的情况下用 50% 分位数做激进预测。我在做流量预测时常用这个方案配合置信区间画图业务方和领导的接受度会高很多——他们看到的不再是一个孤零零的数字而是一个有边界的预测区间。5. 强化学习时间序列的常见问题排查四个踩坑记录5.1 训练不收敛但 loss 也不爆炸奖励信号太稀疏现象是训练日志里Total Reward一直在某个区间震荡没有明显上升趋势但 Q loss 并没有发散。最常见的原因是奖励信号太稀疏序列很长但大部分时间步的误差差不多模型区分不了哪些动作更好。排查时先打印奖励的分布。如果绝大多数奖励集中在-0.1到0.1之间说明奖励没有区分度。解决办法是奖励缩放或奖励分解提高误差的放大倍数或者在奖励里加入方向性信息让“方向对但幅度差”和“方向错”能分开。另一种做法是缩短 episode 长度拆成多个子序列提高训练频率。5.2 指标正常但上线就翻车数据泄漏发生在状态构造阶段离线评估时 RMSE 很低一上线预测结果就是一条水平线。这个现象十有八九是特征构造里用了未来信息。最常见的是标准化用了全量数据的均值和方差或滚动统计量没有shift(1)。前者的后果是测试集的分布信息泄漏到训练过程后者的后果更隐蔽——模型直接“看到”了当前时刻的真实值DQN 学会的其实是复读机遇到真实在线数据抓瞎。解决方法是严格模拟在线预测的数据流构造一个“伪在线”验证脚本遍历时间步每步只用当前时刻之前的数据生成特征再调用模型预测。这样能发现绝大多数数据泄漏问题。5.3 经验回放池只存“过去”的样本模型适应不了分布变化时间序列的数据分布是会漂移的。DQN 的经验回放池默认均匀采样早期样本和近期样本权重相同。当序列发生概念漂移时模型会被大量过期样本拖着迟迟跟不上新趋势。常见的做法是加权采样近期样本的采样概率更高或者直接把回放池大小减小让旧样本快速被淘汰。还有一种更直观的方案每隔一段时间用最近一周的数据单独微调模型然后再合并回回放池训练。这个方案牺牲了短期稳定性但在金融、流量这类高度非平稳的场景里应变速度比稳定更重要。5.4 状态空间和动作空间范围差异过大Q 值更新振荡如果状态特征是原始价格比如几千到几万而动作空间是离散索引0 到 19两者量级差异悬殊神经网络要同时拟合这两个量级的输入输出收敛会很吃力。一种做法是对状态做标准化另一种是动作不从原始值映射而是改为“比率动作”比如预测值是当前值 * (1 bias)这样动作实际影响的是一个比率而不是绝对值量级差异自然消失。deep-RL-time-series的原始代码里如果直接用原始价格做状态大概率会遇到这个问题我建议优先检查这个点。6. 从训练到可信三步验证法和置信区间可视化训练完成只代表模型拟合了历史不代表它能可靠预测未来。我的验证习惯分三步走。第一步是滚动回测把测试集按时间切成 N 段模型逐段预测每一段预测完成后将真实值加入历史重新训练或更新模拟真实在线更新的节奏。这种回测比一次性预测更能暴露模型的适应能力。第二步是基准对比至少要跑三个基线——naive 预测用最近一期值、ARIMA 或 ETS、以及监督学习 LSTM。如果强化学习模型的累计误差没有显著优于最差的基线基本上可以判断方向错了。第三步是置信区间检查用 90% 分位数预测的覆盖率来评估不确定性估计是否合理。import matplotlib.pyplot as plt def plot_prediction_with_interval(true, pred_median, pred_low, pred_high): fig, ax plt.subplots(figsize(14, 5)) x range(len(true)) ax.plot(x, true, labelTrue, colorblack, linewidth1.5) ax.plot(x, pred_median, labelPred (50%), colorblue, linewidth1.2) ax.fill_between(x, pred_low, pred_high, colorblue, alpha0.2, labelPred interval (10%-90%)) ax.legend() ax.set_xlabel(Time step) ax.set_ylabel(Value) plt.tight_layout() plt.show()覆盖率检验的方法统计真实值落在分位数区间内的比例如果 80% 的区间实际覆盖率只有 50%说明模型对不确定性估计过于乐观这时候要回查奖励函数设计——很可能是追求精度的同时压制了输出方差模型变保守了。最终建议落到一个习惯上把验证代码和训练代码放进同一个 pipeline每次修改奖励函数、网络结构或特征自动跑一遍回测和覆盖率检验用数据说话而不是靠感觉调参。这就是我对这套强化学习时间序列方案最核心的判断依据也希望这些经验能帮你少走一些弯路祝顺利。提示训练阶段的超参数调优先放在一边先确认数据泄漏和奖励对齐有没有做好这两个点出错时任何超参数调整都是在错误的地基上盖楼。本文还有配套的精品资源点击获取