Reward Function

A reward is returned at every step(). It defines the learning signal, whereas KPIs assess the completed episode. A reward and a reported KPI do not have to be the same quantity.

The default citylearn.reward_function.RewardFunction penalizes positive grid imports. Other supplied rewards cover objectives such as thermal comfort and EV charging. See citylearn.reward_function for constructors and formulas.

Choose a reward

Pass the reward class through CityLearnEnv(reward_function=...) or select an importable class in the schema:

{
  "reward_function": {
    "type": "citylearn.reward_function.RewardFunction"
  }
}

This is the reward section of a schema, not a complete dataset definition.

Define a custom reward

Subclass RewardFunction and implement calculate(observations). Return one reward per building for decentralized control, or one aggregate reward for centralized control. The inherited central_agent property exposes the environment setting.

The existing emissions-based example is:

from typing import Any, List, Mapping, Union
from citylearn.reward_function import RewardFunction

class CustomReward(RewardFunction):
    """Calculates custom user-defined multi-agent reward.
        
    Reward is the :py:attr:`net_electricity_consumption_emission`
    for entire district if central agent setup otherwise it is the
    :py:attr:`net_electricity_consumption_emission` each building.

    Parameters
    ----------
    env_metadata: Mapping[str, Any]:
        General static information about the environment.
    """
    
    def __init__(self, env_metadata: Mapping[str, Any]):
        super().__init__(env_metadata)
 
    def calculate(self, observations: List[Mapping[str, Union[int, float]]]) -> List[float]:
        r"""Calculates reward.

        Parameters
        ----------
        observations: List[Mapping[str, Union[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.
        """

        net_electricity_consumption_emission = [o['net_electricity_consumption_emission'] for o in observations]

        if self.central_agent:
            reward = [-sum(net_electricity_consumption_emission)]
        else:
            reward = [-v for v in net_electricity_consumption_emission]

        return reward

A reward can declare required_observation_names to receive a smaller observation payload. The Custom agents and rewards tutorial demonstrates this together with a complete simulation.