简介:这是一份基于深度强化学习(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,新版本已经改名为gymnasium,API 也有变化。如果你是第一次跑这套代码,看到ImportError: cannot import name 'Envelope' from 'gym'这类报错,八成就是版本问题。
先把依赖装好:
conda create -n rl_ts python=3.9 conda activate rl_ts pip install torch==2.1.2 pip install gymnasium==0.29.1 pandas numpy matplotlib提示:如果你用的是 Apple Silicon 的 Mac,PyTorch 建议装
pip install torch==2.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_ratio=0.7, val_ratio=0.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_size=24, horizon=1, action_bins=20): 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, high=np.inf, shape=(window_size + 1,), dtype=np.float32 ) self.current_idx = window_size def reset(self, seed=None): 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, episodes=500, batch_size=64, gamma=0.99, lr=1e-3, epsilon_start=1.0, epsilon_end=0.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(), lr=lr) replay_buffer = deque(maxlen=20000) 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(dim=1).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(f"Episode {episode}, Total Reward: {total_reward:.4f}, Epsilon: {epsilon:.3f}") return online_net这里有几个参数值得解释:gamma=0.99是折扣因子,数值越大代表模型越看重长期回报。在时间序列场景里,gamma 不建议设成 1.0,因为未来太远的事件不确定性太高,过于看重远期反而会让训练不稳定。epsilon_start=1.0意味着初始阶段完全随机探索,然后以 0.995 的速率衰减,到故事中后期模型基本靠策略行动。经验回放池容量20000对时间序列来说略小,如果你训练的序列很长,建议提升到50000以上,防止早期样本被过早淘汰。
4. 把预测精度抠上去:特征工程、奖励塑形与网络结构的三个关键改动
4.1 输入特征不是越多越好:滞后差分、波动率与周期项怎么选
模型性能的天花板往往不取决于网络结构,而取决于输入特征。直接从序列上滑窗切出来的原始数值,信息密度很低。我一般会在进模型之前做四类特征拼接:
第一类是滞后值,也就是过去 p 个时刻的观测值,这是基础。第二类是差分值,一阶差分能消除趋势,二阶差分能捕捉加速度。第三类是滚动统计量,比如过去 24 小时的均值、标准差、最大值和最小值,这类特征对突变点的响应更快。第四类是时间刻度特征,编码星期几、小时、是否节假日,用 sin/cos 编码而不是整数编码,避免「星期一=1、星期二=2」带来虚假的线性关系。
def build_features(series, window=24): df = pd.DataFrame({'value': series}) # 滞后特征 for lag in range(1, window + 1): df[f'lag_{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_periods=1).mean().shift(1) df['roll_std'] = df['value'].rolling(window, min_periods=1).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, label='True', color='black', linewidth=1.5) ax.plot(x, pred_median, label='Pred (50%)', color='blue', linewidth=1.2) ax.fill_between(x, pred_low, pred_high, color='blue', alpha=0.2, label='Pred interval (10%-90%)') ax.legend() ax.set_xlabel('Time step') ax.set_ylabel('Value') plt.tight_layout() plt.show()覆盖率检验的方法:统计真实值落在分位数区间内的比例,如果 80% 的区间实际覆盖率只有 50%,说明模型对不确定性估计过于乐观,这时候要回查奖励函数设计——很可能是追求精度的同时压制了输出方差,模型变保守了。最终建议落到一个习惯上:把验证代码和训练代码放进同一个 pipeline,每次修改奖励函数、网络结构或特征,自动跑一遍回测和覆盖率检验,用数据说话而不是靠感觉调参。这就是我对这套强化学习时间序列方案最核心的判断依据,也希望这些经验能帮你少走一些弯路,祝顺利。
提示:训练阶段的超参数调优先放在一边,先确认数据泄漏和奖励对齐有没有做好,这两个点出错时,任何超参数调整都是在错误的地基上盖楼。
本文还有配套的精品资源,点击获取