深圳优化网站排名,app 网站运营ppt,做公司 网站建设价格,wordpress标题去重深度强化学习可以将深度学习与强化学习相结合#xff1a;深度学习擅长从原始数据中学习复杂的表示#xff0c;强化学习则使代理能够通过反复试验在给定环境中学习最佳动作。通过DRL#xff0c;研究人员和投资者可以开发能够分析历史数据的模型#xff0c;理解复杂的市场动态…深度强化学习可以将深度学习与强化学习相结合深度学习擅长从原始数据中学习复杂的表示强化学习则使代理能够通过反复试验在给定环境中学习最佳动作。通过DRL研究人员和投资者可以开发能够分析历史数据的模型理解复杂的市场动态并对股票购买、销售或持有做出明智的决策。 下面我们一边写代码一边介绍这些相关的知识
数据集 importnumpyasnpimportpandasaspdimportcopyimportnumpyasnpimportchainerimportchainer.functionsasFimportchainer.linksasLfromplotlyimporttoolsfromplotly.graph_objsimport*fromplotly.offlineimportinit_notebook_mode, iplot, iplot_mplfromtqdmimporttqdm_notebookastqdminit_notebook_mode()这里主要使用使用Jupyter notebook和plotly进行可视化所以需要一些额外的设置下面开始读取数据 try:datapd.read_csv(../input/Data/Stocks/goog.us.txt)data[Date] pd.to_datetime(data[Date])datadata.set_index(Date)except (FileNotFoundError): importdatetimeimportpandas_datareaderaspdrfrompandasimportSeries, DataFrame startdatetime.datetime(2010, 1, 1)enddatetime.datetime(2017, 1, 11) datapdr.get_data_yahoo(AAPL, start, end)print(data.index.min(), data.index.max())split_indexint(len(data)/2)date_splitdata.index[split_index]traindata[:split_index]testdata[split_index:]#date_split 2016-01-01date_split2016-01-01traindata[:date_split]testdata[date_split:]print(len(data), len(train), len(test))display(data)代码从数据集中读取数据。进行测试和验证集的拆分然后使用’ display 函数代码在Jupyter笔记本中显示导入的数据。 defplot_train_test(train, test, date_split):data [Candlestick(xtrain.index, opentrain[Open], hightrain[High], lowtrain[Low], closetrain[Close], nametrain),Candlestick(xtest.index, opentest[Open], hightest[High], lowtest[Low], closetest[Close], nametest)]layout {shapes: [{x0: date_split, x1: date_split, y0: 0, y1: 1, xref: x, yref: paper, line: {color: rgb(0,0,0), width: 1}}],annotations: [{x: date_split, y: 1.0, xref: x, yref: paper, showarrow: False, xanchor: left, text: test data},{x: date_split, y: 1.0, xref: x, yref: paper, showarrow: False, xanchor: right, text: train data }]}figureFigure(datadata, layoutlayout)iplot(figure)这段代码定义了一个名为plot_train_test的函数该函数使用Python绘图库Plotly创建可视化图。基于指定的日期图表将股票数据分为训练集和测试集。输入参数包括train、test和date_split。
可视化结果如下: plot_train_test(train, test, date_split)环境
下面我们开始编写强化学习相关的内容 classEnvironment:def__init__(self, data, history_t90):self.datadataself.history_thistory_tself.reset()defreset(self):self.t0self.doneFalseself.profits0self.positions []self.position_value0self.history [0for_inrange(self.history_t)]return [self.position_value] self.history# obsdefstep(self, act):reward0# act 0: stay, 1: buy, 2: sellifact1:self.positions.append(self.data.iloc[self.t, :][Close])elifact2: # selliflen(self.positions) 0:reward-1else:profits0forpinself.positions:profits (self.data.iloc[self.t, :][Close] -p)rewardprofitsself.profitsprofitsself.positions []# set next timeself.t1self.position_value0forpinself.positions:self.position_value (self.data.iloc[self.t, :][Close] -p)self.history.pop(0)self.history.append(self.data.iloc[self.t, :][Close] -self.data.iloc[(self.t-1), :][Close])# clipping rewardifreward0:reward1elifreward0:reward-1return [self.position_value] self.history, reward, self.done, self.profits# obs, reward, done, profits首先定义强化学习的环境这里的Environment的类模拟了一个简单的交易环境。使用历史股票价格数据代理可以根据这些数据决定是否购买、出售或持有股票。
init()接受两个参数:data表示股票价格数据;history_t定义环境应该维持多少时间步长。通过设置data和history_t值并调用reset()构造函数初始化了环境。
Reset()初始化或重置环境的内部状态变量包括当前时间步长(self.t)、完成标志、总利润、未平仓头寸、头寸值和历史价格。该方法返回由头寸价值和价格历史组成的观测值。
step()方法可以基于一个动作更新环境的状态。动作用整数表示:0表示持有1表示购买2表示出售。如果代理人决定买入股票的当前收盘价将被添加到头寸列表中。一旦经纪人决定卖出该方法计算每个未平仓头寸的利润或损失并相应地更新利润变量。然后所有未平仓头寸被平仓。根据卖出行为中产生的利润或损失奖励被削减到- 1,0或1。
代理可以使用Environment类学习并根据历史股票价格数据做出决策Environment类模拟股票交易环境。在受控环境中可以训练强化学习代理来制定交易策略。 env Environment(train)print(env.reset())for _ in range(3):pact np.random.randint(3)print(env.step(pact))DQN deftrain_dqn(env, epoch_num50):classQ_Network(chainer.Chain):def__init__(self, input_size, hidden_size, output_size):super(Q_Network, self).__init__(fc1L.Linear(input_size, hidden_size),fc2L.Linear(hidden_size, hidden_size),fc3L.Linear(hidden_size, output_size)) def__call__(self, x):hF.relu(self.fc1(x))hF.relu(self.fc2(h))yself.fc3(h)returnydefreset(self):self.zerograds()QQ_Network(input_sizeenv.history_t1, hidden_size100, output_size3)Q_astcopy.deepcopy(Q)optimizerchainer.optimizers.Adam()optimizer.setup(Q)step_maxlen(env.data)-1memory_size200batch_size20epsilon1.0epsilon_decrease1e-3epsilon_min0.1start_reduce_epsilon200train_freq10update_q_freq20gamma0.97show_log_freq5 memory []total_step0total_rewards []total_losses [] starttime.time()forepochinrange(epoch_num): pobsenv.reset()step0doneFalsetotal_reward0total_loss0whilenotdoneandstepstep_max: # select actpactnp.random.randint(3)ifnp.random.rand() epsilon:pactQ(np.array(pobs, dtypenp.float32).reshape(1, -1))pactnp.argmax(pact.data) # actobs, reward, done, profitenv.step(pact) # add memorymemory.append((pobs, pact, reward, obs, done))iflen(memory) memory_size:memory.pop(0) # train or update qiflen(memory) memory_size:iftotal_step%train_freq0:shuffled_memorynp.random.permutation(memory)memory_idxrange(len(shuffled_memory))foriinmemory_idx[::batch_size]:batchnp.array(shuffled_memory[i:ibatch_size])b_pobsnp.array(batch[:, 0].tolist(), dtypenp.float32).reshape(batch_size, -1)b_pactnp.array(batch[:, 1].tolist(), dtypenp.int32)b_rewardnp.array(batch[:, 2].tolist(), dtypenp.int32)b_obsnp.array(batch[:, 3].tolist(), dtypenp.float32).reshape(batch_size, -1)b_donenp.array(batch[:, 4].tolist(), dtypenp.bool)qQ(b_pobs)maxqnp.max(Q_ast(b_obs).data, axis1)targetcopy.deepcopy(q.data)forjinrange(batch_size):target[j, b_pact[j]] b_reward[j]gamma*maxq[j]*(notb_done[j])Q.reset()deephub_lossF.mean_squared_error(q, target)total_lossloss.dataloss.backward()optimizer.update()iftotal_step%update_q_freq0:Q_astcopy.deepcopy(Q) # epsilonifepsilonepsilon_minandtotal_stepstart_reduce_epsilon:epsilon-epsilon_decrease # next steptotal_rewardrewardpobsobsstep1total_step1 total_rewards.append(total_reward)total_losses.append(total_loss)if (epoch1) %show_log_freq0:log_rewardsum(total_rewards[((epoch1)-show_log_freq):])/show_log_freqlog_losssum(total_losses[((epoch1)-show_log_freq):])/show_log_freqelapsed_timetime.time()-startprint(\t.join(map(str, [epoch1, epsilon, total_step, log_reward, log_loss, elapsed_time])))starttime.time()returnQ, total_losses, total_rewards这段代码定义了一个函数train_dqn()它为一个简单的股票交易环境训练一个Deep Q-Network (DQN)。该函数接受两个参数:一个是env参数表示交易环境;另一个是epoch_num参数指定要训练多少epoch。
代码定义了一个Q_Network类它是Chainer的Chain类的一个子类。在Q-Network中有三个完全连接的层前两层具有ReLU激活函数。模型梯度通过reset()方法归零。
创建Q- network的两个实例Q和Q_ast以及用于更新模型参数的Adam优化器。对于DQN训练定义了几个超参数包括缓冲区内存大小、批处理大小、epsilon、gamma和更新频率。
为了跟踪模型在训练期间的表现在内存列表中创建total_rewards和total_losses列表。在每个epoch的开始环境被重置一些变量被初始化。
代理根据当前状态和epsilon-greedy探索策略选择一个动作(持有、买入或卖出)。然后代理在环境中执行动作获得奖励并观察新的状态。在缓冲区中存储了经验元组(前一个状态、动作、奖励、新状态和完成标志)。
为了训练DQN当缓冲区满时从内存中采样一批经验。利用Q_ast网络和Bellman方程计算了目标q值。损失计算为预测q值与目标q值之间的均方误差。计算梯度优化器更新模型参数。
目标网络Q_ast使用主网络q的权值定期更新随着智能体的学习epsilon值线性减小促进更多的利用。每个时期总奖励和损失都会累积起来结果也会被记录下来。
训练结束时train_dqn()返回训练后的Q-Network、总损失和总奖励。DQN模型可用于根据输入的股票价格数据和模拟的交易环境制定交易策略。 dqn, total_losses, total_rewards train_dqn(Environment(train), epoch_num25)这段代码使用来自指定环境的训练数据(使用train_dqn函数)训练DQN模型并返回训练后的模型以及每个训练历元的总损失和奖励。
def plot_loss_reward(total_losses, total_rewards):figure tools.make_subplots(rows1, cols2, subplot_titles(loss, reward), print_gridFalse)figure.append_trace(Scatter(ytotal_losses, modelines, linedict(colorskyblue)), 1, 1)figure.append_trace(Scatter(ytotal_rewards, modelines, linedict(colororange)), 1, 2)figure[layout][xaxis1].update(titleepoch)figure[layout][xaxis2].update(titleepoch)figure[layout].update(height400, width900, showlegendFalse)iplot(figure)plot_loss_reward”使用Plotly库的“make_subplots”函数创建一个带有两个子图的图形。在训练周期内该图显示了损失值和奖励值的趋势提供了对DQN模型性能的洞察。
plot_loss_reward(total_losses, total_rewards)显示了在训练时期损失和奖励值的趋势。在DQN模型(可能用于股票市场预测)的训练过程中代码使用该函数绘制损失和回报值。
def plot_train_test_by_q(train_env, test_env, Q, algorithm_name):# trainpobs train_env.reset()train_acts []train_rewards []train_ongoing_profits []for _ in range(len(train_env.data)-1):pact Q(np.array(pobs, dtypenp.float32).reshape(1, -1))pact np.argmax(pact.data)train_acts.append(pact)obs, reward, done, profit train_env.step(pact)train_rewards.append(reward)train_ongoing_profits.append(profit) pobs obstrain_profits train_env.profits# testpobs test_env.reset()test_acts []test_rewards []test_ongoing_profits [] for _ in range(len(test_env.data)-1):pact Q(np.array(pobs, dtypenp.float32).reshape(1, -1))pact np.argmax(pact.data)test_acts.append(pact)deep_hub_obs, reward, done, profit test_env.step(pact)test_rewards.append(reward)test_ongoing_profits.append(profit) pobs obstest_profits test_env.profits# plottrain_copy train_env.data.copy()test_copy test_env.data.copy()train_copy[act] train_acts [np.nan]train_copy[reward] train_rewards [np.nan]test_copy[act] test_acts [np.nan]test_copy[reward] test_rewards [np.nan]train0 train_copy[train_copy[act] 0]train1 train_copy[train_copy[act] 1]train2 train_copy[train_copy[act] 2]test0 test_copy[test_copy[act] 0]test1 test_copy[test_copy[act] 1]test2 test_copy[test_copy[act] 2]act_color0, act_color1, act_color2 gray, cyan, magenta data [Candlestick(xtrain0.index, opentrain0[Open], hightrain0[High], lowtrain0[Low], closetrain0[Close], increasingdict(linedict(coloract_color0)), decreasingdict(linedict(coloract_color0))),Candlestick(xtrain1.index, opentrain1[Open], hightrain1[High], lowtrain1[Low], closetrain1[Close], increasingdict(linedict(coloract_color1)), decreasingdict(linedict(coloract_color1))),Candlestick(xtrain2.index, opentrain2[Open], hightrain2[High], lowtrain2[Low], closetrain2[Close], increasingdict(linedict(coloract_color2)), decreasingdict(linedict(coloract_color2))),Candlestick(xtest0.index, opentest0[Open], hightest0[High], lowtest0[Low], closetest0[Close], increasingdict(linedict(coloract_color0)), decreasingdict(linedict(coloract_color0))),Candlestick(xtest1.index, opentest1[Open], hightest1[High], lowtest1[Low], closetest1[Close], increasingdict(linedict(coloract_color1)), decreasingdict(linedict(coloract_color1))),Candlestick(xtest2.index, opentest2[Open], hightest2[High], lowtest2[Low], closetest2[Close], increasingdict(linedict(coloract_color2)), decreasingdict(linedict(coloract_color2)))]title {}: train s-reward {}, profits {}, test s-reward {}, profits {}.format(deephub_algorithm_name,int(sum(train_rewards)),int(train_profits),int(sum(test_rewards)),int(test_profits))layout {title: title,showlegend: False,shapes: [{x0: date_split, x1: date_split, y0: 0, y1: 1, xref: x, yref: paper, line: {color: rgb(0,0,0), width: 1}}],annotations: [{x: date_split, y: 1.0, xref: x, yref: paper, showarrow: False, xanchor: left, text: test data},{x: date_split, y: 1.0, xref: x, yref: paper, showarrow: False, xanchor: right, text: train data }]}figure Figure(datadata, layoutlayout)iplot(figure)return train_ongoing_profits, test_ongoing_profitsPlot_train_test_by_q()将训练好的DQN模型在训练和测试数据集上的交易行为和性能可视化。
使用训练好的Q-Network该函数初始化环境并迭代训练和测试数据。对于这两个数据集它都会累积行动、奖励和持续利润。
为了分析或比较算法的性能该函数返回训练和测试数据集的持续利润。
train_profits, test_profits plot_train_test_by_q(Environment(train), Environment(test), dqn, DQN)基于DQN模型的预测’ train_profits ‘变量接收从训练数据中获得的利润。’ test_profits 接收测试数据作为DQN模型预测的结果而获得的利润。
代码在训练和测试数据上评估训练好的DQN模型并计算每个数据集上获得的利润。这种评价可能有助于确定DQN模型的准确性和有效性。
我们还可以将将DQN模型的性能与用于股市预测的“买入并持有”策略进行比较。Matplotlib的’ plt 模块用于生成绘图。
plt.figure(figsize(23,8))
plt.plot(data.index,((data[Close]-data[Close][0])/data[Close][-1]), labelbuy and hold)
plt.plot(train.index, ([0] train_profits)/data[Close][-1], labelrl (train))
plt.plot(test.index, (([0] test_profits) train_profits[-1])/data[Close][-1], labelrl (test))
plt.ylabel(relative gain)
plt.legend()
plt.show()代码绘制了在训练数据上获得的利润。在该图中x轴表示训练数据的指数y轴表示DQN模型预测的相对收益。相对收益是通过将利润除以输入数据中的最后收盘价来计算的。
使用DQN模型代码绘制了在测试数据上获得的利润。x轴表示测试数据的指数y轴表示DQN模型预测的相对利润增益。通过将训练利润相加并除以导入数据中的最后收盘价来计算相对收益。该图的标签为“rl (test)”。
使用Matplotlib库的“show”函数显示该图。在训练和测试数据上该图显示了“买入并持有”策略和DQN模型预测的相对利润增益。将DQN模型与“买入并持有”等简单策略进行比较可以深入了解其有效性。
Double DQN
Double Deep Q-Network (DDQN) 是一种用于强化学习中的深度学习算法特别是在处理离散动作空间的 Q-Learning 问题时非常有效。DDQN 是对传统 Deep Q-Network (DQN) 的一种改进旨在解决 DQN 在估计 Q 值时可能存在的过高估计overestimation问题。
DDQN 使用一个额外的神经网络来评估选取最大 Q 值的动作。它不再直接使用目标 Q 网络预测的最大 Q 值来更新当前 Q 网络的 Q 值而是使用当前 Q 网络选择的动作在目标 Q 网络中预测的 Q 值来更新。这种方法通过减少动作选择与目标 Q 值计算之间的相关性有助于减轻 Q 值的过高估计问题。
def train_ddqn(env, epoch_num50):class Q_Network(chainer.Chain): def __init__(self, input_size, hidden_size, output_size):super(Q_Network, self).__init__(fc1 L.Linear(input_size, hidden_size),fc2 L.Linear(hidden_size, hidden_size),fc3 L.Linear(hidden_size, output_size)) def __call__(self, x):h F.relu(self.fc1(x))h F.relu(self.fc2(h))y self.fc3(h)return ydef reset(self):self.zerograds()Q Q_Network(input_sizeenv.history_t1, hidden_size100, output_size3)Q_ast copy.deepcopy(Q)optimizer chainer.optimizers.Adam()optimizer.setup(Q)step_max len(env.data)-1memory_size 200batch_size 50epsilon 1.0epsilon_decrease 1e-3epsilon_min 0.1start_reduce_epsilon 200train_freq 10update_q_freq 20gamma 0.97show_log_freq 5 memory []total_step 0total_rewards []total_losses [] start time.time()for epoch in range(epoch_num): pobs env.reset()step 0done Falsetotal_reward 0total_loss 0while not done and step step_max: # select actpact np.random.randint(3)if np.random.rand() epsilon:pact Q(np.array(pobs, dtypenp.float32).reshape(1, -1))pact np.argmax(pact.data) # actobs, reward, done, profit env.step(pact) # add memorymemory.append((pobs, pact, reward, obs, done))if len(memory) memory_size:memory.pop(0) # train or update qif len(memory) memory_size:if total_step % train_freq 0:deep_hub_shuffled_memory np.random.permutation(memory)memory_idx range(len(shuffled_memory))for i in memory_idx[::batch_size]:batch np.array(shuffled_memory[i:ibatch_size])b_pobs np.array(batch[:, 0].tolist(), dtypenp.float32).reshape(batch_size, -1)b_pact_deephub np.array(batch[:, 1].tolist(), dtypenp.int32)b_reward np.array(batch[:, 2].tolist(), dtypenp.int32)b_obs np.array(batch[:, 3].tolist(), dtypenp.float32).reshape(batch_size, -1)b_done np.array(batch[:, 4].tolist(), dtypenp.bool) q Q(b_pobs) DQN - Double DQNmaxq np.max(Q_ast(b_obs).data, axis1) indices np.argmax(q.data, axis1)maxqs Q_ast(b_obs).data target copy.deepcopy(q.data)for j in range(batch_size): DQN - Double DQNtarget[j, b_pact[j]] b_reward[j]gamma*maxq[j]*(not b_done[j]) target[j, b_pact[j]] b_reward[j]gamma*maxqs[j, indices[j]]*(not b_done[j]) Q.reset()loss F.mean_squared_error(q, target)total_loss loss.dataloss.backward()optimizer.update()if total_step % update_q_freq 0:Q_ast copy.deepcopy(Q) # epsilonif epsilon epsilon_min and total_step start_reduce_epsilon:epsilon - epsilon_decrease # next steptotal_reward rewardpobs obsstep 1total_step 1total_rewards.append(total_reward)total_losses.append(total_loss)if (epoch1) % show_log_freq 0:log_reward sum(total_rewards[((epoch1)-show_log_freq):])/show_log_freqlog_loss sum(total_losses[((epoch1)-show_log_freq):])/show_log_freqelapsed_time time.time()-startprint(\t.join(map(str, [epoch1, epsilon, total_step, log_reward, log_loss, elapsed_time])))start time.time()return Q, total_losses, total_rewards上面代码定义了一个函数train_ddqn()该函数训练Double Deep Q-Network (DDQN)来解决交易环境。
ddqn, total_losses, total_rewards train_ddqn(Environment(train), epoch_num50)plot_loss_reward(total_losses, total_rewards)可视化了在训练时期的损失和奖励值的趋势。在DDQN模型(可能用于预测股票市场价格)的训练过程中该函数绘制损失和回报值。
train_profits, test_profits plot_train_test_by_q(Environment(train), Environment(test), ddqn, Double DQN)在训练和测试数据上评估训练后的DDQN模型的性能为每个数据集获得利润。对于股票市场预测或其他需要强化学习的决策任务此评估的结果可能有用。
plt.figure(figsize(23,8))
plt.plot(data.index,((data[Close]-data[Close][0])/data[Close][-1]), labelbuy and hold)
plt.plot(train.index, ([0] train_profits)/data[Close][-1], labelrl (train))
plt.plot(test.index, (([0] test_profits) train_profits[-1])/data[Close][-1], labelrl (test))
plt.ylabel(relative gain)
plt.legend()
plt.show()可以看到Double DQN要更高一些。这和Double Deep Q-Network的介绍 (DDQN)通过使用两个神经网络来分别估计当前策略选择的动作和目标 Q 值的最大动作有效解决了传统 DQN 中的 Q 值过高估计问题提高了在离散动作空间下的强化学习性能和稳定性。是相吻合的
Dueling Double DQN
Dueling Double Deep Q-Network (Dueling DDQN) 是一种结合了两种技术的强化学习算法Dueling网络结构和Double DQN。它旨在进一步提高 Q-Learning 的效率和稳定性特别是在处理离散动作空间的问题时非常有效。
def train_dddqn(env, epoch_num50): Double DQN - Dueling Double DQNclass Q_Network(chainer.Chain): def __init__(self, input_size, hidden_size, output_size):super(Q_Network, self).__init__(fc1 L.Linear(input_size, hidden_size),fc2 L.Linear(hidden_size, hidden_size),fc3 L.Linear(hidden_size, output_size)) def __call__(self, x):h F.relu(self.fc1(x))h F.relu(self.fc2(h))y self.fc3(h)return y def reset(self):self.zerograds() class Q_Network(chainer.Chain): def __init__(self, input_size, hidden_size, output_size):super(Q_Network, self).__init__(fc1 L.Linear(input_size, hidden_size),fc2 L.Linear(hidden_size, hidden_size),fc3 L.Linear(hidden_size, hidden_size//2),fc4 L.Linear(hidden_size, hidden_size//2),state_value L.Linear(hidden_size//2, 1),advantage_value L.Linear(hidden_size//2, output_size))self.input_size input_sizeself.hidden_size hidden_sizeself.output_size output_size def __call__(self, x):h F.relu(self.fc1(x))h F.relu(self.fc2(h))hs F.relu(self.fc3(h))ha F.relu(self.fc4(h))state_value self.state_value(hs)advantage_value self.advantage_value(ha)advantage_mean (F.sum(advantage_value, axis1)/float(self.output_size)).reshape(-1, 1)q_value F.concat([state_value for _ in range(self.output_size)], axis1) (advantage_value - F.concat([advantage_mean for _ in range(self.output_size)], axis1))return q_valuedef reset(self):self.zerograds() Q Q_Network(input_sizeenv.history_t1, hidden_size100, output_size3)Q_ast copy.deepcopy(Q)optimizer chainer.optimizers.Adam()optimizer.setup(Q) step_max len(env.data)-1memory_size 200batch_size 50epsilon 1.0epsilon_decrease 1e-3epsilon_min 0.1start_reduce_epsilon 200train_freq 10update_q_freq 20gamma 0.97show_log_freq 5 memory []total_step 0total_rewards []total_losses []start time.time()for epoch in range(epoch_num):pobs env.reset()step 0done Falsetotal_reward 0total_loss 0while not done and step step_max: # select actpact np.random.randint(3)if np.random.rand() epsilon:pact Q(np.array(pobs, dtypenp.float32).reshape(1, -1))pact np.argmax(pact.data) # actobs, reward, done, profit env.step(pact) # add memorymemory.append((pobs, pact, reward, obs, done))if len(memory) memory_size:memory.pop(0) # train or update qif len(memory) memory_size:if total_step % train_freq 0:shuffled_memory np.random.permutation(memory)memory_idx range(len(shuffled_memory))for i in memory_idx[::batch_size]:deephub_batch np.array(shuffled_memory[i:ibatch_size])b_pobs np.array(batch[:, 0].tolist(), dtypenp.float32).reshape(batch_size, -1)b_pact np.array(batch[:, 1].tolist(), dtypenp.int32)b_reward np.array(batch[:, 2].tolist(), dtypenp.int32)b_obs np.array(batch[:, 3].tolist(), dtypenp.float32).reshape(batch_size, -1)b_done np.array(batch[:, 4].tolist(), dtypenp.bool) q Q(b_pobs) DQN - Double DQNmaxq np.max(Q_ast(b_obs).data, axis1) indices np.argmax(q.data, axis1)maxqs Q_ast(b_obs).data target copy.deepcopy(q.data)for j in range(batch_size): DQN - Double DQNtarget[j, b_pact[j]] b_reward[j]gamma*maxq[j]*(not b_done[j]) target[j, b_pact[j]] b_reward[j]gamma*maxqs[j, indices[j]]*(not b_done[j]) Q.reset()loss F.mean_squared_error(q, target)total_loss loss.dataloss.backward()optimizer.update()if total_step % update_q_freq 0:Q_ast copy.deepcopy(Q) # epsilonif epsilon epsilon_min and total_step start_reduce_epsilon:epsilon - epsilon_decrease # next steptotal_reward rewardpobs obsstep 1total_step 1total_rewards.append(total_reward)total_losses.append(total_loss)if (epoch1) % show_log_freq 0:log_reward sum(total_rewards[((epoch1)-show_log_freq):])/show_log_freqlog_loss sum(total_losses[((epoch1)-show_log_freq):])/show_log_freqelapsed_time time.time()-startprint(\t.join(map(str, [epoch1, epsilon, total_step, log_reward, log_loss, elapsed_time])))start time.time()return Q, total_losses, total_rewards在call方法中前两层在两个流之间共享然后分成两个独立的流。状态价值流有一个输出单个值的额外线性层(state_value)而优势价值流有一个为每个动作输出值的额外线性层(advantage_value)。最终的q值由状态值和优势值结合计算并减去平均优势值以保持稳定性。代码的其余部分与Double DQN实现非常相似。
dddqn, total_losses, total_rewards train_dddqn(Environment(train), epoch_num25)plot_loss_reward(total_losses, total_rewards)train_profits, test_profits plot_train_test_by_q(Environment(train), Environment(test), dddqn, Dueling Double DQN)plt.figure(figsize(23,8))
plt.plot(data.index,((data[Close]-data[Close][0])/data[Close][-1]), labelbuy and hold)
plt.plot(train.index, ([0] train_profits)/data[Close][-1], labelrl (train))
plt.plot(test.index, (([0] test_profits) train_profits[-1])/data[Close][-1], labelrl (test))
plt.plot(test.index, (([0] test_profits) - data[Close][0] data[Close][len(train_profits)])/data[Close][-1], labelrl (test))
plt.ylabel(relative gain)
plt.legend()
plt.show()Dueling Double Deep Q-Network (Dueling DDQN) 是一种结合了 Dueling 网络结构和 Double DQN 的强化学习算法。它通过将 Q 函数分解为状态值函数和优势函数来提高效率同时利用 Double DQN 的思想来减少 Q 值的过高估计从而在处理离散动作空间下的强化学习问题中表现出色。
总结
让我们对传统的 Deep Q-Network (DQN), Double DQN, Dueling DQN 和 Dueling Double DQN 进行对比总结看看它们各自的特点和优劣势。
1、Deep Q-Network (DQN)
特点- 使用深度神经网络来估计 Q 函数从而学习到每个状态下每个动作的价值。- 使用经验回放和固定 Q 目标网络来提高稳定性和收敛性。优点- 引入深度学习提高了 Q 函数的表示能力能够处理复杂的状态和动作空间。- 经验回放和固定 Q 目标网络有助于稳定训练过程减少样本间的相关性。缺点- 存在 Q 值的过高估计问题尤其是在动作空间较大时更为明显这可能导致训练不稳定和性能下降。
2、Double Deep Q-Network (Double DQN)
特点- 解决了 DQN 中 Q 值过高估计的问题。- 引入一个额外的目标 Q 网络来计算目标 Q 值减少更新时的相关性。优点- 减少了 Q 值的过高估计提高了训练的稳定性和收敛性。缺点- 算法结构相对简单对于某些复杂问题可能需要更高的表示能力。
3、Dueling Double Deep Q-Network (Dueling DDQN)
特点- 结合了 Dueling 网络结构和 Double DQN 的优势。- 使用 Dueling 网络结构来分解 Q 函数提高了效率和学习表示能力。- 使用 Double DQN 的思想来减少 Q 值的过高估计问题。优点- 综合了两种技术的优势能够在更广泛的问题空间中表现出色。- 提高了训练的稳定性和效率有助于更快地收敛到较好的策略。缺点- 算法实现和调优可能比单一 DQN 及其改进版更复杂。
总结比较
效果和稳定性Dueling DDQN 在处理动作空间较大的问题时表现出更高的效率和稳定性因为它们能够更有效地分离状态值和动作优势。过高估计问题Dueling DDQN 解决了传统 DQN 中 Q 值过高估计的问题其中 Double DQN 通过目标网络降低相关性而 Dueling 结构则通过优势函数减少过高估计。复杂性Dueling DDQN 相对于传统 DQN 和 Double DQN 更复杂需要更多的实现和理解成本但也带来了更好的性能。
传统 DQN 适用于简单的强化学习任务而 Double DQN、Dueling DDQN 则适用于更复杂和具有挑战性的问题根据问题的特性选择合适的算法可以有效提升训练效率和性能。
最后我们也看到深度强化学习预测股票是可行的因为他不再预测具体的股票价格而是针对收益预测买进卖出和持有我们这里只是使用了股票本身的数据如果有更多的外生数据那么强化学习应该可以模拟更准确的人工操作。
数据集地址:
https://avoid.overfit.cn/post/9442a4a4b0be43e3b2996d2f4137c588
作者Connor Roberts
