# define action map action_map = { 1: 1/12, # Rule for 1 AM 2: 1/12, # Rule for 2 AM 3: 1/12, # Rule for 3 AM 4: 1/12, # Rule for 4 AM 5: 1/12, # Rule for 5 AM 6: 1/12, # Rule for 6 AM 7: 1/12, # Rule for 7 AM 8: 1/12, # Rule for 8 AM 9: 1/12, # Rule for 9 AM 10: 1/12, # Rule for 10 AM 11: 1/12, # Rule for 11 AM 12: 1/12, # Rule for 12 PM 13: -1/12, # Rule for 1 PM 14: -1/12, # Rule for 2 PM 15: -1/12, # Rule for 3 PM 16: -1/12, # Rule for 4 PM 17: -1/12, # Rule for 5 PM 18: -1/12, # Rule for 6 PM 19: -1/12, # Rule for 7 PM 20: -1/12, # Rule for 8 PM 21: -1/12, # Rule for 9 PM 22: -1/12, # Rule for 10 PM 23: -1/12, # Rule for 11 PM 24: -1/12, # Rule for 12 AM }
# display simulation summary plot_simulation_summary({ 'Baseline': baseline_env, # Uncomment line below if you have completed Exercise 1 'Random': random_env, 'RBC': rbc_env, })
其他的几乎一样，就是类名字略有不同
Q- learning
原理
不多说算法原理了，给个p-code
The general Q-Learning algorithm is as follows:
Initialize the Q-table for all state-action pairs.
Set the learning rate $\alpha$ ($0 < \alpha < 1$) and the discount factor $\gamma$ ($0 < \gamma < 1$).
Repeat the following steps for each episode:
Observe the initial state $s$.
Choose an action $a$ based on the epsilon-greedy policy (a random action is chosen with probability epsilon, $\epsilon$ and the action with the highest Q-value is chosen with probability $1 - \epsilon$).
Take the action $a$ and observe the next state $s'$ and the reward $r$.
Update the Q-value of the state-action pair $(s,a)$ using the Bellman equation.
Set $s = s'$.
Repeat step 3 for a large number of episodes or until convergence is reached.
# ----------------- CALCULATE NUMBER OF TRAINING EPISODES ----------------- i = 3# 一个qtable更新n次 m = tql_env.observation_space[0].n # 24小时 n = tql_env.action_space[0].n # 12个动作 t = tql_env.unwrapped.time_steps - 1 tql_episodes = m*n*i/t tql_episodes = int(tql_episodes) print('Q-Table dimension:', (m, n)) print('Number of episodes to train:', tql_episodes)
# ----------------------- SET MODEL HYPERPARAMETERS ----------------------- tql_kwargs = { 'epsilon': 1.0, 'minimum_epsilon': 0.01, 'epsilon_decay': 0.0001, 'learning_rate': 0.005, 'discount_factor': 0.99, }
# ----------------- CALCULATE NUMBER OF TRAINING EPISODES ----------------- fraction = 0.25 sac_episodes = int(tql_episodes*fraction) print('Fraction of Tabular Q-Learning episodes used:', fraction) print('Number of episodes to train:', sac_episodes) sac_episode_timesteps = sac_env.unwrapped.time_steps - 1 sac_total_timesteps = sac_episodes*sac_episode_timesteps
# ------------------------------- TRAIN MODEL ----------------------------- for i in tqdm(range(sac_episodes)): sac_model = sac_model.learn( total_timesteps=sac_episode_timesteps, reset_num_timesteps=False, )
classCustomReward(RewardFunction): def__init__(self, env_metadata: dict[str, Any]): r"""Initialize CustomReward. Parameters ---------- env_metadata: dict[str, Any]: General static information about the environment. """
super().__init__(env_metadata)
defcalculate( self, observations: list[dict[str, int | float]] ) -> list[float]: r"""Returns reward for most recent action. The reward is designed to minimize electricity cost. It is calculated for each building, i and summed to provide the agent with a reward that is representative of all n buildings. It encourages net-zero energy use by penalizing grid load satisfaction when there is energy in the battery as well as penalizing net export when the battery is not fully charged through the penalty term. There is neither penalty nor reward when the battery is fully charged during net export to the grid. Whereas, when the battery is charged to capacity and there is net import from the grid the penalty is maximized. Parameters ---------- observations: list[dict[str, int | float]] List of all building observations at current :py:attr:`citylearn.citylearn.CityLearnEnv.time_step` that are got from calling :py:meth:`citylearn.building.Building.observations`. Returns ------- reward: list[float] Reward for transition to current timestep. """
reward_list = []
for o, m inzip(observations, self.env_metadata['buildings']): cost = o['net_electricity_consumption']*o['electricity_pricing'] battery_soc = o['electrical_storage_soc'] penalty = -(1.0 + np.sign(cost)*battery_soc) reward = penalty*abs(cost) reward_list.append(reward)
