0

我正在尝试将 q-learning 应用于代表储能套利的自定义强化学习环境（用电池进行电力交易，价格低时充电，价格上涨时放电）。环境有效，但我无法对其应用 q-learning。环境下方是一个能够运行环境的脚本，但我不确定我应该将状态变量设为什么。关于如何应用 q-learning 来优化充电/放电循环的任何想法？重置功能从具有每小时电价的数据集开始第二天。数据框的图片如下。

类 BatteryEnv(gym.Env):

def __init__(self, df):

    self.dict_actions = {0:'discharge',1:'charge',2:'wait'}
    self.df = df
    self.action_space = spaces.Discrete(3)
    self.observation_space = spaces.Box(low=0, high=100, shape=(1,1))
    
    self.reward_list = []
    self.actual_load_list = []#observations
    self.SOE_list=[] #State of energy 
  
    self.state_idx = 0 #iteration (hour of the day)
    self.SOE = 0 #SOE
    self.MAX_charge = 20 #C-rate kinda
    self.Capacity =100
    

def step(self, action): 
    #mapping integer to action for actual load calculation
    str_action = self.dict_actions[action]
    
    #increase state idx within episode (1= 1 hour)
    self.state_idx+=1  
    
    #calculating our actual load
    if str_action == 'charge' and self.SOE < self.Capacity:
        SOE_charge = np.clip(self.Capacity - self.SOE, 0, self.MAX_charge)
        self.SOE += SOE_charge
        obs = SOE_charge * self.df['prices'][self.state_idx]
        
    elif str_action == 'discharge' and self.SOE > 0:
        SOE_discharge = np.clip(self.SOE, 0, self.MAX_charge)
        self.SOE -= SOE_discharge
        obs = -SOE_discharge * self.df['prices'][self.state_idx]

        
    else:
        self.SOE += 0
        obs = 0 * self.df['prices'][self.state_idx]

    
    # appending actual load to list for monitoring and comparison purposes
    self.actual_load_list.append(obs)
    self.SOE_list.append(self.SOE)
    
    #reward system
    if obs<0: #if observation is positive we spending money. if negative we earning
        reward =1
    else:
        reward =-1
    
    # appending curr reward to list for monitoring and comparison purposes
    self.reward_list.append(reward) 

    #checking whether our episode (day interval) ends
    if self.df.iloc[self.state_idx,:].Daynum != self.df.iloc[self.state_idx-1].Daynum: 
        done = True
    else:
        done = False
        
    return obs, reward, done
    
def reset(self): 
    return df.iloc[self.state_idx,:]

def render():
    pass

以下代码能够表明环境正在运行。

for episode in range(7):
observation = env.reset()
for t in range(24): #can't be smaller than 24 as 24 time points equal to 1 episode (1 day)
    #print(observation)
    action = env.action_space.sample() #random actions
    observation, reward, done = env.step(action)
    if done:
        print("Episode finished after {} timesteps".format(t+1)), print (observation), print(reward)
        break

每个时间步长为一小时，价格为该小时的电价，daynum 为 365 中的天数

4

1 回答 1

0

我认为我能够使代码与 Q-learning 一起工作。但是，奖励和重置功能需要一些工作才能更好地执行。

class BatteryEnv(gym.Env):

def __init__(self, prices = np.array(df.prices), daynum = np.array(df.Daynum)):
    
    #self.df = df
    
    self.prices = prices
    self.daynum = daynum
    
    self.dict_actions = {0:'discharge',1:'charge',2:'wait'}

    self.action_space = spaces.Discrete(3)
    
    # our observation space is just one float value - our load 
    self.observation_space = spaces.Box(low=0, high=100, shape=(1,1))
    
    # reward list for monitoring
    self.reward_list = []

    # lists 4 monitoring
    self.actual_load_list = []
    self.SOE_list=[] #State of energy 
    self.chargio = [] #charge & discharge
    self.SOEe=[] #State of energy
    
    # index of current state within current episode
    self.state_idx = 0 #iteration
    self.SOE = 0 #SOE
    self.MAX_charge = 20 #C-rate kinda
    self.Capacity =100
    
    self.state = 0
    

def step(self, action): 
    #mapping integer to action for actual load calculation
    str_action = self.dict_actions[action]
    
    #increase state idx within episode (day)
    self.state_idx+=1  
    
    #calculating our actual load
    if str_action == 'charge' and self.SOE < self.Capacity:
        SOE_charge = np.clip(self.Capacity - self.SOE, 0, self.MAX_charge)
        self.state += SOE_charge
        self.SOEe.append(self.SOE)
        self.chargio.append(SOE_charge)
        obs = SOE_charge * self.prices[self.state_idx]
        
    elif str_action == 'discharge' and self.SOE > 0:
        SOE_discharge = np.clip(self.SOE, 0, self.MAX_charge)
        self.state -= SOE_discharge
        self.SOEe.append(self.SOE)
        self.chargio.append(-SOE_discharge)
        obs = -SOE_discharge * self.prices[self.state_idx]

        
    else:
        self.state += 0
        self.chargio.append(0)
        self.SOEe.append(self.SOE)
        obs = 0

    
    # appending actual load to list for monitoring and comparison purposes
    self.actual_load_list.append(obs)
    self.SOE_list.append(self.SOE)

    
    #reward system
    if obs<0: #if observation is positive we spending money. if negative we earning
        reward =1
    else:
        reward =-1
    
    # appending curr reward to list for monitoring and comparison purposes
    self.reward_list.append(reward) 

    #checking whether our episode (day interval) ends
    if self.daynum[self.state_idx] != self.daynum[self.state_idx-1]: 
        done = True
    else:
        done = False
            
        
    info = {
        #'step': self.state_idx,
        'SOE': self.SOE,
        #'reward': reward,
        'chargio': self.chargio
            }
        
    return obs, reward, done, info
    

def reset(self):
    self.state = 0
    return self.state


def render():
    pass

应用 q-learning：

env.reset()
env = BatteryEnv()
discrete_os_size = [20] * len(env.observation_space.high)
discrete_os_win_size = (env.observation_space.high - 
env.observation_space.low)/discrete_os_size
discrete_os_win_size #buckets of 10
learning_rate = 0.1
discount =0.95 #measure of how important future actions are
episodes =25000
q_table = np.random.uniform(low=-2, high=2, size=(discrete_os_size + [env.action_space.n]))

def get_discrete_state(state): #change SOE for other states
discrete_state = (state - env.observation_space.low)/discrete_os_win_size
return tuple(discrete_state.astype(np.int))

discrete_state =get_discrete_state(env.reset())

SOE=[]

for episode in range (episodes):
if episode % 5000 ==0:
    print(episode)

discrete_state =get_discrete_state(env.reset())
done = False
while not done:
    action = np.argmax(q_table[discrete_state])
    new_state, reward, done, _ =env.step(action)

    new_discrete_state = get_discrete_state(new_state)

    if not done:
        max_future_q = np.max(q_table[new_discrete_state])
        current_q = q_table[discrete_state + (action,)]
        new_q = (1-learning_rate) * current_q + learning_rate *(reward + discount * max_future_q)
        q_table[discrete_state +(action,)] = new_q

    #elif new_state[0] >= env.go:

    discrete_state = new_discrete_state
    
    SOE.append(new_state)



    print(reward, new_state)
于 2021-07-04T12:27:41.997 回答