简介在时间序列预测任务中引入深度强化学习是近年来的研究热点之一。该zip包以DRL为工具解决序列预测问题面向具备Python和机器学习基础、希望进阶强化学习的开发者。项目围绕DQN等模型展开将预测转化为智能体在环境中的决策过程适用于金融、流量、气象等动态非平稳数据的场景。与ARIMA等传统方法相比DRL能捕捉更复杂的非线性动态特征。资源共35个文件以Python源码为核心涵盖智能体、模拟器、采样器、可视化等模块另有json/pickle存放配置与序列数据yml文件用于重建依赖环境压缩后仅760KB轻量易用。目录结构清晰src、data、examples分层例如src下包含agents.py、simulators.py、ksp_sampler.py等实现data提供SinSamplerDB、PairSamplerDB等多种采样数据便于对照代码理解状态、动作、奖励设计及模型评估。已有525人学习下载适合希望从理论快速转向DRL时序预测实操的读者可直接参考其工程组织方式并迁移到自己的实验或项目中。1. deep-RL-time-series一个把时间序列预测改写成强化学习问题的开源包把时间序列预测硬生生改写成强化学习问题之后预测本身就不再是拟合下一个点而是变成智能体为了拿到最大累计奖励得学会怎么一步步把误差压下去。这套 deep-RL-time-series 代码就是干这个的它用 DQN 一类的深度强化学习算法在正弦序列基准上做预测并把数据生成sampler、环境仿真emulator、智能体agents、训练入口main / ksp_main全部拆成独立模块。你换一个数据采样器就能把同一套训练流程迁移到金融时序、气象数据这类业务场景上。它适合想从示例代码入手、真正跑通 DRL 做时序预测全流程的开发者也适合被 ARIMA、LSTM 折腾过、想换个思路看看强化学习怎么处理非平稳序列的人。2. 数据生成层SinSamplerDB 与 PairSamplerDB 的序列构造逻辑2.1 SinSamplerDB正弦序列怎么生成、参数怎么调这个项目把数据生成单独抽了一个 sampler 层而不是在训练循环里临时拼数据。data目录下能看到SinSamplerDB、PairSamplerDB、KSPSamplerDB三个数据源其中SinSamplerDB是最基础的它负责生成带噪声的正弦序列用来做 DRL 预测算法的第一个验证基准。正弦序列的好处是形状确定、周期已知模型学到的是这个窗口后续怎么走而不是这个数据集背后有什么隐藏规律——这正好用来验证强化学习环境搭建得对不对。我一般会先小批量生成一段正弦数据肉眼确认数据形状再进训练。下面这段代码演示了SinSamplerDB最常见的实例化方式import numpy as np import matplotlib.pyplot as plt # 假设 sampler 目录下 SinSamplerDB 接收 amplitude / frequency / phase / noise 四个核心参数 from sampler import SinSamplerDB sampler SinSamplerDB( amplitude1.0, # 正弦振幅决定序列波动幅度 frequency0.05, # 角频率0.05 约等于 125 个点一个完整周期 phase0.0, # 初始相位 noise0.05, # 高斯噪声标准差噪声太大会淹没周期信号 seq_len1000 # 生成序列总长度 ) seq sampler.sample() t np.arange(len(seq)) plt.plot(t, seq, linewidth0.8) plt.title(SinSamplerDB sample: amplitude1.0, freq0.05, noise0.05) plt.savefig(sin_check.png, dpi120)代码逻辑并不复杂sample()返回一个一维np.ndarray长度由seq_len控制。注意frequency0.05这个值的含义它指的是每个时间步的弧度增量所以完整周期约等于2π / 0.05 ≈ 125个点。如果你的业务数据周期是 24 小时或 7 天就把这个参数换成2π / 周期长度。noise参数建议从 0.01 起步加到 0.1 以上时 DQN 类算法会明显变难收敛这也是后续调参时第一个要动的旋钮。2.2 PairSamplerDB 与 randjump配对采样和分布突变的用意data目录里还有两组看起来很怪的文件名concat_half_base_A、concat_half_base_B以及randjump_100,1(10, 30)[]_A、randjump_100,1(10, 30)[]_B。这两个命名的信息量很大。concat_half_base的意思是把两段不同参数生成的正弦序列各取一半首尾拼接制造分布切换randjump_100,1(10, 30)[]则是随机跳跃场景——序列每走 10 到 30 个点整体向上或向下跳变 1 个单位总长大约 100 个点。这两个数据结构对应的不是更好看的正弦波而是刻意制造非平稳性。传统时序模型在这种数据上最容易翻车因为窗口内的统计特征突然失效。强化学习智能体理论上可以通过奖励信号学会检测到突变后降低预测置信度。from sampler import PairSamplerDB pair_sampler PairSamplerDB( base_aconcat_half_base_A, # 前半段用 A 参数生成 base_bconcat_half_base_B, # 后半段用 B 参数生成 jump_every(10, 30), # 每隔 10~30 个点随机跳跃一次 jump_magnitude1.0, # 跳跃幅度 seq_len500 ) seq_a, seq_b pair_sampler.sample_pair() # 实际训练时通常以 (seq_a, seq_b) 配对作为 state 和 target这里有一个关键点PairSamplerDB返回的是配对数据A 序列和 B 序列在时间轴上必须严格对齐。项目文件名里_A、_B的后缀也是在强调这种配对关系。如果你在自己的数据上复刻这套逻辑A/B 对齐一旦错位训练出来的智能体行为会非常诡异——误差曲线正常下降但预测曲线完全错拍。2.3 KSP 场景业务化环境在测什么ksp_sampler.py和ksp_emulator.py组成了项目里的 KSP 场景。KSP 具体全称 README 里应该有说明从代码结构看它是把正弦基准推广到更接近业务的环境采样器换成KSPSamplerDB仿真器换成ksp_emulator训练入口换成ksp_main.py。我理解它的目的是验证同一套 DRL 预测框架在奖励函数变化后还能不能学出来。在这个场景里预测误差的衡量方式通常不是简单的 MSE而可能加入阈值判定——误差在一定范围内给正奖励超出范围给负奖励甚至零奖励。这种非光滑奖励对 DQN 来说是个考验Q 值估计方差会变大训练时要重点盯 reward 曲线的稳定性。如果你要迁移到金融时序预测KSP 场景反而是更值得参考的模板因为业务上通常也只关心预测是否落在某个可接受区间内。3. 从预测到 MDP状态、动作、奖励三件套怎么定义3.1 状态与动作空间为什么不能直接输出连续值把时间序列预测构造成强化学习问题第一步是定义状态、动作、奖励。常见做法是状态用最近 L 个历史观测值组成的窗口向量动作是智能体对下一步值的预测奖励由预测误差计算。状态窗口长度 L 直接决定环境是部分可观测还是近似可观测——L 太小智能体看不到周期上下文预测全靠猜L 太大状态维度膨胀DQN 的 Q 网络参数变多训练变慢。动作空间的处理几乎是所有 DRL 时序预测项目的第一道坎。连续动作空间直接套 DQN 会面临输出层无法枚举 Q 值的问题项目里 agents 模块的思路大概率是走离散化路线。一种常见实现是把预测范围分成若干等距区间每个区间对应一个离散动作智能体的任务变成选择预测值落在哪个桶里。桶数太少预测精度受限于量化误差桶数太多动作空间变大探索效率下降。import gym import numpy as np from gym import spaces class SinEnv(gym.Env): 把 SinSamplerDB 包装成 gym 环境状态为历史窗口动作为预测值离散桶。 def __init__(self, sampler, window10, num_bins20, reward_scale10.0): super().__init__() self.sampler sampler self.window window self.num_bins num_bins self.reward_scale reward_scale self.series sampler.sample() # 一次性生成整段序列 self.max_pred float(np.max(self.series)) # 预测值上限 self.min_pred float(np.min(self.series)) # 预测值下限 # 状态最近 window 个点的观测值动作num_bins 个离散预测桶 self.observation_space spaces.Box(low-np.inf, highnp.inf, shape(window,)) self.action_space spaces.Discrete(num_bins) self.t window def _bin_to_value(self, action: int) - float: 把离散动作映射回连续预测值线性映射到序列取值区间。 frac (action 0.5) / self.num_bins return self.min_pred frac * (self.max_pred - self.min_pred) def step(self, action): true_val self.series[self.t] # 当前时间步的真实值 pred_val self._bin_to_value(action) # 智能体给出的预测 err abs(pred_val - true_val) reward -err * self.reward_scale # 误差越小奖励越高负奖励缩放 self.t 1 done self.t len(self.series) - 1 state self.series[self.t - self.window:self.t] return state.astype(np.float32), float(reward), done, {true: true_val, pred: pred_val} def reset(self): self.t self.window return self.series[self.t - self.window:self.t].astype(np.float32)这段代码展示了环境封装的核心逻辑step()里每次只往前进一个时间步状态窗口始终落后于待预测点。注意我把奖励乘了reward_scale10因为原始误差落在 0~1 区间如果不缩放DQN 的 Q 值回归目标会非常小MSE 损失直接被神经网络末层的梯度噪声淹没。_bin_to_value做了线性映射num_bins设为 20 时每个桶覆盖约 5% 的取值跨度算是一个兼顾精度和探索效率的默认值。3.2 奖励函数误差的绝对值、平方还是分桶奖励函数设计是 DRL 时序预测最玄学的部分。用绝对误差-|pred - true|的话奖励始终是负的且数值很小智能体学到的是尽量少出错但梯度信号太温和。用平方误差-(pred - true)^2大误差会被放大惩罚能更快纠正离谱预测但小误差区域的梯度更平缓。用分桶奖励——误差小于阈值得1大于阈值得-1——训练稳定性最好却丢失了误差的连续信息。我实践的结论是训练前期用分桶奖励让智能体先学会大致落在正确区域训练后期切换成平方误差做精修。项目里 KSP 场景的奖励设定应该就是这个思路的变体。换奖励函数时有一个坑必须注意改了奖励函数之前的经验回放缓冲区里的旧样本全部作废必须清空重来否则新旧奖励尺度不一致会让 Q 值震荡。3.3 emulator、sampler、agent 三者怎么拼成闭环从文件结构看simulators.py负责仿真器emulator.py是环境主体sampler.py提供数据agents.py定义智能体。闭环关系是训练循环调用agent.act(state)得到动作传给emulator.step(action)emulator 从 sampler 生成的数据序列中取出真实值计算奖励返回新状态循环往复。这个拆分的好处是换数据只要改 sampler换预测目标只要改 emulator 的奖励计算算法本身不用动。我在跑这类项目时习惯先把闭环拆成 30 步手动验证一下用固定 seed 的智能体跑 30 步打印每一步的state、action、true、pred、reward。只要这五项能对得上才认为环境封装没问题否则后面训练出的任何曲线都没有意义。4. 跑通训练主循环main.py 与 agents.py 的关键参数和收敛判定4.1 训练主循环DQN 骨架与关键超参main.py是正弦场景的训练入口ksp_main.py是 KSP 场景的入口。两者骨架相似区别在于环境类型、奖励函数和场景配置。以正弦场景为例训练主循环的常见结构是初始化环境、初始化智能体、循环执行动作采集经验、定期从经验池采样更新网络。下面给出一个按项目结构简化的训练循环骨架import numpy as np import torch import torch.nn as nn import torch.optim as optim # 项目里 agents.py 一般包含两个核心模块 # Mapper负责粗粒度策略选择Approximator负责 Q 值细粒度拟合 from agents import Mapper, Approximator from emulator import create_sin_emulator # 内部封装 SinSamplerDB SinEnv env create_sin_emulator( window10, num_bins20, reward_scale10.0 ) replay_buffer [] # 经验回放实际项目中可替换为 deque(maxlen20000) batch_size 64 gamma 0.99 lr 1e-3 target_update_steps 1000 epsilon 1.0 epsilon_min 0.1 epsilon_decay 0.995 q_net Approximator(state_dim10, action_dim20) target_net Approximator(state_dim10, action_dim20) target_net.load_state_dict(q_net.state_dict()) optimizer optim.Adam(q_net.parameters(), lrlr) loss_fn nn.MSELoss() state env.reset() for step in range(20000): # epsilon-greedy 探索随机动作 vs 当前 Q 网络最优动作 if np.random.rand() epsilon: action env.action_space.sample() else: state_t torch.FloatTensor(state).unsqueeze(0) with torch.no_grad(): q_vals q_net(state_t) action int(torch.argmax(q_vals, dim1).item()) next_state, reward, done, info env.step(action) replay_buffer.append((state, action, reward, next_state, done)) state next_state if done: state env.reset() if len(replay_buffer) batch_size: batch [replay_buffer[i] for i in np.random.choice(len(replay_buffer), batch_size)] s_batch torch.FloatTensor([b[0] for b in batch]) a_batch torch.LongTensor([b[1] for b in batch]).unsqueeze(1) r_batch torch.FloatTensor([b[2] for b in batch]).unsqueeze(1) ns_batch torch.FloatTensor([b[3] for b in batch]) d_batch torch.FloatTensor([b[4] for b in batch]).unsqueeze(1) q_pred q_net(s_batch).gather(1, a_batch) with torch.no_grad(): q_target r_batch gamma * (1 - d_batch) * target_net(ns_batch).max(1, keepdimTrue)[0] loss loss_fn(q_pred, q_target) optimizer.zero_grad() loss.backward() optimizer.step() if step % target_update_steps 0: target_net.load_state_dict(q_net.state_dict()) epsilon max(epsilon_min, epsilon * epsilon_decay) if step % 500 0: print(fstep{step} loss{loss.item():.4f} epsilon{epsilon:.2f})这段代码有两个细节值得展开。第一q_pred用gather(1, a_batch)只取实际执行动作对应的 Q 值这是 DQN 的标准写法。第二target_net(ns_batch).max(1, keepdimTrue)[0]求的是下一状态所有动作的 Q 值最大值gamma控制未来奖励的折现程度gamma0.99意味着智能体愿意为未来积累奖励而牺牲即时奖励——但时序预测场景里每一步的预测都是独立的gamma取 0.9 甚至更低通常效果更好这一点容易被忽略。4.2 超参数表从哪里起步、往哪个方向调参数建议起始值调整方向与依据window状态窗口10~20周期约 125 点时窗口取 10~20 够用周期更长可适当加大num_bins动作桶数20桶数增加精度但探索变难超过 50 要配合更大batch_sizereward_scale10.0奖励量级应控制在 0.1~1.0 之间太小梯度消失太大训练震荡gamma0.9~0.99单步预测任务用 0.9 足够多步连续预测再用 0.99batch_size64训练不稳时加大到 128减少梯度方差target_update_steps500~2000太大导致目标长期滞后太小会陷入自举震荡epsilon_decay0.995想让探索阶段更长就调到 0.998总步数也会相应拉长gamma是我在多个 DRL 时序项目里反复验证过的点。不少教程把gamma0.99当默认值但时序预测的每个时间步误差是即时产生的不存在为了长期回报牺牲当前步的逻辑所以折现因子应该更小。这个参数调对了收敛速度肉眼可见地变快。4.3 main.py 与 ksp_main.py 的入口差异两个入口脚本最大的差异在环境与奖励函数上。main.py走正弦环境奖励是简单的负误差缩放ksp_main.py走 KSP 仿真器奖励通常带阈值判定逻辑。如果你的目标是把这套代码迁移到自己业务上先跑通main.py确认整套链路没问题再改ksp_main.py里的环境构造和奖励函数。直接上手改业务环境出了问题很难分清是算法实现错误还是环境封装错误。训练过程中需要盯着三个信号loss 下降趋势、reward 滑动均值、epsilon 衰减后的探索占比。loss 下降但 reward 不涨说明 Q 网络在拟合但策略没变好优先检查状态窗口是否包含足够预测信息reward 涨但波动巨大说明奖励尺度太大或 batch 太小优先降reward_scale或加batch_size两者都不动基本可以断定环境里存在数据泄漏或奖励恒为同一个值。5. 避坑五个会直接让训练翻车的雷区5.1 时间泄漏训练误差极低测试误差惨不忍睹现象训练过程中 reward 曲线一路向好loss 降到接近零但用最后一个时间段的数据评估时预测曲线整体错位误差比随机预测还大。原因SinSamplerDB 生成序列后状态窗口里包含了待预测点本身或邻近时间信息。concat_half_base_A这类拼接数据里前半段序列的末尾已经携带了后半段起始位置的信息如果切分窗口时把边界点也塞进状态就是开卷考试。另一个常见泄漏点是经验回放缓冲区内相邻样本时间重叠智能体直接从上一帧看到过答案里学。解决确保状态窗口右端与预测目标之间至少隔一个时间点即状态取series[t-window:t]目标永远是series[t]。评估时按时间顺序切分前 80% 训练、后 20% 测试而不是随机抽样切分。5.2 奖励尺度太小Q 值回归被梯度噪声淹没现象训练 loss 曲线不稳定前几千步剧烈震荡之后长期横盘在某个高位reward 几乎不动。原因原始误差通常在 0~1 范围内奖励-|err|的量级是 0.01~1。DQN 的目标值由即时奖励加折扣 Q 值构成奖励量级太小意味着梯度信号微弱神经网络末层的随机初始化噪声就能把它盖住。解决把奖励乘一个缩放因子让单步奖励量级落在 0.1~1.0。也可以改用分桶奖励误差小于阈值给1、否则给-1牺牲连续性换取稳定的梯度方向。修改奖励后务必清空经验回放缓冲区否则新旧奖励尺度混杂会让 Q 值目标自相矛盾。5.3 Python 3.6 缓存与依赖环境不对应现象clone 代码后直接python main.py报错SyntaxError或者 import 阶段出现奇怪的二进制错误提示cpython-36.pyc无法加载。原因__pycache__里带cpython-36.pyc后缀的编译缓存是 Python 3.6 时代生成的本机 Python 版本如果是 3.7 以上解释器不会直接使用旧缓存但某些带有from __future__语法或旧版第三方库的源码在更高版本可能会行为异常。.hypothesis目录则是 pytest 假设测试的缓存也会拖慢首次启动。解决先删掉__pycache__和.hypothesis再按env.yml重建 conda 环境# 1. 删除所有 Python 缓存目录 find . -type f -name *.pyc -delete find . -type d -name __pycache__ -exec rm -rf {} # 2. 按 env.yml 重建环境 conda env create -f env.yml # 3. 启动前重新编译所有模块确保本机版本字节码 conda activate drl-ts python -m compileall src/ emulator.py sampler.py agents.py这里有个细节env.yml里的依赖版本如果锁得比较老比如numpy1.16、gym0.17在本机新 Python 版本上可能会出现冲突。我一般会先看env.yml里锁的 Python 主版本再用conda create -n drl-ts python3.6单独建环境避免污染其他项目环境。5.4 随机种子没固定相同参数两次训练结果天差地别现象跑两次相同训练脚本一次收敛一次发散reward 曲线形状完全不同甚至最优预测值都不同。原因Python 的random、NumPy 的np.random、PyTorch 的torch.manual_seed有各自独立的随机状态只设其中一个种子其他模块仍处于随机状态。SinSamplerDB 生成数据时如果用默认随机源每次生成的噪声序列也不同。解决在入口脚本里同时固定三个种子并把数据生成器的噪声种子也固定import random import numpy as np import torch random.seed(42) np.random.seed(42) torch.manual_seed(42) # 如果 sampler 内部用 np.random 生成噪声需要先调用 np.random.seed 再实例化 sampler sampler SinSamplerDB(..., noise_seed42)5.5 PairSampler 的 A/B 样本错位训练偶发 NaN 或 loss 爆炸现象训练跑到几百步时loss 突然跳到nan或者预测曲线出现周期性尖峰每个突变点附近误差巨大。原因randjump数据里突变点位置是随机的状态窗口恰好跨越突变点时窗口内数值分布与相邻样本差异极大Q 值估计到一个异常区域梯度爆炸。PairSamplerDB的 A/B 序列长度不一致或索引错位时也会产生这种跳变。解决训练前先打印 A/B 序列长度并断言相等在环境step()里检测状态窗口是否包含突变点如果包含则跳过该样本或重置到突变点之后给梯度加裁剪torch.nn.utils.clip_grad_norm_(q_net.parameters(), 1.0)防止单步异常样本破坏整个网络。6. 迭代到自己的数据换采样器、盯可视化、调奖励尺度读代码这件事看懂训练循环只是第一步。真正决定这套框架能不能用在你自己业务上的是数据采样器怎么替换、结果怎么验证。这套项目的模块拆分已经替你想清楚了sampler.py定义数据接口emulator.py调用采样器并把序列包装成环境agents.py只关心状态空间、动作空间和奖励不关心数据来源。所以迁移到新数据源的关键就是替换 sampler 并保证接口兼容。替换采样器的三步值得讲清楚。第一步新建一个MyDataSampler类实现sample()方法返回一维np.ndarray——这是SinSamplerDB对外暴露的核心接口只要返回值是一个数组环境层就无感知。第二步把归一化逻辑放进采样器内部项目里正弦序列天然在[-1, 1]区间但业务数据往往有量纲差异金融时序的价格、气象数据的温度数值范围完全不同。我惯用的做法是在采样器里返回归一化后的序列同时把原始数据的均值方差存下来预测结束后再反归一化。第三步在create_sin_emulator的环境工厂函数里把SinSamplerDB替换成自定义采样器其余代码不动。import numpy as np class MyDataSampler: 业务数据采样器从 CSV 读入做差分与归一化输出与 SinSamplerDB 同构的序列。 def __init__(self, csv_path: str, noise_seed: int 42): raw np.loadtxt(csv_path, delimiter,, skiprows1, usecols1) # 假设第二列是目标值 diff np.diff(raw) # 差分去趋势 self.mean float(diff.mean()) self.std float(diff.std()) self.series (diff - self.mean) / self.std # z-score 归一化 self.noise_seed noise_seed def sample(self) - np.ndarray: rng np.random.default_rng(self.noise_seed) # 加一点高斯噪声模拟观测误差噪声幅度用 std 的 2% return self.series rng.normal(0.0, self.std * 0.02, sizeself.series.shape)注意这个采样器里我做了一个差分处理——这来自一个实操教训金融时序的绝对价格通常是非平稳的直接喂给 DRL 模型状态窗口里的均值漂移会干扰预测。差分以后序列变成近似平稳的收益率序列再归一化让数值范围落在[-3, 3]区间DRL 模型学起来要轻松得多。数据源换好之后验证环节比训练本身更容易踩坑。visualizer.py这个模块在项目里就是干这个的——把模型的预测序列和真实序列叠加画出来。我跑这段代码的时候会盯着三个东西第一预测曲线是否整体滞后于真实曲线如果滞后明显说明状态窗口信息不够或者动作桶数太少导致量化误差过大第二突变点附近的预测误差是否显著高于平稳段如果突变点全部崩盘说明智能体只学到了平均值的预测没有学会检测分布变化这时要把randjump数据加进训练集第三预测和真实之间的误差是否随时间累积如果误差越来越大说明把非平稳序列直接喂给了模型需要回到采样器做差分。# 项目提供的可视化入口--logdir 指向训练日志目录 python visualizer.py --logdir runs/sin_experiment_1 --savefig prediction.png关于 KSP 场景的奖励调整我的建议是先保留项目原始的负误差奖励跑通技术链路再用你的业务指标替换奖励函数。比如金融时序预测真正关心的不是绝对误差而是方向正确率和相对误差比例那就把奖励改成方向正确给 1、方向错误给 -1加相对误差小于 5% 额外给 0.5这种混合奖励才能引导智能体学到业务上真正有用的策略。这套项目我前后复现了三遍。第一遍是照着坑踩的Python 环境不干净导致跑出来的结果全是错的我还以为是算法调参问题折腾了一整天才发现是__pycache__里的旧字节码在作怪。第二遍我学乖了换环境先删缓存、固定三个种子、打印前 30 步的 state-action-reward 三元组确认环境闭环正确结果训练一次收敛。从那以后我每次跑 DRL 时序项目都强制走一遍同样的流程清理缓存、固定种子、小步验证闭环、再放长训练。这个习惯帮我免掉了至少五次无效训练希望也能帮到你。本文还有配套的精品资源点击获取