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[assume.common.market_objects.Product], **kwargs) list[assume.common.market_objects.Order]#

Calculates the bids for the next time step.

Parameters:
Returns:

The bids

Return type:

Orderbook

calculate_reward(unit, marketconfig: MarketConfig, orderbook: list[assume.common.market_objects.Order])#

Calculates the reward for the given unit

Parameters:
  • unit (BaseUnit) – the unit

  • marketconfig (MarketConfig) – The market configuration.

  • orderbook (Orderbook) – The orderbook.

class assume.common.base.BaseUnit(id: str, unit_operator: str, technology: str, bidding_strategies: dict[str, assume.common.base.BaseStrategy], index: DatetimeIndex, node: str = '', forecaster: Forecaster = None, **kwargs)#

Bases: object

A base class for a unit.

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 (pd.DatetimeIndex) – The index of the unit.

  • outputs (dict[str, pd.Series]) – The output of the unit.

  • forecaster (Forecaster) – The forecast of the unit.

as_dict() dict#

Returns a dictionary representation of the unit.

Returns:

a dictionary representation of the unit

Return type:

dict

calculate_bids(market_config: MarketConfig, product_tuples: list[tuple]) list[assume.common.market_objects.Order]#

Calculate 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

calculate_cashflow(product_type: str, orderbook: list[assume.common.market_objects.Order])#

calculates the cashflow for the given product_type

Parameters:
  • product_type (str) – the product type

  • orderbook (Orderbook) – The orderbook.

calculate_generation_cost(start: datetime, end: datetime, product_type: str)#
calculate_marginal_cost(start: Timestamp, power: float) float#

calculates the marginal cost for the given power

Parameters:
  • start (pd.Timestamp) – 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: Timestamp, end: Timestamp) Series#

check if the total dispatch plan is feasible This checks if the market feedback is feasible for the given unit. And sets the closest dispatch if not. The end param should be inclusive.

Parameters:
  • start (pd.Timestamp) – the start time of the dispatch

  • end (pd.Timestamp) – the end time of the dispatch

Returns:

the volume of the unit within the given time range

Return type:

pd.Series

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

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

Parameters:
  • dt (datetime) – the datetime

  • product_type (str) – the product type

Returns:

the output before the given datetime

Return type:

float

get_starting_costs(op_time: int)#

op_time is hours running from get_operation_time returns the costs if start_up is planned :param op_time: operation time :type op_time: int :return: start_costs :rtype: float

set_dispatch_plan(marketconfig: MarketConfig, orderbook: list[assume.common.market_objects.Order]) None#

adds dispatch plan from current market result to total dispatch plan

Parameters:
  • marketconfig (MarketConfig) – The market configuration.

  • orderbook (Orderbook) – The orderbook.

class assume.common.base.LearningConfig#

Bases: TypedDict

A class for the learning configuration.

Parameters:
  • observation_dimension (int) – The observation dimension.

  • action_dimension (int) – The action dimension.

  • continue_learning (bool) – Whether to continue learning.

  • load_model_path (str) – The path to the model to load.

  • max_bid_price (float) – The maximum bid price.

  • learning_mode (bool) – Whether to use learning mode.

  • algorithm (str) – The algorithm to use.

  • learning_rate (float) – The learning rate.

  • training_episodes (int) – The number of training episodes.

  • episodes_collecting_initial_experience (int) – The number of episodes collecting initial experience.

  • train_freq (int) – The training frequency.

  • gradient_steps (int) – The number of gradient steps.

  • batch_size (int) – The batch size.

  • gamma (float) – The discount factor.

  • device (str) – The device to use.

:param noise_sigma : The standard deviation of the noise. :type noise_sigma: float

Parameters:
  • noise_scale (int) – Controls the initial strength of the noise.

  • noise_dt (int) – Determines how quickly the noise weakens over time.

  • trained_actors_path (str) – The path to the learned model to load.

action_dimension: int#
algorithm: str#
batch_size: int#
continue_learning: bool#
device: str#
episodes_collecting_initial_experience: int#
gamma: float#
gradient_steps: int#
learning_mode: bool#
learning_rate: float#
load_model_path: str#
max_bid_price: float#
noise_dt: int#
noise_scale: int#
noise_sigma: float#
observation_dimension: int#
train_freq: int#
trained_actors_path: str#
training_episodes: int#
class assume.common.base.LearningStrategy(*args, **kwargs)#

