Common components and methods

Common components and methods#

Submodules#

assume.common.base module#

class assume.common.base.BaseStrategy(*args, **kwargs)#

Bases: object

A base class for a bidding strategy.

Parameters:

*args (list) – The arguments.
**kwargs (dict) – The keyword arguments.

calculate_bids(unit: BaseUnit, market_config: MarketConfig, product_tuples: list[Product], **kwargs) → list[Order]#

Calculates the bids for the next time step.

Parameters:

unit (BaseUnit) – The unit.
market_config (MarketConfig) – The market configuration.
product_tuples (list[Product]) – The product tuples.

Returns:

The bids.

Return type:

Orderbook

calculate_reward(unit: BaseUnit, marketconfig: MarketConfig, orderbook: list[Order])#

Calculates the reward for the given unit.

Parameters:

unit (BaseUnit) – The unit.
marketconfig (MarketConfig) – The market configuration.
orderbook (Orderbook) – The orderbook.

remove_empty_bids(bids: list) → list#

Removes empty bids from the orderbook. Use this method to clean the bids before submitting them to the market to speed up the market clearing process, and if zero volume bids are not required for the specific market.

Parameters:: bids (list) – The bids.
Returns:: The cleaned bids.
Return type:: list

class assume.common.base.BaseUnit(id: str, unit_operator: str, technology: str, bidding_strategies: dict[str, BaseStrategy], forecaster: UnitForecaster, node: str = 'node0', location: tuple[float, float] = (0.0, 0.0), **kwargs)#

Bases: object

A base class for a unit. This class is used as a foundation for all units.

Parameters:

id (str) – The ID of the unit.
unit_operator (str) – The operator of the unit.
technology (str) – The technology of the unit.
bidding_strategies (dict[str, BaseStrategy]) – The bidding strategies of the unit.
index (FastIndex) – The index of the unit.
node (str, optional) – The node of the unit. Defaults to “”.
forecaster (Forecaster, optional) – The forecast of the unit. Defaults to None.
location (tuple[float, float], optional) – The location of the unit. Defaults to (0.0, 0.0).
**kwargs – Additional keyword arguments.

as_dict() → dict[str, str | int]#

Returns a dictionary representation of the unit.

Returns:: A dictionary representation of the unit.

calculate_bids(market_config: MarketConfig, product_tuples: list[tuple]) → list[Order]#

Calculates the bids for the next time step.

Parameters:

market_config (MarketConfig) – The market configuration.
product_tuples (list[tuple]) – The product tuples.

Returns:

The bids.

Return type:

Orderbook

Raises:

KeyError – If the product type is not found in the bidding strategies.

calculate_cashflow(product_type: str, orderbook: list[Order])#

Calculates the cashflow for the given product type.

Parameters:

product_type – The product type.
orderbook – The orderbook.

calculate_cashflow_and_reward(marketconfig: MarketConfig, orderbook: list[Order]) → None#

Calculates the cashflow and the reward for the given unit.

Parameters:

marketconfig (MarketConfig) – The market configuration.
orderbook (Orderbook) – The orderbook.

calculate_generation_cost(start: datetime, end: datetime, product_type: str) → None#

Calculates the generation cost for a specific product type within the given time range, but only if the end is the last index in the time series.

Parameters:

start (datetime.datetime) – The start time for the calculation.
end (datetime.datetime) – The end time for the calculation.
product_type (str) – The type of product for which the generation cost is to be calculated.

calculate_marginal_cost(start: datetime, power: float) → float#

