强化学习

    对银行借贷进行建模，企业确定每个时期的累计资金需求量，然后在与银行协商的过程中，银行给出贷款利率，企业确定实际贷款金额。主要针对于银行借贷业务频繁、经营状况良好且拥有一定资产的大型企业。对于这类企业，在银行资金充裕情况下，基本上都可以借到款项。企业每个时期都有一定的资金需求，而贷款时间越长，贷款利率越高，故企业会充分考虑资金的时间价值，尽量缩短贷款时间，避免不必要的资金成本。假设企业通过合理的时间划分，实现企业在每个时段初贷款，在该时段末还款。使用强化学习进行求解，相关部分代码如下：

构建环境

class LoanEnv(gym.Env):def __init__(self,company_num=20,max_episode_steps=2):self.max_episode_steps=max_episode_steps #定义一个回合的最大步数self.current_steps=None  #存储当前的步数self.current_cost_period=None #存储当前的阶段self.cost_nonconstant_variable= None #定义成本中非固定参数self.cost_constant_variable=0.002  #定义成本中固定参数self.loss_balance_factor=0.6       #定义银行与企业损失占总损失的比例self.company_num=company_num #定义企业数量self.random_need_mean=10#定义随机需求的均值（为了简化模型，固定值）self.random_need_stderror=5 #定义随机需求的标准差，保证有94%的概率取正值（为了简化模型，固定值）self.rate_sensitivity=0.004  #定义利率敏感性，用以计算企业损失(为了简化模型，固定值)self.overstock_rate=0.5   #定义需求积压率，用以计算资金需求self.baseline_rate=0.0435  #基准利率self.commitment_rate=0.005  #承诺费率（为了简化模型，固定值）self.floating_cap=0.3      #利率浮动上限self.floating_floor=0.2   #利率浮动下限self.rate_floor=(1 - self.floating_floor) * self.baseline_rate self.rate_cap=(1 + self.floating_cap) * self.baseline_rate      self.needs_low=0          #定义需求下界self.needs_high=10000     #定义需求上界self.period_low=0         #定义阶段下界self.period_high=20       #定义阶段上界self.cost_low=0           #定义成本下界self.cost_high=10000      #定义成本上界self.low=np.array([self.needs_low,self.period_low,self.cost_low]) #定义状态空间下界self.high=np.array([self.needs_high,self.period_high,self.cost_high]) #定义状态空间上界self.action_space = spaces.Box(low=self.rate_floor, high=self.rate_cap, shape=(self.company_num,), dtype=np.float64)  #设置动作空间self.observation_space = spaces.Box(low=self.low,high=self.high,dtype=np.float64)                               #设置观测空间self.seed()def seed(self, seed=None):   self.np_random, seed = seeding.np_random(seed)return [seed]def cost_period(self):  #每一年，即12步，重新选择一次银行成本中非固定参数new_cost_period= np.floor((self.current_steps - 1) / 12 + 1 ) #计算当前的成本阶段if new_cost_period != self.current_cost_period:  #如果当前的成本阶段与实际不相符合，更新成本中的非固定参数以及成本阶段self.cost_nonconstant_variable=np.random.choice([0.000005,0.00001,0.00002])print("成本非固定参数:",self.cost_nonconstant_variable)self.current_cost_period=new_cost_periodprint('当前成本阶段:',self.current_cost_period)def get_cost_period(self): #获取当前的损失阶段return self.current_cost_perioddef need_function(self,actual_needs): #定义状态转移expected_needs=self.state[0]random_needs=np.random.normal(loc=self.random_need_mean,scale=self.random_need_stderror,size=self.company_num)  #使用正态分布，产生随机需求print("随机需求：",random_needs)need_extend_rates=[np.random.choice([0,0.1,0.2,0.3,0.4,0.5]) for i in range(self.company_num)] #随机选择需求扩展率print("需求扩展率:",need_extend_rates)expected_needs=self.overstock_rate * (expected_needs-actual_needs) + need_extend_rates * actual_needs +  random_needs  #定义下一时段的预计期望需求expected_needs=np.clip(expected_needs,self.needs_low,self.needs_high)print("下一阶段的需求：",expected_needs)return np.array(expected_needs) def get_state(self):  #获取当前状态return np.array(self.state)def company_loss(self,actual_needs,actions): #定义企业损失expected_needs=self.state[0]company_losses=[self.rate_sensitivity * (expected_needs[i] - actual_needs[i]) ** 2 + actions[i] * actual_needs[i] +  self.commitment_rate * expected_needs[i]  for i in range(self.company_num)]#产生企业损失数组print("企业损失：",company_losses)total_company_losses=np.sum(company_losses) #对企业损失数组进行求和print("总的企业损失:",total_company_losses)return total_company_lossesdef get_company_loss(self,actual_needs,actions):  #获取企业损失return np.array(self.company_loss(actual_needs,actions))def bank_cost(self):total_expected_needs=np.sum(self.state[0]) #对实际需求求和print("总的需求:",total_expected_needs)costs=self.cost_constant_variable * total_expected_needs + self.cost_nonconstant_variable * (total_expected_needs) **2 cost=np.clip(costs,self.cost_low,self.cost_high)print("银行成本：",cost)return costdef bank_loss(self,actual_needs,actions):total_actual_needs=np.sum(actual_needs,dtype=np.float64)total_expected_needs=np.sum(self.state[0],dtype=np.float64)bank_losses=self.cost_constant_variable * total_expected_needs + self.cost_nonconstant_variable * (total_expected_needs) **2 - np.sum(actions * actual_needs) - self.commitment_rate * total_expected_needs#计算银行损失print("银行损失:",bank_losses)return bank_lossesdef get_bank_loss(self,actual_needs,actions): #获取银行损失return self.bank_loss(actual_needs,actions)def step(self, actions):assert self.action_space.contains(actions), "%r (%s) invalid" % (actions, type(actions))  #检查动作是否符合动作空间的要求self.current_steps +=1  #步数+1if self.current_steps <= self.max_episode_steps:self.cost_period()actual_needs=self.actual_need(actions) #根据利率计算实际需求expected_needs=self.need_function(actual_needs) #根据预期需求以及实际需求计算下一期预期需求period=self.current_cost_period #定义所属阶段cost=self.bank_cost()   #定义银行成本bank_losses=self.bank_loss(actual_needs,actions)  #计算银行损失total_company_losses=self.company_loss(actual_needs,actions) #计算企业损失total_losses=self.loss_balance_factor * bank_losses +(1-self.loss_balance_factor) * total_company_losses #定义总的损失print("总的损失：",total_losses)reward=-total_losses  #定义奖励if self.current_steps == self.max_episode_steps: #判断回合是否结束done=Trueelse:done=Falseself.state=(expected_needs,period,cost)  #定义下一步的状态print(self.state)print("当前的步数：",self.current_steps)return np.array(self.state), reward, done, {}else:self.reset()return self.step(actions)def reset(self): #重置环境 (初始值全部赋值100万)                expected_needs= 100 * np.ones((self.company_num,)) self.current_steps=0self.current_cost_period=0bankcost=0self.state=(expected_needs,self.current_cost_period,bankcost)return self.statedef actual_need(self,actions): #计算实际需求expected_needs=self.state[0]actual_needs=[(2 *  self.rate_sensitivity * expected_needs[i] - actions[i] ) / (2 * self.rate_sensitivity ) for i in range(self.company_num)]#计算实际需求print("实际需求：",actual_needs)return np.array(actual_needs)def get_actual_need(self,actions): #获取实际需求return np.array(self.actual_need(actions))