Bases: BaseStrategy

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

Parameters:
  • args (list) – The arguments.

  • kwargs (dict) – The keyword arguments.

act_dim: int#
obs_dim: int#
class assume.common.base.SupportsMinMax(id: str, unit_operator: str, technology: str, bidding_strategies: dict[str, assume.common.base.BaseStrategy], index: DatetimeIndex, node: str = '', forecaster: Forecaster = None, **kwargs)#

Bases: BaseUnit

Base Class used for Powerplant derived classes

Parameters:
  • min_power (float) – The minimum power output of the unit.

  • max_power (float) – The maximum power output of the unit.

  • ramp_down (float) – How much power can be decreased in one time step.

  • ramp_up (float) – How much power can be increased in one time step.

  • efficiency (float) – The efficiency of the unit.

  • emission_factor (float) – The emission factor of the unit.

  • min_operating_time (int) – The minimum time the unit has to be on.

  • min_down_time (int) – The minimum time the unit has to be off.

calculate_min_max_power(start: Timestamp, end: Timestamp, product_type='energy') tuple[pandas.core.series.Series, pandas.core.series.Series]#

Calculates the min and max power for the given time period

Parameters:
  • start (pd.Timestamp) – the start time of the dispatch

  • end (pd.Timestamp) – 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[pd.Series, pd.Series]

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

Calculates the ramp for the given power

Parameters:
  • previous_power (float) – the previous power output of the unit

  • power (float) – the power output of the unit

  • current_power (float) – the current power output of the unit

Returns:

the ramp for the given power

Return type:

float

efficiency: float#
emission_factor: float#
get_average_operation_times(start: datetime)#

calculates the average uninterupted operation time :param start: the current time :type start: datetime :return: avg_op_time :rtype: float :return: avg_down_time :rtype: float

get_clean_spread(prices: DataFrame)#

returns the clean spread for the given prices

Parameters:

prices (pd.DataFrame) – the prices

Returns:

the clean spread for the given prices

Return type:

float

get_operation_time(start: datetime)#

returns the operation time if unit is on since 4 hours, it returns 4 if the unit is off since 4 hours, it returns -4 The value at start is not considered

Parameters:

start (datetime) – the start time

Returns:

the operation time

Return type:

int

get_starting_costs(op_time: int)#

op_time is hours running from get_operation_time returns the costs if start_up is planned :param op_time: operation time :type op_time: int :return: start_costs :rtype: float

max_power: float#
min_down_time: int#
min_operating_time: int#
min_power: float#
ramp_down: float#
ramp_up: float#
class assume.common.base.SupportsMinMaxCharge(id: str, unit_operator: str, technology: str, bidding_strategies: dict[str, assume.common.base.BaseStrategy], index: DatetimeIndex, node: str = '', forecaster: Forecaster = None, **kwargs)#

Bases: BaseUnit

Base Class used for Storage derived classes

Parameters:
  • initial_soc (float) – The initial state of charge of the storage.

  • min_power_charge (float) – How much power must be charged at least in one time step.

  • max_power_charge (float) – How much power can be charged at most in one time step.

  • min_power_discharge (float) – How much power must be discharged at least in one time step.

  • max_power_discharge (float) – How much power can be discharged at most in one time step.

  • ramp_up_discharge (float) – How much power can be increased in discharging in one time step.

  • ramp_down_discharge (float) – How much power can be decreased in discharging in one time step.

  • ramp_up_charge (float) – How much power can be increased in charging in one time step.

  • ramp_down_charge (float) – How much power can be decreased in charging in one time step.

  • max_volume (float) – The maximum volume of the storage.

  • efficiency_charge (float) – The efficiency of charging.

  • efficiency_discharge (float) – The efficiency of discharging.

calculate_min_max_charge(start: Timestamp, end: Timestamp, product_type='energy') tuple[pandas.core.series.Series, pandas.core.series.Series]#

calculates the min and max charging power for the given time period

Parameters:
  • start (pd.Timestamp) – the start time of the dispatch

  • end (pd.Timestamp) – the end time of the dispatch

  • product_type (str) – the product type of the unit

Returns:

the min and max charging power for the given time period

Return type:

tuple[pd.Series, pd.Series]