Calculates the marginal cost for the given power.`

Parameters:

start (datetime.datetime) – The start time of the dispatch.
power (float) – The power output of the unit.

Returns:

The marginal cost for the given power.

Return type:

float

execute_current_dispatch(start: datetime, end: datetime) → ndarray#

Checks if the total dispatch plan is feasible.

This method checks if the market feedback is feasible for the given unit and sets the closest dispatch if not. The end parameter should be inclusive.

Parameters:

start – The start time of the dispatch.
end – The end time of the dispatch.

Returns:

The volume of the unit within the given time range.

get_output_before(dt: datetime, product_type: str = 'energy') → float#

Returns output before the given datetime.

If the datetime is before the start of the index, 0 is returned.

Parameters:

dt – The datetime.
product_type – The product type (default is “energy”).

Returns:

The output before the given datetime.

get_starting_costs(op_time: int) → float#

Returns the costs if start-up is planned.

Parameters:: op_time – Operation time in hours running from get_operation_time.
Returns:: Start-up costs.

set_dispatch_plan(marketconfig: MarketConfig, orderbook: list[Order]) → None#

Iterates through the orderbook, adding the accepted volumes to the corresponding time slots in the dispatch plan.

Parameters:

marketconfig (MarketConfig) – The market configuration.
orderbook (Orderbook) – The orderbook.

class assume.common.base.ExchangeStrategy(*args, **kwargs)#: Bases: BaseStrategy

class assume.common.base.LearningConfig(learning_mode: bool = False, evaluation_mode: bool = False, continue_learning: bool = False, trained_policies_save_path: str | None = None, trained_policies_load_path: str | None = None, min_bid_price: float | None = -100.0, max_bid_price: float | None = 100.0, device: str = 'cpu', episodes_collecting_initial_experience: int = 5, exploration_noise_std: float = 0.2, training_episodes: int = 100, validation_episodes_interval: int = 5, train_freq: str = '24h', batch_size: int = 128, gradient_steps: int = 100, learning_rate: float = 0.001, learning_rate_schedule: str | None = None, early_stopping_steps: int | None = None, early_stopping_threshold: float = 0.05, algorithm: str = 'matd3', replay_buffer_size: int = 50000, gamma: float = 0.99, actor_architecture: str = 'mlp', policy_delay: int = 2, noise_sigma: float = 0.1, noise_scale: int = 1, noise_dt: int = 1, action_noise_schedule: str | None = None, tau: float = 0.005, target_policy_noise: float = 0.2, target_noise_clip: float = 0.5)#

Bases: object

A class for the learning configuration.

Parameters:

learning_mode (bool) – Should we use learning mode at all? If False, the learning bidding strategy is loaded from trained_policies_load_path and no training occurs. Default is False.
evaluation_mode (bool) – This setting is modified internally. Whether to run in evaluation mode. If True, the agent uses the learned policy without exploration noise and no training updates occur. Default is False.
continue_learning (bool) – Whether to use pre-learned strategies and then continue learning. If True, loads existing policies from trained_policies_load_path and continues training. Default is False.
trained_policies_save_path (str | None) – The directory path - relative to the scenario’s inputs_path - where newly trained RL policies (actor and critic networks) will be saved. Only needed when learning_mode is True. Value is set in setup_world(). Defaults to None.
trained_policies_load_path (str | None) – The directory path - relative to the scenario’s inputs_path - from which pre-trained policies should be loaded. Needed when continue_learning is True or using pre-trained strategies. Default is None.
min_bid_price (float | None) – The minimum bid price which limits the action of the actor to this price. Used to constrain the actor’s output to a realistic price range. Default is -100.0.
max_bid_price (float | None) – The maximum bid price which limits the action of the actor to this price. Used to constrain the actor’s output to a realistic price range. Default is 100.0.
device (str) – The device to use for PyTorch computations. Options include “cpu”, “cuda”, or specific CUDA devices like “cuda:0”. Default is “cpu”.
episodes_collecting_initial_experience (int) – The number of episodes at the start during which random actions are chosen instead of using the actor network. This helps populate the replay buffer with diverse experiences. Default is 5.
exploration_noise_std (float) – The standard deviation of Gaussian noise added to actions during exploration in the environment. Higher values encourage more exploration. Default is 0.2.
training_episodes (int) – The number of training episodes, where one episode is the entire simulation horizon specified in the general config. Default is 100.
validation_episodes_interval (int) – The interval (in episodes) at which validation episodes are run to evaluate the current policy’s performance without training updates. Default is 5.
train_freq (str) – Defines the frequency in time steps at which the actor and critic networks are updated. Accepts time strings like “24h” for 24 hours or “1d” for 1 day. Default is “24h”.
batch_size (int) – The batch size of experiences sampled from the replay buffer for each training update. Larger batches provide more stable gradients but require more memory. In environments with many leanring agents we advise small batch sizes. Default is 128.
gradient_steps (int) – The number of gradient descent steps performed during each training update. More steps can lead to better learning but increase computation time. Default is 100.
learning_rate (float) – The learning rate (step size) for the optimizer, which controls how much the policy and value networks are updated during training. Default is 0.001.
learning_rate_schedule (str | None) – Which learning rate decay schedule to use. Currently only “linear” decay is available, which linearly decreases the learning rate over time. Default is None (constant learning rate).
early_stopping_steps (int | None) – The number of validation steps over which the moving average reward is calculated for early stopping. If the reward doesn’t change by early_stopping_threshold over this many steps, training stops. If None, defaults to training_episodes / validation_episodes_interval + 1.
early_stopping_threshold (float) – The minimum improvement in moving average reward required to avoid early stopping. If the reward improvement is less than this threshold over early_stopping_steps, training is terminated early. Default is 0.05.
algorithm (str) – Specifies which reinforcement learning algorithm to use. Currently, only “matd3” (Multi-Agent Twin Delayed Deep Deterministic Policy Gradient) is implemented. Default is “matd3”.
replay_buffer_size (int) – The maximum number of transitions stored in the replay buffer for experience replay. Larger buffers allow for more diverse training samples. Default is 500000.
gamma (float) – The discount factor for future rewards, ranging from 0 to 1. Higher values give more weight to long-term rewards in decision-making. Default is 0.99.
actor_architecture (str) – The architecture of the neural networks used for the actors. Options include “mlp” (Multi-Layer Perceptron) and “lstm” (Long Short-Term Memory). Default is “mlp”.
policy_delay (int) – The frequency (in gradient steps) at which the actor policy is updated. TD3 updates the critic more frequently than the actor to stabilize training. Default is 2.
noise_sigma (float) – The standard deviation of the Ornstein-Uhlenbeck or Gaussian noise distribution used to generate exploration noise added to actions. Default is 0.1.
noise_scale (int) – The scale factor multiplied by the noise drawn from the distribution. Larger values increase exploration. Default is 1.
noise_dt (int) – The time step parameter for the Ornstein-Uhlenbeck process, which determines how quickly the noise decays over time. Used for noise scheduling. Default is 1.
action_noise_schedule (str | None) – Which action noise decay schedule to use. Currently only “linear” decay is available, which linearly decreases exploration noise over training. Default is “linear”.
tau (float) – The soft update coefficient for updating target networks. Controls how slowly target networks track the main networks. Smaller values mean slower updates. Default is 0.005.
target_policy_noise (float) – The standard deviation of noise added to target policy actions during critic updates. This smoothing helps prevent overfitting to narrow policy peaks. Default is 0.2.
target_noise_clip (float) – The maximum absolute value for clipping the target policy noise. Prevents the noise from being too large. Default is 0.5.

action_noise_schedule: str | None = None#

actor_architecture: str = 'mlp'#

algorithm: str = 'matd3'#

batch_size: int = 128#

continue_learning: bool = False#

device: str = 'cpu'#

early_stopping_steps: int | None = None#

early_stopping_threshold: float = 0.05#

episodes_collecting_initial_experience: int = 5#

evaluation_mode: bool = False#

exploration_noise_std: float = 0.2#

gamma: float = 0.99#

gradient_steps: int = 100#

learning_mode: bool = False#

learning_rate: float = 0.001#

learning_rate_schedule: str | None = None#

max_bid_price: float | None = 100.0#

min_bid_price: float | None = -100.0#

noise_dt: int = 1#

noise_scale: int = 1#

noise_sigma: float = 0.1#

policy_delay: int = 2#

replay_buffer_size: int = 50000#

target_noise_clip: float = 0.5#

target_policy_noise: float = 0.2#

tau: float = 0.005#

train_freq: str = '24h'#

trained_policies_load_path: str | None = None#

trained_policies_save_path: str | None = None#

training_episodes: int = 100#

validation_episodes_interval: int = 5#

class assume.common.base.LearningStrategy(learning_role, foresight: int, act_dim: int, unique_obs_dim: int, num_timeseries_obs_dim: int = 3, *args, **kwargs)#

Bases: BaseStrategy

A strategy which provides learning functionality, has a method to calculate the reward.

It is important to keep in mind, that the DRL method and the centralized critic relies on unique observations of individual units. The algorithm is designed in such a way, that the unique observations are always placed at the end of the observation space. Please follow this convention when designing your create_observation method and the observation space.

foresight#

Number of steps of for- and backwards looking in observations.

Type:: int

act_dim#

The action dimension.

Type:: int

unique_obs_dim#

The unique observation dimension.

Type:: int

num_timeseries_obs_dim#

The number of observation timeseries dimension.

Type:: int

learning_role#

The learning role orchestrating the learning.

Type:: Learning

Parameters:

*args (list) – The arguments.
**kwargs (dict) – The keyword arguments.

class assume.common.base.MinMaxChargeStrategy(*args, **kwargs)#: Bases: BaseStrategy

class assume.common.base.MinMaxStrategy(*args, **kwargs)#: Bases: BaseStrategy

class assume.common.base.SupportsMinMax(**kwargs)#

Bases: BaseUnit

Base class used for units supporting continuous dispatch and without energy storage. This class is best to be used as foundation for classes of power plants and similar units.

calculate_min_max_power(start: datetime, end: datetime, product_type: str = 'energy') → tuple[ndarray, ndarray]#

Calculates the min and max power for the given time period.

Parameters:

start (datetime.datetime) – The start time of the dispatch.
end (datetime.datetime) – The end time of the dispatch.
product_type (str) – The product type of the unit.

Returns:

The min and max power for the given time period.

Return type:

tuple[np.ndarray, np.ndarray]

calculate_ramp(op_time: int, previous_power: float, power: float, current_power: float = 0) → float#

Corrects the possible power to offer according to ramping restrictions.

Parameters:

op_time (int) – The operation time.
previous_power (float) – The previous power output of the unit.
power (float) – The planned power offer of the unit.
current_power (float) – The current power output of the unit.

Returns:

The corrected possible power to offer according to ramping restrictions.

Return type:

float

efficiency: float#

emission_factor: float#

get_operation_time(start: datetime) → int#

Returns the time the unit is operating (positive) or shut down (negative).

Parameters:: start (datetime.datetime) – The start time.
Returns:: The operation time as a positive integer if operating, or negative if shut down.
Return type:: int

get_starting_costs(op_time: int) → float#

Returns the start-up cost for the given operation time. If operation time is positive, the unit is running, so no start-up costs are returned. If operation time is negative, the unit is not running, so start-up costs are returned according to the start-up costs of the unit and the hot/warm/cold start times.

Parameters:: op_time (int) – The operation time.
Returns:: The start-up costs depending on the down time.
Return type:: float

max_power: float#

min_down_time: int = 0#

min_operating_time: int = 0#

min_power: float#

ramp_down: float = None#

ramp_up: float = None#

class assume.common.base.SupportsMinMaxCharge(**kwargs)#

Bases: BaseUnit

Base Class used for units with energy storage.

The volume is always the amount of energy which is put on (if positive) the market or (if negative) taken from the market. A demand does always have a negative volume - as it buys/consumes energy . A powerplant does always have a positive volume - as it produces energy.

All charging related params are negative, as charging is a demand. This is special for SupportsMinMaxCharge as both charge and discharge is available.

calculate_min_max_charge(start: datetime, end: datetime, soc: float = None) → tuple[ndarray, ndarray]#

Calculates the min and max charging power for the given time period.

Parameters:

start (datetime.datetime) – The start time of the dispatch.
end (datetime.datetime) – The end time of the dispatch.
soc (float, optional) – The current state-of-charge (between 0 and 1). Defaults to None.

Returns:

The min and max charging power for the given time period.

Return type:

tuple[np.ndarray, np.ndarray]

calculate_min_max_discharge(start: datetime, end: datetime, soc: float = None) → tuple[ndarray, ndarray]#

Calculates the min and max discharging power for the given time period.

Parameters:

start (datetime.datetime) – The start time of the dispatch.
end (datetime.datetime) – The end time of the dispatch.
soc (float, optional) – The current state-of-charge (between 0 and 1). Defaults to None.

Returns:

The min and max discharging power for the given time period.

Return type:

tuple[np.ndarray, np.ndarray]

calculate_ramp_charge(previous_power: float, power_charge: float, current_power: float = 0) → float#

Adjusts the charging power to the ramping constraints.

Parameters:

previous_power (float) – The previous power output of the unit.
power_charge (float) – The charging power output of the unit.
current_power (float, optional) – The current power output of the unit. Defaults to 0.

Returns:

The charging power adjusted to the ramping constraints.

Return type:

float

calculate_ramp_discharge(previous_power: float, power_discharge: float, current_power: float = 0) → float#

Adjusts the discharging power to the ramping constraints. Given previous_power in the step before, we want to offer power_discharge (positive) to some market, while we already sold current_power on some other market.

power_discharge is a power delta to our current output, this function checks if this amount is still feasible to sell.

Parameters:

previous_power (float) – The previous power output of the unit.
power_discharge (float) – The discharging power output of the unit.
current_power (float, optional) – The current power output of the unit already sold on another market. Defaults to 0.

Returns:

The discharging power adjusted to the ramping constraints.

Return type:

float

capacity: float#

efficiency_charge: float#

efficiency_discharge: float#

initial_soc: float#

max_power_charge: float#

max_power_discharge: float#

min_power_charge: float#

min_power_discharge: float#

ramp_down_charge: float | None#

ramp_down_discharge: float | None#

ramp_up_charge: float | None#

ramp_up_discharge: float | None#

set_dispatch_plan(marketconfig: MarketConfig, orderbook: list[Order]) → None#: Updates the SOC for storage units.

assume.common.exceptions module#

exception assume.common.exceptions.AssumeException#: Bases: Exception

exception assume.common.exceptions.InvalidTypeException#: Bases: AssumeException

assume.common.mango_serializer module#

assume.common.mango_serializer.datetime_json_serializer()#

assume.common.mango_serializer.generic_json_serializer()#

assume.common.mango_serializer.generic_pb_serializer()#

assume.common.mango_serializer.mango_codec_factory()#

assume.common.market_objects module#

class assume.common.market_objects.ClearingMessage#

Bases: TypedDict

Message which is sent from the market to agents to clear a market.

Parameters:

context (str) – the context of the message
market_id (str) – the id of the market
accepted_orders (Orderbook) – the orders accepted by the market
rejected_orders (Orderbook) – the orders rejected by the market

accepted_orders: list[Order]#

context: str#

market_id: str#

rejected_orders: list[Order]#

class assume.common.market_objects.DataRequestMessage#

Bases: TypedDict

Message for requesting data from a market.

Parameters:

context (str) – the context of the message
market_id (str) – the id of the market
metric (str) – the specific metric being requested
start_time (datetime.datetime) – the start time of the data request
end_time (datetime.datetime) – the end time of the data request

context: str#

end_time: datetime#

market_id: str#

metric: str#

start_time: datetime#

class assume.common.market_objects.MarketConfig(market_id: str = 'market', opening_hours: ~dateutil.rrule.rrule = <dateutil.rrule.rrule object>, opening_duration: ~datetime.timedelta = datetime.timedelta(seconds=3600), market_mechanism: str = 'pay_as_clear', market_products: list[~assume.common.market_objects.MarketProduct] = <factory>, product_type: str = 'energy', maximum_bid_volume: float | None = 2000.0, maximum_bid_price: float | None = 3000.0, minimum_bid_price: float = -500.0, maximum_gradient: float | None = None, additional_fields: list[str] = <factory>, volume_unit: str = 'MW', volume_tick: float | None = None, price_unit: str = '€/MWh', price_tick: float | None = None, supports_get_unmatched: bool = False, eligible_obligations_lambda: ~collections.abc.Callable[[~mango.agent.core.Agent], bool] = <function MarketConfig.<lambda>>, param_dict: dict = <factory>, addr: ~mango.messages.message.AgentAddress | None = None)#

Bases: object

Describes the configuration of a market.

Parameters:

market_id (str) – the ID of the market
opening_hours (dateutil.rrule.rrule) – the opening hours of the market
opening_duration (datetime.timedelta) – the duration of the opening hours
market_mechanism (str) – name of method used for clearing
market_products (list[MarketProduct]) – list of available products to be traded at the market
product_type (str) – energy or capacity or heat
maximum_bid_volume (float | None) – the maximum valid bid volume of the market
maximum_bid_price (float | None) – the maximum bid price of the market
minimum_bid_price (float) – the minimum bid price of the market
maximum_gradient (float | None) – max allowed change between bids
additional_fields (list[str]) – additional fields of the market
volume_unit (str) – the volume unit of the market (e.g. MW)
volume_tick (float | None) – step increments of volume (e.g. 0.1)
price_unit (str) – the price unit of the market (e.g. €/MWh)
price_tick (float | None) – step increments of price (e.g. 0.1)
supports_get_unmatched (bool) – whether the market supports get unmatched
eligible_obligations_lambda (eligible_lambda) – lambda function which determines if an agent is eligible to trade this product
param_dict (dict) – dict which can have additional params for the used clearing_mechanism
addr (str) – the address of the market
aid (str) – automatic id of the market

additional_fields: list[str]#

addr: AgentAddress | None = None#

eligible_obligations_lambda()#

market_id: str = 'market'#

market_mechanism: str = 'pay_as_clear'#

market_products: list[MarketProduct]#

maximum_bid_price: float | None = 3000.0#

maximum_bid_volume: float | None = 2000.0#

maximum_gradient: float | None = None#

minimum_bid_price: float = -500.0#

opening_duration: timedelta = datetime.timedelta(seconds=3600)#

opening_hours: rrule = <dateutil.rrule.rrule object>#

param_dict: dict#

price_tick: float | None = None#

price_unit: str = '€/MWh'#

product_type: str = 'energy'#

supports_get_unmatched: bool = False#

volume_tick: float | None = None#

volume_unit: str = 'MW'#

class assume.common.market_objects.MarketProduct(duration: relativedelta | rrule, count: int, first_delivery: relativedelta = relativedelta(), only_hours: OnlyHours | None = None, eligible_lambda_function: Callable[[Agent], bool] | None = None)#

Bases: object

Describes the configuration of a market product which is available at a market.

Parameters:

duration (dateutil.relativedelta.relativedelta | dateutil.rrule.rrule) – the duration of the product
count (int) – how many future durations can be traded, must be >= 1
first_delivery (dateutil.relativedelta.relativedelta) – when does the first delivery begin, in relation to market start
only_hours (OnlyHours | None) – tuple of hours from which this order is available, on multi day products
eligible_lambda_function (eligible_lambda | None) – lambda function which determines if an agent is eligible to trade this product

count: int#

duration: relativedelta | rrule#

eligible_lambda_function: Callable[[Agent], bool] | None = None#

first_delivery: relativedelta = relativedelta()#

only_hours: OnlyHours | None = None#

class assume.common.market_objects.MetaDict#

Bases: TypedDict

Message Meta of a FIPA ACL Message.

Parameters:

sender_addr (str | list) – the address of the sender
sender_id (str) – the id of the sender
reply_to (str) – to which agent follow up messages should be sent
conversation_id (str) – the id of the conversation
performative (str) – the performative of the message
protocol (str) – the protocol used
language (str) – the language used
encoding (str) – the encoding used
ontology (str) – the ontology used
reply_with (str) – what the answer should contain as in_reply_to
in_reply_to (str) – str used to reference an earlier action
reply_by (str) – latest time to accept replies

conversation_id: str#

encoding: str#

in_reply_to: str#

language: str#

ontology: str#

performative: str#

protocol: str#

reply_by: str#

reply_to: str#

reply_with: str#

sender_addr: str | list#

sender_id: str#

class assume.common.market_objects.OnlyHours(begin_hour: int, end_hour: int)#

Bases: NamedTuple

Used for peak and off-peak bids. Allows to set a begin and hour of the peak.

For example OnlyHours(8,16) would be used to have a bid which is valid for every day from 8 to 16 a clock.

begin_hour: int#: Alias for field number 0

end_hour: int#: Alias for field number 1

class assume.common.market_objects.OpeningMessage#

Bases: TypedDict

Message which is sent from the market to participating agent to open a market.

Parameters:

context (str) – the context of the message
market_id (str) – the id of the market
start_time (float) – the start time of the market
end_time (float) – the stop time of the market
products (list[Product]) – list of products which are available at the market to be traded

context: str#

end_time: float#

market_id: str#

products: list[Product]#

start_time: float#

class assume.common.market_objects.Order#

Bases: TypedDict

Describes an order which can be either generation (volume > 0) or demand (volume < 0)

Parameters:

bid_id (str) – the id of the bid
start_time (datetime.datetime) – the start time of the order
end_time (datetime.datetime) – the end time of the order
volume (Number | dict[datetime, Number]) – the volume of the order (positive if generation)
accepted_volume (Number | dict[datetime, Number]) – the accepted volume of the order
price (Number) – the price of the order
accepted_price (Number | dict[datetime, Number]) – the accepted price of the order
agent_addr (str) – the address of the agent
node (str) – the node of market where the order is placed
only_hours (OnlyHours | None) – tuple of hours from which this order is available, on multi day products

accepted_price: Number | dict[datetime, Number]#

accepted_volume: Number | dict[datetime, Number]#

agent_addr: str#

bid_id: str#

end_time: datetime#

node: str#

only_hours: OnlyHours | None#

price: Number#

start_time: datetime#

volume: Number | dict[datetime, Number]#

class assume.common.market_objects.OrderBookMessage#

Bases: TypedDict

Message containing the order book of a market.

Parameters:

context (str) – the context of the message
market_id (str) – the id of the market
orderbook (Orderbook) – the order book of the market

context: str#

market_id: str#

orderbook: list[Order]#

class assume.common.market_objects.Product(start: datetime, end: datetime, only_hours: OnlyHours | None = None)#

Bases: NamedTuple

An actual product with start and end.

Parameters:

start (datetime.datetime) – the start time of the product
end (datetime.datetime) – the end time of the product
only_hours (OnlyHours | None) – tuple of hours from which this order is available, on multi day products

end: datetime#: Alias for field number 1

only_hours: OnlyHours | None#: Alias for field number 2

start: datetime#: Alias for field number 0

class assume.common.market_objects.RegistrationMessage#

Bases: TypedDict

Message for agent registration at a market.

Parameters:

context (str) – the context of the message
market_id (str) – the id of the market
information (dict) – additional information for registration

context: str#

information: dict#

market_id: str#

class assume.common.market_objects.RegistrationReplyMessage#

Bases: TypedDict

Reply message for agent registration at a market.

Parameters:

context (str) – the context of the message
market_id (str) – the id of the market
accepted (bool) – whether the registration is accepted

accepted: bool#

context: str#

market_id: str#

assume.common.market_objects.only_co2emissionless(unit)#: same as only_renewables plus nuclear

assume.common.market_objects.only_renewables(unit)#: check that technology is renewable. This allows all names containing wind, solar, bio or demand

assume.common.market_objects.power_plant_not_negative(unit)#

assume.common.outputs module#

class assume.common.outputs.DatabaseMaintenance(db_uri: str)#

Bases: object

A utility class for managing simulation data stored in a database.

This class creates a database engine from a provided URI and offers methods to:

Retrieve a list of unique simulation IDs across all tables.
Delete specific simulations from every table.
Delete all simulations, or all except those specified, across every table.

It assumes that each table (except for system tables like “spatial_ref_sys”) contains a column named ‘simulation’ that uniquely identifies the simulation.

Parameters:: db_uri (str) – The URI of the database engine used to create a SQLAlchemy engine.

delete_all_simulations(exclude: list[str] = None) → None#

Deletes all simulation records from every table, with an option to exclude specific simulations.

If an exclusion list is provided, simulations with those IDs will not be deleted. Otherwise, all simulation records are removed from all tables (excluding system tables).

Parameters:: exclude (list[str], optional) – A list of simulation IDs that should NOT be deleted. If None, all simulation records are deleted.

delete_simulations(simulation_ids: list[str]) → None#

Deletes specific simulation records from all tables.

This method deletes rows from every table where the ‘simulation’ column matches any of the provided simulation IDs. An index is created on the simulation column to optimize the deletion, if one does not already exist.

Parameters:: simulation_ids (list[str]) – A list of simulation IDs to delete.

get_unique_simulation_ids() → list[str]#

Retrieves a list of unique simulation IDs found in all tables.

This method inspects all tables in the database (skipping system tables such as “spatial_ref_sys”) and returns the distinct simulation IDs found in the ‘simulation’ column.

Returns:: A list of unique simulation IDs.
Return type:: list[str]

class assume.common.outputs.OutputDef#

Bases: TypedDict

from_table: str#

name: str#

value: str#

class assume.common.outputs.WriteOutput(simulation_id: str, start: datetime, end: datetime, save_frequency_hours, db_uri='', export_csv_path: str = '', outputs_buffer_size_mb: int = 300, learning_mode: bool = False, evaluation_mode: bool = False, episode: int = None, eval_episode: int = None, additional_kpis: dict[str, OutputDef] = {})#

Bases: Role

Initializes an instance of the WriteOutput class.

Parameters:

simulation_id (str) – The ID of the simulation as a unique classifier.
start (datetime.datetime) – The start datetime of the simulation run.
end (datetime.datetime) – The end datetime of the simulation run.
save_frequency_hours (int) – The frequency in hours for storing data in the db and/or csv files.
db_uri – The uri of the database engine. Defaults to ‘’.
export_csv_path (str, optional) – The path for exporting CSV files, no path results in not writing the csv. Defaults to “”.
outputs_buffer_size_mb (int, optional) – The maximum storage size (in MB) for storing output data before saving it. Defaults to 300 MB.
learning_mode (bool, optional) – Indicates if the simulation is in learning mode. Defaults to False.
evaluation_mode (bool, optional) – Indicates if the simulation is in evaluation mode. Defaults to False.
additional_kpis (dict[str, OutputDef], optional) – makes it possible to define additional kpis evaluated

check_columns(table: str, df: DataFrame, index: bool = True)#

Checks and adds columns to the database table if necessary.

Parameters:

table (str) – The name of the database table.
df (pandas.DataFrame) – The DataFrame to be checked.

convert_flows(data: dict[tuple[datetime, str], float])#

Convert the flows of the grid results into a dataframe.

Parameters:: data – The records to be put into the table. Formatted like, “(datetime, line), flow” if generated by pyomo or df if it comes from pypsa.

convert_market_dispatch(market_dispatch: list[dict])#

Convert the planned dispatch of the units to a DataFrame.

Parameters:: data (any) – The records to be put into the table. Formatted like, “datetime, power, market_id, unit_id”.

convert_market_orders(market_orders: any, market_id: str)#

Convert market orders to a dataframe.

Parameters:

market_orders (any) – The market orders.
market_id (str) – The id of the market.

convert_market_results(market_results: list[dict])#

Convert market results to a dataframe.

Parameters:: market_meta (dict) – The market metadata, which includes the clearing price and volume.

convert_rl_grad_params(rl_grad_params: list[dict])#

Convert the RL (actor-critic) parameters for each gradient step to a dataframe.

Parameters:: rl_grad_params (dict) – The RL parameters per gradient step.

convert_rl_params(rl_params: list[dict])#

Convert the RL parameters such as reward, regret, and profit to a dataframe.

Parameters:: rl_params (dict) – The RL parameters.

convert_unit_dispatch(unit_dispatch: list[dict])#

Convert the actual dispatch of the units to a DataFrame.

Parameters:: unit_dispatch (list) – A list of dictionaries containing unit dispatch data. Each dictionary includes arrays for multiple values (e.g., power, costs) and other metadata.

convert_units_definition(unit_info: dict)#

Convert unit definitions to a dataframe.

Parameters:: unit_info (dict) – The unit information.

delete_db_scenario(simulation_id: str)#

Deletes all data from the database for the given simulation id.

Parameters:: simulation_id (str) – The ID of the simulation as a unique classifier.

get_sum_reward(episode: int, evaluation_mode=True)#

Retrieves the total reward for each learning unit.

Returns:: The total reward for each learning unit.
Return type:: np.ndarray

handle_output_message(content: dict, meta: MetaDict)#

Handles the incoming messages and performs corresponding actions.

Parameters:

content (dict) – The content of the message.
meta (MetaDict) – The metadata associated with the message.

on_ready()#: Called after the start of all container using activate

async on_stop()#: This function makes it possible to calculate Key Performance Indicators. It is called when the simulation is finished. It collects average price, total cost, total volume and capacity factors and uses them to calculate the KPIs. The KPIs are then stored in the database and CSV files.

setup()#: Sets up the WriteOutput instance by subscribing to messages and scheduling recurrent tasks of storing the data.

async store_dfs()#: Stores the data frames to CSV files and the database. Is scheduled as a recurrent task based on the frequency.

store_grid(grid: dict[str, DataFrame], market_id: str)#: Stores the grid data to the database. This is done once at the beginning for every agent which takes care of a grid.

assume.common.units_operator module#

class assume.common.units_operator.UnitsOperator(available_markets: list[MarketConfig], portfolio_strategies: dict[str, UnitOperatorStrategy] = {})#

Bases: Role

The UnitsOperator is the agent that manages the units. It receives the opening hours of the market and sends back the bids for the market.

available_markets#

The available markets.

Type:: list[MarketConfig]

registered_markets#

The registered markets.

Type:: dict[str, MarketConfig]

last_sent_dispatch#

The last sent dispatch.

Type:: int

portfolio_strategies#

The portfolio strategy.

Type:: UnitOperatorStrategy

valid_orders#

The valid orders.

Type:: defaultdict

units#

The units.

Type:: dict[str, BaseUnit]

id#

The id of the agent.

Type:: str

context#

The context of the agent.

Type:: Context

Parameters:

available_markets (list[MarketConfig]) – The available markets.
portfolio_strategies (dict[str, UnitOperatorStrategy], optional) – Optimized portfolio strategy. Defaults to an empty dict.

add_unit(unit: BaseUnit) → None#

Create a unit.

Parameters:: unit (BaseUnit) – The unit to be added.

calculate_unit_cashflow_and_reward(orderbook: list[Order], marketconfig: MarketConfig) → None#

Feeds the current market result back to the units.

Parameters:

orderbook (Orderbook) – The orderbook of the market.
marketconfig (MarketConfig) – The market configuration.

get_actual_dispatch(product_type: str, last: datetime) → tuple[list[tuple[datetime, float, str, str]], list[dict]]#

Retrieves the actual dispatch since the last dispatch and commits it in the unit. We calculate the series of the actual market results dataframe with accepted bids. And the unit_dispatch for all units taken care of in the UnitsOperator.

Parameters:

product_type (str) – The product type for which this is done
last (datetime.datetime) – the last date until which the dispatch was already sent

Returns:

market_dispatch and unit_dispatch dataframes

Return type:

tuple[list[tuple[datetime, float, str, str]], list[dict]]

handle_data_request(content: DataRequestMessage, meta: MetaDict) → None#

Handles the data request received from other agents.

Parameters:

content (DataRequestMessage) – The content of the data request message.
meta (MetaDict) – The meta data of the market.

handle_market_feedback(content: ClearingMessage, meta: MetaDict) → None#

Handles the feedback which is received from a market we did bid at.

Parameters:

content (ClearingMessage) – The content of the clearing message.
meta (MetaDict) – The meta data of the market.

handle_opening(opening: OpeningMessage, meta: MetaDict) → None#

When we receive an opening from the market, we schedule sending back our list of orders as a response.

Parameters:

opening (OpeningMessage) – The opening message.
meta (MetaDict) – The meta data of the market.

handle_registration_feedback(content: RegistrationMessage, meta: MetaDict) → None#

Handles the feedback received from a market regarding registration.

Parameters:

content (RegistrationMessage) – The content of the registration message.
meta (MetaDict) – The meta data of the market.

on_ready()#: Called after the start of all container using activate

participate(market: MarketConfig) → bool#

Method which decides if we want to participate on a given Market. This always returns true for now.

Parameters:: market (MarketConfig) – The market to participate in.
Returns:: True if participate, False otherwise.
Return type:: bool

async register_market(market: MarketConfig) → None#

Parameters:: market (MarketConfig) – The market to register.

set_unit_dispatch(orderbook: list[Order], marketconfig: MarketConfig) → None#

Feeds the current market result back to the units.

Parameters:

orderbook (Orderbook) – The orderbook of the market.
marketconfig (MarketConfig) – The market configuration.

setup()#: Lifecycle hook in, which will be called on adding the role to agent. The role context is known from hereon.

async store_units() → None#

async submit_bids(opening: OpeningMessage, meta: MetaDict) → None#

Formulates an orderbook and sends it to the market.

Parameters:

opening (OpeningMessage) – The opening message.
meta (MetaDict) – The meta data of the market.

write_actual_dispatch(product_type: str) → None#

Sends the actual aggregated dispatch curve to the output agent.

Parameters:: product_type (str) – The type of the product.

assume.common.utils module#

assume.common.utils.adjust_unit_operator_for_learning(bidding_strategies: dict[str, str], world_bidding_strategies: dict[str, BaseStrategy], unit_operator_id: str)#

Check if any of the bidding strategies are learning strategies. And change the unit operator to RL if learning strategies are found.

Parameters:

bidding_strategies (dict[str, str]) – The bidding strategies for the unit.
world_bidding_strategies (dict[str, BaseStrategy]) – The bidding strategies of the World
unit_operator_id (str) – The identifier of the unit operator.

Returns:

The corrected unit operator identifier.

Return type:

str

assume.common.utils.aggregate_step_amount(orderbook: list[Order], begin=None, end=None, groupby=None)#

Step function with bought volume, allows setting timeframe through begin and end, and group by columns in groupby.

Parameters:

orderbook (Orderbook) – The orderbook.
begin (datetime, optional) – The begin time. Defaults to None.
end (datetime, optional) – The end time. Defaults to None.
groupby (list[str], optional) – The columns to group by. Defaults to None.

Returns:

The aggregated orderbook timeseries.

Return type:

list[tuple[datetime, float, str, str]]

Examples

If called without groupby, this returns the aggregated orderbook timeseries

assume.common.utils.calculate_content_size(content: list | dict) → int#: Calculate the size of a content in bytes.

assume.common.utils.confirm_learning_save_path(save_path: str, continue_learning: bool) → None#: Check save_path and ask user how to proceed if it exists. Raises AssumeException if user declines to proceed.

assume.common.utils.convert_tensors(data)#

Recursively checks if the data contains PyTorch tensors and converts them to native Python types (floats, ints, lists).

Supports: - pandas.Series (vectorized for efficiency) - Lists of dictionaries (including nested structures) - Any nested list/dict structure

Parameters:: data (pandas.Series, list, dict, or other Python data types)
Returns:: The data with all tensors converted to Python-native types.

assume.common.utils.convert_to_rrule_freq(string: str) → tuple[int, int]#

Convert a string to a rrule frequency and interval.

Parameters:: string (str) – The string to be converted. Should be in the format of “1h” or “1d” or “1w”.
Returns:: The rrule frequency and interval.
Return type:: tuple[int, int]

assume.common.utils.create_incidence_matrix(lines, buses, zones_id=None)#

assume.common.utils.create_rrule(start, end, freq)#

assume.common.utils.datetime2timestamp(datetime: datetime)#

assume.common.utils.get_available_products(market_products: list[MarketProduct], startdate: datetime)#

Get all available products for a given startdate.

Parameters:

market_products (list[MarketProduct]) – List of market products.
startdate (datetime.datetime) – The startdate.

Returns:

List of available products.

Return type:

list[MarketProduct]

assume.common.utils.get_products_index(orderbook: list[Order]) → DatetimeIndex#

Creates an index containing all start times of orders in orderbook and all between.

Parameters:: orderbook (Orderbook) – The orderbook.
Returns:: The index containing all start times of orders in orderbook and all between.
Return type:: pd.DatetimeIndex

assume.common.utils.get_supported_solver(default_solver: str | None = None)#

assume.common.utils.initializer(func)#

Automatically assigns the parameters.

Parameters:: func (callable) – The function to be initialized.
Returns:: The wrapper function.
Return type:: callable

Examples

>>> class process:
...     @initializer
...     def __init__(self, cmd, reachable=False, user='root'):
...         pass
>>> p = process('halt', True)
>>> p.cmd, p.reachable, p.user
('halt', True, 'root')

assume.common.utils.interactive_input(prompt: str, default: str = '') → str#

assume.common.utils.min_max_scale(x, min_val: float, max_val: float)#

Min-Max scaling of a value x to the range [0, 1]

Parameters:

x – value(s) to scale
min_val – minimum value of the parameter
max_val – maximum value of the parameter

assume.common.utils.normalize_availability(powerplants_df, availability_df)#

assume.common.utils.parse_duration(duration_str)#

assume.common.utils.plot_orderbook(orderbook: list[Order], results: list[dict])#

Plot the merit order of bids for each node in a separate subplot.

Parameters:

orderbook (Orderbook) – The orderbook.
results (list[dict]) – The results of the clearing.

Returns:

The figure and axes of the plot.

Return type:

tuple[matplotlib.figure.Figure, matplotlib.axes.Axes]

assume.common.utils.rename_study_case(path: str, old_key: str, new_key: str)#

Rename key in config file and save changes.

This function makes changes to the config file. Background is so that study cases can be simulated multiple times under different names with the same configuration.

Parameters:

path (str) – The path to the config file.
old_key (str) – The original name of the key without adjustments. E.g. study_case from available_examples: “base”.
new_key (str) – The name of the key with adjustments. E.g. added run number: “base_run_1”.

assume.common.utils.separate_orders(orderbook: list[Order])#

Separate orders with several hours into single hour orders.

Parameters:: orderbook (Orderbook) – The orderbook.
Returns:: The updated orderbook.
Return type:: list

Note

This function separates orders with several hours into single hour orders and modifies the orderbook in place.

assume.common.utils.str_to_bool(val)#

Convert a string representation of truth to True or False.

True values are ‘y’, ‘yes’, ‘t’, ‘true’, ‘on’, and ‘1’; false values are ‘n’, ‘no’, ‘f’, ‘false’, ‘off’, and ‘0’. Raises ValueError if ‘val’ is anything else.

assume.common.utils.timestamp2datetime(timestamp: float)#

assume.common.utils.visualize_orderbook(order_book: list[Order])#

Visualize the orderbook.

Parameters:: order_book (Orderbook) – The orderbook.

assume.common.grid_utils module#

assume.common.grid_utils.add_backup_generators(network: Network, backup_marginal_cost: float = 100000.0) → None#

Add generators normally to the grid

Parameters:

network (pypsa.Network) – the pypsa network to which the generators are
generators (pandas.DataFrame) – the generators dataframe

assume.common.grid_utils.add_generators(network: Network, generators: DataFrame) → None#

Add generators normally to the grid

Parameters:

network (pypsa.Network) – the pypsa network to which the generators are
generators (pandas.DataFrame) – the generators dataframe

assume.common.grid_utils.add_loads(network: Network, loads: DataFrame) → None#

Add loads normally to the grid

Parameters:

network (pypsa.Network) – the pypsa network to which the loads are
loads (pandas.DataFrame) – the loads dataframe

assume.common.grid_utils.add_nodal_loads(network: Network, loads: DataFrame) → None#: This adds loads to the nodal PyPSA network with respective bus data to which they are connected. The loads are added as generators with negative sign so their dispatch can be also curtailed, since regular load in PyPSA represents only an inelastic demand.

assume.common.grid_utils.add_redispatch_generators(network: Network, generators: DataFrame, backup_marginal_cost: float = 100000.0) → None#

Adds the given generators for redispatch. This includes functions to optimize up as well as down and adds backup capacities of powerplants to be able to adjust accordingly when a congestion happens.

Parameters:

network (pypsa.Network) – the pypsa network to which the generators are
generators (pandas.DataFrame) – the generators dataframe
backup_marginal_cost (float, optional) – The cost of dispatching the backup units in [€/MW]. Defaults to 1e5.

assume.common.grid_utils.add_redispatch_loads(network: Network, loads: DataFrame) → None#: This adds loads to the redispatch PyPSA network with respective bus data to which they are connected

assume.common.grid_utils.calculate_network_meta(network, product: MarketProduct, i: int)#

This function calculates the meta data such as supply and demand volumes, and nodal prices.

Parameters:

product (MarketProduct) – The product for which clearing happens.
i (int) – The index of the product in the market products list.

Returns:

The meta data.

Return type:

dict

assume.common.grid_utils.read_pypsa_grid(network: Network, grid_dict: dict[str, DataFrame])#

Generates the pypsa grid from a grid dictionary. Does not add the generators, as they are added in different ways, depending on whether redispatch is used.

Parameters:

network (pypsa.Network) – the pypsa network to which the components will be added
grid_dict (dict[str, pd.DataFrame]) – the dictionary containing dataframes for generators, loads, buses and links

Common components and methods

Contents

Common components and methods#

Submodules#

assume.common.base module#

assume.common.exceptions module#

assume.common.mango_serializer module#

assume.common.market_objects module#

assume.common.outputs module#

assume.common.units_operator module#

assume.common.utils module#

assume.common.grid_utils module#

Module contents#