强化学习求解

#初始化
class OrnsteinUhlenbeckProcess():def __init__(self,size,mu=0.,sigma=1.,theta=0.15,dt=0.01):self.size=sizeself.mu=muself.sigma=sigmaself.theta=thetaself.dt=dt  #差分方程的粒度def reset(self,x=0.):#开始一套新的过程self.x=x * np.ones(self.size)def __call__(self): #输出一组值n=np.random.normal(size=self.size)self.x +=(self.theta * (self.mu-self.x) * self.dt +self.sigma * np.sqrt(self.dt) * n)return self.x

class DQNReplayer:def __init__(self,capacity): #参数capacity表示最多可以存储多少条经验，当要存储的经验数超过capacity时，会用新的经验覆盖最早存入的经验self.memory=pd.DataFrame(index=range(capacity),columns=['observation','action','reward','next_observation','done']) #定义存储结构self.i=0    #定义目前所在的行数self.count=0  #定义目前存储的经验数量self.capacity=capacity  #赋值容量def store(self, *args): #在python中，当*和**符号出现在函数定义的参数中时，表示任意数目参数收集。*arg表示任意多个无名参数，类型为tuple;**kwargs表示关键字参数，为dict，使用时需将*arg放在**kwargs之前self.memory.loc[self.i]=args #loc为按标签取数据，第一个参数选择index，第二个参数选择column（如果省略，等同于“：”，即取该行全部数据）；iloc即按位置选择数据，即第n行，第n列数据，所以传入的是位置的整数型参数。self.i=(self.i+1) % self.capacity #行数指标下移一个，取余操作是为了防止超过容量self.count=min(self.count+1,self.capacity)  #即时存储数量加1，取最小也是为了防止超过容量def sample(self,size):indices=np.random.choice(self.count,size=size)  #从目前的存储经验集中，等概率抽取size个数据return(np.stack(self.memory.loc[indices,field]) for field in self.memory.columns)class DDPGAgent():def __init__(self,env,actor_kwargs,critic_kwargs,replayer_capacity=1000000,replayer_initial_transitions=10000,gamma=0.99,batches=20,batch_size=240,net_learning_rate=0.02,noise_scale=0.1,explore=True):#在未到达replayer_initial_transitions值，随机选取动作，随机试探，也不训练网络，从而保证刚开始有足够多的试探observation_dim=len(env.state[0]) + 2 #由于状态的第一个元素，是一个数组，所有在计算维度时，需要将其单独提出action_dim=env.action_space.shape[0] #动作空间也是一维数组，shape[0]返回元素个数，同上observation_action_dim=observation_dim + action_dimself.action_low=env.action_space.low #动作空间下界self.action_high=env.action_space.high #动作空间上界self.gamma=gammaself.net_learning_rate=net_learning_rate #目标网络学习率self.explore=explore #是否为动作加噪声进行探索self.batches=batchesself.batch_size=batch_sizeself.replayer=DQNReplayer(replayer_capacity)self.replayer_initial_transitions=replayer_initial_transitionsself.noise= OrnsteinUhlenbeckProcess(size=(action_dim,),sigma=noise_scale) #噪声对象self.noise.reset() #初始化噪声对象self.actor_evaluate_net=self.build_network(input_size=observation_dim,**actor_kwargs) #执行者评估网络self.actor_target_net=self.build_network(input_size=observation_dim,**actor_kwargs) #执行者目标网络self.critic_evaluate_net=self.build_network(input_size=observation_action_dim,**critic_kwargs) #评论者评估网络self.critic_target_net=self.build_network(input_size=observation_action_dim,**critic_kwargs) #评论者目标网络self.update_target_net(self.actor_target_net,self.actor_evaluate_net) #更新执行者目标网络self.update_target_net(self.critic_target_net,self.critic_evaluate_net) #更新评论者目标网络def update_target_net(self,target_net,evaluate_net,learning_rate=1.): #更新目标网络,learning_rate参数用于衡量从评估网络学习的程度target_weights=target_net.get_weights()   #获得目标网络权重evaluate_weights=evaluate_net.get_weights() #获得评估网络权重average_weights=[(1.-learning_rate) * t + learning_rate * e for t,e in zip(target_weights,evaluate_weights)] #利用learning_rate参数,实现在两者之间学习的平衡，从而产生平均权重target_net.set_weights(average_weights)  #将平均权重赋值给目标网络权重def build_network(self,input_size,hidden_sizes,output_size=1,activation=tf.nn.relu,output_activation=None,loss=keras.losses.mse,learning_rate=None):  #构建网络#由于是连续动作空间，我们无法将所有的动作都取出，并产生相应的输出单元，我们只能采用确定性性算法，即最后只输出一个动作（动作价值）model=keras.Sequential()for layer,hidden_size in enumerate(hidden_sizes): #利用for循环定义隐藏层，enumerate函数可以为数组产生index，默认从0开始kwargs={'input_shape': (input_size,)} if layer == 0 else {} #因为使用keras构建神经网络模型，第一层即输入层需要指定输入单元数，其他层则不需要model.add(keras.layers.Dense(units=hidden_size,activation=activation,kernel_initializer=keras.initializers.glorot_uniform(seed=0),**kwargs))model.add(keras.layers.Dense(units=output_size,activation=activation,kernel_initializer=keras.initializers.glorot_uniform(seed=0),))#定义输出层optimizer=keras.optimizers.Adam(learning_rate) #定义优化器model.compile(optimizer=optimizer,loss=loss) ##compile函数用于配置训练模型return modeldef decide(self,observation): #决策，即选择动作if self.explore and self.replayer.count < self.replayer_initial_transitions: #数据量不够时，鼓励随机探索return 100 * np.random.uniform(self.action_low,self.action_high) #输出随机动作,因为动作值为利率过小，所以放大100action=self.actor_evaluate_net.predict(observation[np.newaxis])[0]  #确定性算法，输出单元数为1if self.explore : #为动作加噪声noise=self.noise()   #前面根据OrnsteinUhlenbeckProcess已经初始化过action=np.clip(action + noise,100 * self.action_low,100 * self.action_high) #np.clip函数用于截断，第一个参数为输入数据，第二个参数为数据下界，第三个数据上界，对输入数据中的每个元素进行判别#小于下界的元素值，以下界值代替相应的元素值；大于上界的元素值，以上界值代替相应的值return action def learn(self,observation,action,reward,next_observation,done):self.replayer.store(observation,action,reward,next_observation,done) #存储经验if self.replayer.count > self.replayer_initial_transitions:if done :  #240步为一回合self.noise.reset() #为下一回合重置噪声for batch in range(self.batches):observations,actions,rewards,next_observations,dones=self.replayer.sample(self.batch_size) #经验回放 #训练执行者网络observation_tensor=tf.convert_to_tensor(observations,dtype=tf.float32) #数据转化成TensorFlow中的张量类型with tf.GradientTape() as tape : #使用eager模式#tf语句都要通过sess.run()来运行，但在eager模式中，操作一旦调用便立刻执行，如下面的变量定义action_tensor=self.actor_evaluate_net(observation_tensor)input_tensor=tf.concat([observation_tensor,action_tensor],axis=1) #评论者的输入#tf.concat函数用于连接矩阵，axis=0表示第一维上进行，axis=1表示在第二维上进行连接#这里表示分别将一个状态值和一个动作值连接在一起q_tensor=self.critic_evaluate_net(input_tensor)loss_tensor=-tf.reduce_mean(q_tensor)  #执行者网络#由于执行者网络训练的是策略，而且是确定性就意味着只有一个输出动作，而我们的目标的是该输出动作对应的动作价值应该最大grad_tensors=tape.gradient(loss_tensor,self.actor_evaluate_net.Variables) #计算执行者评估网络的参数#第一个参数是被求导的式子，第二个参数是被求导的变量self.actor_evaluate.optimizer.apply_gradient(zip(grad_tensors,self.actor_evaluate_net.Variables))#注意这里optimizer在前面构建网络的时候已经定义好了，采用adam优化器，如果没有定义优化器，是不能直接这样直接使用的#apply_gradient作用是把计算出来的梯度更新到变量上面去#采用zip封装，使得每一个梯度与其对应的变量封装成一个元组#训练评论者网络next_actions=self.actor_target_net.predict(next_observations)observation_actions=np.hstack([observations,actions])#np.hstack 将两个数组按照水平方向，即列顺序，堆叠成一个新数组#np.vstack 将两个数组按照垂直方向，即行顺序，堆叠成一个新数组next_observation_actions=np.hstack([next_observations,next_actions])next_qs=self.critic_target_net.predict(next_observation_actions)[:,0] targets=rewards + self.gamma * next_qs * (1-dones) #目标self.critic_evaluate_net.fit(observation_actions,targets,verbose=0) #更新评论者网络#优化器，损失函数以及模型的配置，在构建网络时已定义好self.update_target_net(self.actor_target_net,self.actor_evaluate_net,self.net_learning_rate) #更新执行者目标网络权重#这里net_learning_rate是在初始化智能体时定义的，取值为0.005，即每次更新执行者网络权重时，只取评估网络权重的0.005#因为每个批次只有64个数据，评估者网络权重也变化较大self.update_target_net(self.critic_target_net,self.critic_evaluate_net,self.net_learning_rate) #更新评估者网络权重#双重延迟深度确定性算法智能体
#在之前深度确定性算法基础上，又加两个用于评估者网络的神经网络，从而组成了两个评论者，每个评论者各有一个评估网络和目标网络
#我们希望通过取两个评论者目标网络的最小值，从而消除最大化误差
class TD3Agent(DDPGAgent):def __init__(self,env,actor_kwargs,critic_kwargs,replayer_capacity=100000,replayer_initial_transitions=10000,gamma=0.99,batches=20,batch_size=240,net_learning_rate=0.02,noise_scale=0.1,explore=True):#构造函数大部分和DDPGAgent相同，以下开始的内容和DDPGAgent类相同observation_dim=len(env.state[0]) + 2 #由于状态的第一个元素，是一个数组，所有在计算维度时，需要将其单独提出action_dim=env.action_space.shape[0]observation_action_dim=observation_dim + action_dimself.action_low=env.action_space.lowself.action_high=env.action_space.highself.gamma=gammaself.net_learning_rate=net_learning_rateself.explore=exploreself.batches=batchesself.batch_size=batch_sizeself.replayer=DQNReplayer(replayer_capacity)self.replayer_initial_transitions=replayer_initial_transitionsself.noise=OrnsteinUhlenbeckProcess(size=(action_dim,),sigma=noise_scale) #声明noise对象self.noise.reset() #重置noise对象#以上是和DDPGAgent类构造函数相同的部分#在构建网络时和DDPGAgent不同，构造了2个执行者网络和4个评论者网络self.actor_evaluate_net=self.build_network(input_size=observation_dim,**actor_kwargs)self.actor_target_net=self.build_network(input_size=observation_dim,**actor_kwargs)self.critic0_evaluate_net=self.build_network(input_size=observation_action_dim,**critic_kwargs)self.critic0_target_net=self.build_network(input_size=observation_action_dim,**critic_kwargs)self.critic1_evaluate_net=self.build_network(input_size=observation_action_dim,**critic_kwargs)self.critic1_target_net=self.build_network(input_size=observation_action_dim,**critic_kwargs)self.update_target_net(self.actor_target_net,self.actor_evaluate_net) #跟新执行者目标网络self.update_target_net(self.critic0_target_net,self.critic0_evaluate_net) #跟新评论者目标网络self.update_target_net(self.critic1_target_net,self.critic1_evaluate_net) #跟新评论者目标网络def learn(self,observation,action,reward,next_observation,done):#学习函数.大部分内容也是和DDPGAgent.learn相同，只是在训练网络时有所区别self.replayer.store(observation,action,reward,next_observation,done)if self.replayer.count >= self.replayer_initial_transitions:if done:self.noise.reset() #回合结束，重置噪声for batch in range(self.batches):observations,actions,rewards,next_observations,dones=self.replayer.sample(self.batch_size)#训练执行者observation_tensor=tf.convert_to_tensor(observations,dtype=tf.float32)with tf.GradientTape() as tape:action_tensor=self.actor_evaluate_net(observation_tensor)input_tensor=tf.concat([observation_tensor,action_tensor],axis=1)q_tensor=self.critic0_evaluate_net(input_tensor)loss_tensor=-tf.reduce_mean(q_tensor)grad_tensor=tape.gradient(loss_tensor,self.actor_evaluate_net.Variables)self.actor_evaluate_net.optimizer.apply_gradients(zip(grad_tensors,self.actor_evaluate_net.Variables))#训练评论者，这部分逻辑和DDPGAgent不同next_actions=self.actor_target_net.predict(next_observations)observation_actions=np.hstack([observations,actions])next_observation_actions=np.hstack([next_observations,next_actions])next_q0s=self.critic0_target_net.predict(next_observation_actions)[:,0]next_q1s=self.critic1_target_net.predict(next_observation_actions)[:,0]next_qs=np.minimum(next_q0s,next_q1s) #取其中较小值targets=rewards + self.gamma * next_qs * (1-dones)self.critic0_evaluate_net.fit(observation_actions,targets[:,np.newaxis],verbose=0)#np.newaxis添加新的一维self.critic1_evaluate_net.fit(observation_actions,targets[:,np.newaxis],verbose=0)self.update_target_net(self.actor_target_net,self.actor_evaluate_net,self.net_learning_rate)self.update_target_net(self.critic0_target_net,self.critic0_evaluate_net,self.net_learning_rate)self.update_target_net(self.critic1_target_net,self.critic1_evaluate_net,self.net_learning_rate)