calculate_min_max_discharge(start: Timestamp, end: Timestamp, product_type='energy') tuple[pandas.core.series.Series, pandas.core.series.Series]#

calculates the min and max discharging power for the given time period

Parameters:
  • start (pd.Timestamp) – the start time of the dispatch

  • end (pd.Timestamp) – the end time of the dispatch

  • product_type (str) – the product type of the unit

Returns:

the min and max discharging power for the given time period

Return type:

tuple[pd.Series, pd.Series]

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

calculates the ramp for the given charging power

Parameters:
  • previous_power (float) – the previous power output of the unit

  • power_charge (float) – the charging power output of the unit

  • current_power (float) – the current power output of the unit

Returns:

the ramp for the given charging power

Return type:

float

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

calculates the ramp for the given discharging power

Parameters:
  • previous_power (float) – the previous power output of the unit

  • power_discharge (float) – the discharging power output of the unit

  • current_power (float) – the current power output of the unit

Returns:

the ramp for the given discharging power

Return type:

float

efficiency_charge: float#
efficiency_discharge: float#
get_clean_spread(prices: DataFrame)#

returns the clean spread for the given prices

Parameters:

prices (pd.DataFrame) – the prices

Returns:

the clean spread for the given prices

Return type:

float

get_soc_before(dt: datetime) float#

return SoC before the given datetime. If datetime is before the start of the index, the initial SoC is returned. The SoC is a float between 0 and 1.

Parameters:

dt (datetime) – the datetime

Returns:

the SoC before the given datetime

Return type:

float

initial_soc: float#
max_power_charge: float#
max_power_discharge: float#
max_volume: float#
min_power_charge: float#
min_power_discharge: float#
ramp_down_charge: float#
ramp_down_discharge: float#
ramp_up_charge: float#
ramp_up_discharge: float#

assume.common.exceptions module#

exception assume.common.exceptions.AssumeException#

Bases: Exception

exception assume.common.exceptions.InvalidTypeException#

Bases: AssumeException

assume.common.forecasts module#

class assume.common.forecasts.CsvForecaster(index: Series, powerplants: dict[str, pandas.core.series.Series] = {}, *args, **kwargs)#

Bases: Forecaster

A Forecaster that reads forecasts from csv files.

Args: - index (pd.Series): The index of the forecasts. - powerplants (dict[str, pd.Series]): The power plants.

This class represents a forecaster that provides timeseries for forecasts derived from existing files. It initializes with the provided index. It includes methods to retrieve forecasts for specific columns, availability of units, and prices of fuel types, returning the corresponding timeseries as pandas Series.

calc_forecast_if_needed()#

Calculates the forecasts if they are not already calculated.

This method calculates additional forecasts if they do not already exist, including “price_EOM” and “residual_load_EOM”.

calculate_EOM_price_forecast()#

Calculates the merit order price forecast for the entire time horizon.

Returns: - pd.Series: The merit order price forecast.

This function calculates the merit order price forecast, which is provided as a price forecast to the RL agents, for the entire time horizon at once. The method considers the infeed of renewables, residual demand, and the marginal costs of power plants to derive the price forecast.

The calculation involves the following steps: 1. Calculates the marginal costs for each power plant based on fuel costs, efficiencies, emissions, and fixed costs. 2. Sorts the power plants based on their marginal costs and availability. 3. Computes the cumulative power of available power plants. 4. Determines the price forecast by iterating through the sorted power plants, setting the price for times that can still be provided with a specific technology and cheaper ones.

TODO: - Extend price forecasts for all markets, not just specified for the DAM. - Consider the inclusion of storages in the price forecast calculation.

Returns the resulting merit order price forecast as a pandas Series.

calculate_marginal_cost(pp_series: Series)#

Calculates the marginal cost of a power plant based on the fuel costs and efficiencies of the power plant.

Args: - pp_series (pd.Series): Series containing power plant data.

Returns: - float: The marginal cost of the power plant.

This method calculates the marginal cost of a power plant by taking into account the following factors: - Fuel costs based on the fuel type and efficiency of the power plant. - Emissions costs considering the CO2 price and emission factor of the power plant. - Fixed costs, if specified for the power plant.

The calculation involves the following steps: 1. Determines the fuel price based on the fuel type of the power plant. 2. Calculates the fuel cost by dividing the fuel price by the efficiency of the power plant. 3. Calculates the emissions cost based on the CO2 price and emission factor, adjusted by the efficiency of the power plant. 4. Considers any fixed costs specified for the power plant. 5. Aggregates the fuel cost, emissions cost, and fixed cost to obtain the marginal cost of the power plant.

Returns the resulting marginal cost as a float value.

calculate_residual_demand_forecast()#

Calculates the residual demand forecast.

Returns: - pd.Series: The residual demand forecast.

This method calculates the residual demand forecast by subtracting the total available power from renewable energy (VRE) power plants from the overall demand forecast for each time step.

The calculation involves the following steps: 1. Selects VRE power plants (wind_onshore, wind_offshore, solar) from the powerplants data. 2. Creates a DataFrame, vre_feed_in_df, with columns representing VRE power plants and initializes it with zeros. 3. Calculates the power feed-in for each VRE power plant based on its availability and maximum power. 4. Calculates the residual demand by subtracting the total VRE power feed-in from the overall demand forecast.

Returns the resulting residual demand forecast as a pandas Series.

get_registered_market_participants(market_id)#

Retrieves information about market participants for accurate price forecast.

Args: - market_id (str): The market ID.

Returns: - pd.DataFrame: The registered market participants.

This method retrieves information about market participants to make accurate price forecasts. Currently, the functionality for using different markets and specified registration for the price forecast is not implemented, so it returns the power plants as a DataFrame.

save_forecasts(path)#

Saves the forecasts to a csv file.

Args: - path (str): The path to save the forecasts to.

This method saves the forecasts to a csv file located at the specified path. If no forecasts are provided, an error message is logged.

set_forecast(data: DataFrame | Series | None, prefix='')#

Sets the forecast for a given column.

Args: - data (pd.DataFrame | pd.Series | None): The forecast data. - prefix (str): The prefix of the column.

If data is a DataFrame, it sets the forecast for each column in the DataFrame. If data is a Series, it sets the forecast for the given column. If data is None, no action is taken.

class assume.common.forecasts.Forecaster(index: Series)#

Bases: object

A Forecaster can provide timeseries for forecasts which are derived either from existing files, random noise, or actual forecast methods.

Args: - index (pd.Series): The index of the forecasts.

This class represents a forecaster that provides timeseries for forecasts derived from existing files, random noise, or actual forecast methods. It initializes with the provided index. It includes methods to retrieve forecasts for specific columns, availability of units, and prices of fuel types, returning the corresponding timeseries as pandas Series.

get_availability(unit: str)#

Returns the availability of a given unit.

Args: - unit (str): The unit.

Returns: - pd.Series: The availability of the unit.

This method returns the availability of a given unit as a pandas Series based on the provided index.

get_price(fuel_type: str)#

Returns the price for a given fuel type or zeros if the type does not exist.

Args: - fuel_type (str): The fuel type.

Returns: - pd.Series: The price of the fuel.

This method returns the price for a given fuel type as a pandas Series or zeros if the type does not exist, based on the provided index.

class assume.common.forecasts.NaiveForecast(index: Series, availability: float | list = 1, fuel_price: float | list = 10, co2_price: float | list = 10, demand: float | list = 100, price_forecast: float | list = 50, *args, **kwargs)#

Bases: Forecaster

A forecaster that generates forecasts using naive methods.

Args: - index (pd.Series): The index of the forecasts.

Optional Args: - availability (float | list): The availability of the power plants. - fuel_price (float | list): The fuel price. - co2_price (float | list): The CO2 price. - demand (float | list): The demand. - price_forecast (float | list): The price forecast.

This class represents a forecaster that generates forecasts using naive methods. It inherits from the Forecaster class and initializes with the provided index and optional parameters for availability, fuel price, CO2 price, demand, and price forecast.

class assume.common.forecasts.RandomForecaster(index: Series, powerplants: dict[str, pandas.core.series.Series] = {}, sigma: float = 0.02, *args, **kwargs)#

Bases: CsvForecaster

A forecaster that generates forecasts using random noise.

Args: - index (pd.Series): The index of the forecasts. - powerplants (dict[str, pd.Series]): The power plants. - sigma (float): The standard deviation of the noise.