def play_qlearning(env,agent,train=False,render=False):              #环境与智能体的交互episode_reward=0state=env.reset()comb1 = np.hstack((state[0],state[1]))comb2 = np.hstack((comb1,state[2]))observation=comb2while True:action=agent.decide(observation)action_original= action / 100      #由于在decide函数中，将动作放大了100倍，所以在返回环境中，动作应该缩小100倍，以保持真实的动作state_original,reward_original,done,_=env.step(action_original)comb1=np.hstack((state_original[0],state_original[1]))comb2=np.hstack((comb1,state_original[2]))next_observation=comb2reward=100 * reward_original          episode_reward+=rewardif train:agent.learn(observation,action,reward,next_observation,done)if done:breakobservation=next_observation                                  #注意这里不需要传入下一个动作这个参数，因为是取期望return episode_reward

强化学习算法求解借贷问题相关推荐

基于值的深度强化学习算法
目录 DQN2013 -- Playing Atari with Deep Reinforcement Learning DQN2015 -- Human-level control through ...
目前最好用的大规模强化学习算法训练库是什么？
点击蓝字关注我们本文整理自知乎问答,仅用于学术分享,著作权归作者所有.如有侵权,请联系后台作删文处理. 本文精选知乎问题"目前最好用的大规模强化学习算法训练库是什么?"评论区 ...
只用1/500数据就打败人类！一种采样高效的强化学习算法 | 报告详解
[栏目:前沿进展]近日,清华大学交叉信息研究院高阳研究组在强化学习领域中取得突破,研究组所提出的模型EfficientZero首次在雅达利(Atari )游戏数据上超过同等游戏时长的人类平均水平.Ef ...
上交张伟楠副教授：基于模型的强化学习算法，基本原理以及前沿进展（附视频）
2020 北京智源大会本文属于2020北京智源大会嘉宾演讲的整理报道系列.北京智源大会是北京智源人工智能研究院主办的年度国际性人工智能高端学术交流活动,以国际性.权威性.专业性和前瞻性的" ...
17种深度强化学习算法用Pytorch实现（附链接）
来源:新智元本文约1300字,建议阅读5分钟. 本文为你介绍一个用PyTorch实现了17种深度强化学习算法的教程和代码库,帮助大家在实践中理解深度RL算法. [ 导读 ]深度强化学习已经在许多领域 ...
【重磅】Tensorflow2.0实现29种深度强化学习算法大汇总
点击上方,选择星标或置顶,不定期资源大放送! 阅读大概需要3分钟 Follow小博主,每天更新前沿干货来源:深度强化学习实验室作者:王健树 [导读]今天给大家推荐一个超赞的强化学习项目资料,该项目 ...
探索强化学习算法背后的思想起源！
https://www.toutiao.com/a6630657888442384909/ 接受生物大脑的混乱和电子大脑的秩序人们对人工智能的追求总是与另一场斗争交织在一起,更富有哲理.更浪漫.更不 ...
谷歌实现2种新的强化学习算法，“比肩”DQN，泛化性能更佳！｜ICLR 2021
丰色发自凹非寺量子位报道 | 公众号 QbitAI 强化学习(RL)算法持续"进化"中-- 来自Google Research的研究人员,证明可以使用图表示 (graph ...
近端策略优化深度强化学习算法
PPO:Proximal Policy Optimization Algorithms,其优化的核心目标是: ppo paper 策略梯度以下是马尔可夫决策过程MDP的相关基础以及强化学习的优化目标 ...

强化学习算法求解借贷问题

强化学习

构建环境

强化学习求解

强化学习算法求解借贷问题相关推荐

最新文章

热门文章