This class represents a forecaster that generates forecasts using random noise. It inherits from the CsvForecaster class and initializes with the provided index, power plants, and standard deviation of the noise.

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[assume.common.market_objects.Order]#
context: str#
market_id: str#
rejected_orders: list[assume.common.market_objects.Order]#
class assume.common.market_objects.DataRequestMessage#

Bases: TypedDict

context: str#
end_time: datetime#
market_id: str#
metric: str#
start_time: datetime#
class assume.common.market_objects.MarketConfig(name: str, opening_hours: ~dateutil.rrule.rrule, opening_duration: ~datetime.timedelta, market_mechanism: str, 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, 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: ~typing.Callable[[~mango.agent.core.Agent], bool] = <function MarketConfig.<lambda>>)#

Bases: object

Describes the configuration of a market.

Parameters:
  • name (str) – the name of the market

  • addr (str | None) – the address of the market

  • aid (str | None) – automatic id of the market

  • opening_hours (rr.rrule) – the opening hours of the market

  • opening_duration (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) – the maximum valid bid volume of the market

  • maximum_bid_price (float) – 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 | None) – lambda function which determines if an agent is eligible to trade this product

additional_fields: list[str]#
addr = None#
aid = None#
eligible_obligations_lambda()#
market_mechanism: str#
market_products: list[assume.common.market_objects.MarketProduct]#
maximum_bid_price: float | None = 3000.0#
maximum_bid_volume: float | None = 2000.0#
maximum_gradient: float = None#
minimum_bid_price: float = -500.0#
name: str#
opening_duration: timedelta#
opening_hours: rrule#
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 (rd | rr.rrule) – the duration of the product

  • count (int) – how many future durations can be traded, must be >= 1

  • first_delivery (rd) – 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 http://www.fipa.org/specs/fipa00061/SC00061G.html#_Toc26669700

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[assume.common.market_objects.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) – the start time of the order

  • end_time (datetime) – the end time of the order

  • volume (float) – the volume of the order

  • price (float) – the price of the order

  • accepted_volume (float) – the accepted volume of the order

  • accepted_price (float) – the accepted price of the order

  • only_hours (OnlyHours | None) – tuple of hours from which this order is available, on multi day products

  • agent_id (str) – the id of the agent

accepted_price: Number | dict[datetime.datetime, numbers.Number]#
accepted_volume: Number | dict[datetime.datetime, numbers.Number]#
agent_id: str#
bid_id: str#
end_time: datetime#
only_hours: OnlyHours | None#
price: Number#
start_time: datetime#
volume: Number | dict[datetime.datetime, numbers.Number]#
class assume.common.market_objects.OrderBookMessage#

Bases: TypedDict

context: str#
market_id: str#
orderbook: list[assume.common.market_objects.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) – the start time of the product

  • end (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

context: str#
information: dict#
market_id: str#
class assume.common.market_objects.RegistrationReplyMessage#

Bases: TypedDict

accepted: bool#
context: str#
market_id: str#

assume.common.outputs module#

class assume.common.outputs.WriteOutput(simulation_id: str, start: datetime, end: datetime, db_engine=None, export_csv_path: str = '', save_frequency_hours: int = None, learning_mode: bool = False, evaluation_mode: bool = False)#

Bases: Role

Initializes an instance of the WriteOutput class.

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

  • start (datetime) – The start datetime of the simulation run.

  • end (datetime) – The end datetime of the simulation run.

  • db_engine (optional) – The database engine. Defaults to None.

  • export_csv_path (str, optional) – The path for exporting CSV files, no path results in not writing the csv. Defaults to “”.

  • save_frequency_hours (int) – The frequency in hours for storing data in the db and/or csv files. Defaults to None.

check_columns(table: str, df: DataFrame)#

If a simulation before has been started which does not include an additional field we try to add the field. For now, this only works for float and text. An alternative which finds the correct types would be to use

check_for_tensors(data)#

Checks if the data contains tensors and converts them to floats.

Parameters:

data (any) – The data to be checked.

del_similar_runs()#
delete_db_scenario(simulation_id)#

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

Parameters:

simulation_id (str) – The ID of the simulation as a unique calssifier.

get_sum_reward()#
handle_message(content, meta: MetaDict)#

Handles the incoming messages and performs corresponding actions.

Parameters:
  • content (dict) – The content of the message.

  • meta (any) – The metadata associated with the message. (not needed yet)

async on_stop()#

This function makes it possible to calculate Key Performance Indicators

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.

write_market_dispatch(data)#

Writes the planned dispatch of the units after the market clearing to a csv and db In the case that we have no portfolio optimisation this equals the resulting bids.

Parameters:

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

write_market_orders(market_orders, market_id)#

Writes market orders to the corresponding data frame. Append new data until it is written to db and csv with store_df function.

Parameters:
  • market_orders (any) – The market orders.

  • market_id (str) – The id of the market.

write_market_results(market_meta)#

Writes market results to the corresponding data frame.

Parameters:

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

write_rl_params(rl_params)#

Writes the RL parameters to the corresponding data frame.

Parameters:

rl_params (any) – The RL parameters.

write_unit_dispatch(data)#

Writes the actual dispatch of the units to a csv and db

Parameters:

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

write_units_definition(unit_info: dict)#

Writes unit definitions to the corresponding data frame and directly store it in db and csv. Since that is only done once, no need for recurrent scheduling arises.

Parameters:

unit_info (dict) – The unit information.

assume.common.scenario_loader module#

assume.common.scenario_loader.add_units(units_df: DataFrame, unit_type: str, world: World, forecaster: Forecaster)#

This function adds units to the world from a given dataframe. The callback is used to adjust unit_params depending on the unit_type, before adding the unit to the world.

Parameters:
  • units_df (pd.DataFrame) – the dataframe containing the units

  • unit_type (str) – the type of the unit

  • world (World) – the world

  • forecaster (Forecaster) – the forecaster

async assume.common.scenario_loader.async_load_custom_units(world: World, inputs_path: str, scenario: str, file_name: str, unit_type: str)#

This function loads custom units from a given path.

Parameters:
  • world (World) – the world

  • inputs_path (str) – the path to the inputs folder

  • scenario (str) – the name of the scenario

  • file_name (str) – the name of the file

  • unit_type (str) – the type of the unit

assume.common.scenario_loader.convert_to_rrule_freq(string)#

This function converts a string to a rrule frequency and interval. The string should be in the format of “1h” or “1d” or “1w”.

Parameters:

string (str) – the string to be converted

Returns:

the rrule frequency and interval

Return type:

tuple[int, int]

assume.common.scenario_loader.load_custom_units(world: World, inputs_path: str, scenario: str, file_name: str, unit_type: str)#
assume.common.scenario_loader.load_file(path: str, config: dict, file_name: str, index: DatetimeIndex = None) DataFrame#

This function loads a csv file from a given path and returns a dataframe.

Parameters:
  • path (str) – the path to the csv file

  • config (dict) – the config file

  • file_name (str) – the name of the csv file

  • index (pd.DatetimeIndex) – the index of the dataframe

Returns:

the dataframe

Return type:

pd.DataFrame

assume.common.scenario_loader.load_scenario_folder(world: World, inputs_path: str, scenario: str, study_case: str, perform_learning: bool = True, perform_evaluation: bool = False, episode: int = 1, eval_episode: int = 1, trained_actors_path='')#

Load a scenario from a given path.

Parameters:
  • world (World) – The world.

  • inputs_path (str) – Path to the inputs folder.

  • scenario (str) – Name of the scenario.

  • study_case (str) – Name of the study case.

async assume.common.scenario_loader.load_scenario_folder_async(world: World, inputs_path: str, scenario: str, study_case: str, perform_learning: bool = True, perform_evaluation: bool = False, episode: int = 0, eval_episode: int = 0, trained_actors_path: str = '')#

Load a scenario from a given path. Raises: ValueError: If the scenario or study case is not found.

Parameters:
  • world (World) – The world.

  • inputs_path (str) – Path to the inputs folder.

  • scenario (str) – Name of the scenario.

  • study_case (str) – Name of the study case.

assume.common.scenario_loader.make_market_config(id: str, market_params: dict, world_start: datetime, world_end: datetime)#

This function creates a market config from a given dictionary.

Parameters:
  • id (str) – the id of the market

  • market_params (dict) – the market parameters

  • world_start (datetime) – the start time of the world

  • world_end (datetime) – the end time of the world

Returns:

the market config

Return type:

MarketConfig

assume.common.scenario_loader.run_learning(world: World, inputs_path: str, scenario: str, study_case: str)#

assume.common.units_operator module#

class assume.common.units_operator.UnitsOperator(available_markets: list[assume.common.market_objects.MarketConfig], opt_portfolio: tuple[bool, assume.common.base.BaseStrategy] | None = None)#

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.

Parameters:
async add_unit(unit: BaseUnit)#

Create a unit.

Parameters:

unit (BaseUnit) – the unit to be added

async formulate_bids(market: MarketConfig, products: list[tuple]) list[assume.common.market_objects.Order]#

Takes information from all units that the unit operator manages and formulates the bid to the market from that according to the bidding strategy of the unit itself.

Parameters:
  • market (MarketConfig) – the market to formulate bids for

  • products (list[tuple]) – the products to formulate bids for

:return OrderBook that is submitted as a bid to the market :rtype OrderBook

async formulate_bids_portfolio(market: MarketConfig, products: list[tuple]) list[assume.common.market_objects.Order]#

Takes information from all units that the unit operator manages and formulates the bid to the market from that according to the bidding strategy of the unit operator.

This is the portfolio optimization version

Parameters:
  • market (MarketConfig) – the market to formulate bids for

  • products (list[tuple]) – the products to formulate bids for

Returns:

OrderBook that is submitted as a bid to the market

Return type:

OrderBook

handle_data_request(content: DataRequestMessage, meta: MetaDict)#
handle_market_feedback(content: ClearingMessage, meta: MetaDict)#

handles the feedback which is received from a market we did bid at stores accepted orders, sets the received power writes result back for the learning and executes the dispatch, including ramping for times in the past

Parameters:
  • content (ClearingMessage) – the content of the clearing message

  • meta (MetaDict) – the meta data of the market

handle_opening(opening: OpeningMessage, meta: MetaDict)#

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

Parameters:
handle_registration_feedback(content: RegistrationMessage, meta: MetaDict)#
participate(market: MarketConfig)#

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

async register_market(market: MarketConfig)#

Register a market.

Parameters:

market (MarketConfig) – the market to register

set_unit_dispatch(orderbook: list[assume.common.market_objects.Order], marketconfig: MarketConfig)#

feeds the current market result back to the units this does not respect bids from multiple markets for the same time period, as we only have access to the current orderbook here

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 submit_bids(opening: OpeningMessage, meta: MetaDict)#

formulates an orderbook and sends it to the market. This will handle optional portfolio processing

Parameters:

opening (OpeningMessage) – the opening message

write_actual_dispatch(product_type: str)#

sends the actual aggregated dispatch curve works across multiple markets sends dispatch at or after it actually happens

write_learning_params(orderbook: list[assume.common.market_objects.Order], marketconfig: MarketConfig)#

sends the current rl_strategy update to the output agent

Parameters:
  • orderbook (Orderbook) – the orderbook of the market

  • marketconfig (MarketConfig) – the market configuration

write_learning_to_output(start: datetime, marketconfig: MarketConfig)#
write_to_learning(start: datetime, marketconfig: MarketConfig, obs_dim: int, act_dim: int, device: str, learning_unit_count: int)#

assume.common.utils module#

assume.common.utils.aggregate_step_amount(orderbook: list[assume.common.market_objects.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. This allows to have separate time series per market and bid_id/unit_id. The orderbook must contain all relevant orders. E.g. to calculate the current volume from 01.06 to 02.06, a yearly base order from 01.01-31.12 must also be given, to be considered.

If called without groupby, this returns the aggregated orderbook timeseries

Parameters:
  • orderbook (Orderbook) – the orderbook

  • begin (datetime | None) – the begin time

  • end (datetime | None) – the end time

  • groupby (list[str] | None) – the columns to group by

Returns:

the aggregated orderbook timeseries

Return type:

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

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

Get all available products for a given startdate

Parameters:
  • market_products (list[MarketProduct]) – list of market products

  • startdate (datetime) – the startdate

Returns:

list of available products

Return type:

list[MarketProduct]

assume.common.utils.get_test_demand_orders(power: array)#
assume.common.utils.initializer(func)#

Automatically assigns the parameters. >>> 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.plot_orderbook(orderbook: list[assume.common.market_objects.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.separate_orders(orderbook)#
assume.common.utils.visualize_orderbook(order_book: list[assume.common.market_objects.Order])#

Visualize the orderbook

Parameters:

order_book (Orderbook) – the orderbook

Module contents#