Ctrl+K

Common components and methods

Contents

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: BaseUnit, 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.

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, assume.common.base.BaseStrategy], index: DatetimeIndex, node: str = 'node0', forecaster: Forecaster = None, 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.

id#

The ID of the unit.

Type:

str

unit_operator#

The operator of the unit.

Type:

str

technology#

The technology of the unit.

Type:

str

bidding_strategies#

The bidding strategies of the unit.

Type:

dict[str, BaseStrategy]

index#

The index of the unit.

Type:

pandas.DatetimeIndex

node#

The node of the unit. Defaults to “”.

Type:

str, optional

forecaster#

The forecast of the unit. Defaults to None.

Type:

Forecaster, optional

\*\*kwargs

Additional keyword arguments.

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 (pandas.DatetimeIndex) – 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, Union[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[assume.common.market_objects.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[assume.common.market_objects.Order])#

Calculates the cashflow for the given product type.

Parameters:

  • product_type – The product type.

  • 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.

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: Timestamp, power: float) → float#

Calculates the marginal cost for the given power.

Parameters:

  • start (pandas.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#

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[assume.common.market_objects.Order]) → None#

Iterates through the orderbook, adding the accepted volumes to the corresponding time slots in the dispatch plan. It then calculates the cashflow and the reward for the bidding strategies.

Parameters:

  • marketconfig (MarketConfig) – The market configuration.

  • orderbook (Orderbook) – The orderbook.

class assume.common.base.LearningConfig#

Bases: TypedDict

A class for the learning configuration.

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#
max_bid_price: float#
noise_dt: int#
noise_scale: int#
noise_sigma: float#
observation_dimension: int#
train_freq: int#
trained_policies_save_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.

obs_dim#

The observation dimension.

Type:

int

act_dim#

The action dimension.

Type:

int

Parameters:

  • *args (list) – The arguments.

  • **kwargs (dict) – The keyword arguments.

class assume.common.base.SupportsMinMax(id: str, unit_operator: str, technology: str, bidding_strategies: dict[str, assume.common.base.BaseStrategy], index: DatetimeIndex, node: str = 'node0', forecaster: Forecaster = None, location: tuple[float, float] = (0.0, 0.0), **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: Timestamp, end: Timestamp, product_type: str = 'energy') → tuple[pandas.core.series.Series, pandas.core.series.Series]#

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

Parameters:
Returns:

The min and max power for the given time period.

Return type:

tuple[pandas.Series, pandas.Series]

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_average_operation_times(start: datetime) → tuple[float, float]#

Calculates the average uninterrupted operation and down time.

Parameters:

start (datetime.datetime) – The current time.

Returns:

Tuple of the average operation time avg_op_time and average down time avg_down_time.

Return type:

tuple[float, float]

Note

down_time in general is indicated with negative values

get_clean_spread(prices: DataFrame) → float#

Returns the clean spread for the given prices.

Parameters:

prices (pandas.DataFrame) – The prices.

Returns:

The clean spread for the given prices.

Return type:

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.

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#
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 = 'node0', forecaster: Forecaster = None, location: tuple[float, float] = (0.0, 0.0), **kwargs)#

Bases: BaseUnit

Base Class used for units with energy storage.

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 (pandas.Timestamp) – The start time of the dispatch.

  • end (pandas.Timestamp) – The end time of the dispatch.

  • product_type (str, optional) – The product type of the unit. Defaults to “energy”.

Returns:

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

Return type:

tuple[pandas.Series, pandas.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 (pandas.Timestamp) – The start time of the dispatch.

  • end (pandas.Timestamp) – The end time of the dispatch.

  • product_type (str, optional) – The product type of the unit. Defaults to “energy”.

Returns:

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

Return type:

tuple[pandas.Series, pandas.Series]

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.

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. Defaults to 0.

Returns:

The discharging power adjusted to the ramping constraints.

Return type:

float

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

Returns the clean spread for the given prices.

Parameters:

prices (pandas.DataFrame) – The prices.

Returns:

The clean spread for the given prices.

Return type:

float

get_soc_before(dt: datetime) → float#

Returns the State of Charge (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.datetime) – The current 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_units: dict[str, pandas.core.series.Series] = {}, demand_units: dict[str, pandas.core.series.Series] = {}, market_configs: dict[str, pandas.core.series.Series] = {}, *args, **kwargs)#

Bases: Forecaster

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.

index#

The index of the forecasts.

Type:

pandas.Series

powerplants_units#

The power plants.

Type:

dict[str, pandas.Series]

Parameters:

Example

>>> forecaster = CsvForecaster(index=pd.Series([1, 2, 3]))
>>> forecast = forecaster['temperature']
>>> print(forecast)
calc_forecast_if_needed()#

Calculates the forecasts if they are not already calculated.

This method calculates price forecast and residual load forecast for available markets, if thise don’t already exist.

calculate_marginal_cost(pp_series: Series) → Series#

Calculates time series of marginal costs for a 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.

Parameters:

pp_series (pandas.Series) – Series containing power plant data.

Returns:

The marginal cost of the power plant.

Return type:

pandas.Series

Notes

  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.

calculate_market_price_forecast(market_id)#

Calculates the merit order price forecast 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.

Returns:

The merit order price forecast.

Return type:

pd.Series

Notes

  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.

calculate_residual_load_forecast(market_id) → Series#

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.

Returns:

The residual demand forecast.

Return type:

pd.Series

Notes

  1. Selects VRE power plants (wind_onshore, wind_offshore, solar) from the powerplants_units 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.

get_registered_market_participants(market_id)#

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.

Parameters:

market_id (str) – The market ID.

Returns:

The registered market participants.

Return type:

pd.DataFrame

save_forecasts(path)#

Saves the forecasts to a csv file located at the specified path.

Parameters:

path (str) – The path to save the forecasts to.

Raises:

ValueError – If no forecasts are provided, an error message is logged.

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

Sets the forecast for a given 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.

Parameters:

  • data (pd.DataFrame | pd.Series | None) – The forecast data.

  • prefix (str) – The prefix of the column.

Example

>>> forecaster = CsvForecaster(index=pd.Series([1, 2, 3]))
>>> forecaster.set_forecast(pd.Series([22, 25, 17], name='temperature'), prefix='location_1_')
>>> print(forecaster['location_1_temperature'])
class assume.common.forecasts.Forecaster(index: Series)#

Bases: object

Forecaster represents a base class for forecasters based on 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.

index#

The index of the forecasts.

Type:

pandas.Series

Parameters:

index (pandas.Series) – The index of the forecasts.

Example

>>> forecaster = Forecaster(index=pd.Series([1, 2, 3]))
>>> forecast = forecaster['temperature']
>>> print(forecast)
get_availability(unit: str) → Series#

Returns the availability of a given unit as a pandas Series based on the provided index.

Parameters:

unit (str) – The unit.

Returns:

The availability of the unit.

Return type:

pd.Series

Example: >>> forecaster = Forecaster(index=pd.Series([1, 2, 3])) >>> availability = forecaster.get_availability(‘unit_1’) >>> print(availability)

get_price(fuel_type: str) → Series#

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

Parameters:

fuel_type (str) – The fuel type.

Returns:

The price of the fuel.

Return type:

pd.Series

Example

>>> forecaster = Forecaster(index=pd.Series([1, 2, 3]))
>>> price = forecaster.get_price('lignite')
>>> print(price)
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

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. If the optional parameters are constant values, they are converted to pandas Series with the provided index. If the optional parameters are lists, they are converted to pandas Series with the provided index and the corresponding values.

index#

The index of the forecasts.

Type:

pandas.Series

availability#

The availability of the power plants.

Type:

float | list, optional

fuel_price#

The fuel price.

Type:

float | list, optional

co2_price#

The CO2 price.

Type:

float | list, optional

demand#

The demand.

Type:

float | list, optional

price_forecast#

The price forecast.

Type:

float | list, optional

Parameters:

  • index (pandas.Series) – The index of the forecasts.

  • availability (float | list, optional) – The availability of the power plants.

  • fuel_price (float | list, optional) – The fuel price.

  • co2_price (float | list, optional) – The CO2 price.

  • demand (float | list, optional) – The demand.

  • price_forecast (float | list, optional) – The price forecast.

  • *args – Variable length argument list.

  • **kwargs – Arbitrary keyword arguments.

Example

>>> forecaster = NaiveForecast(demand=100, co2_price=10, fuel_price=10, availability=1, price_forecast=50)
>>> print(forecaster['demand'])

>>> forecaster = NaiveForecast(index=pd.Series([1, 2, 3]), demand=[100, 200, 300], co2_price=[10, 20, 30])
>>> print(forecaster["demand"][2])
class assume.common.forecasts.RandomForecaster(index: Series, powerplants_units: dict[str, pandas.core.series.Series] = {}, sigma: float = 0.02, *args, **kwargs)#

Bases: CsvForecaster

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.

index#

The index of the forecasts.

Type:

pandas.Series

powerplants_units#

The power plants.

Type:

dict[str, pandas.Series]

sigma#

The standard deviation of the noise.

Type:

float

Parameters:

Example

>>> forecaster = RandomForecaster(index=pd.Series([1, 2, 3]))
>>> forecaster.set_forecast(pd.Series([22, 25, 17], name='temperature'), prefix='location_1_')
>>> print(forecaster['location_1_temperature'])

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

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: ~typing.Callable[[~mango.agent.core.Agent], bool] = <function MarketConfig.<lambda>>, param_dict: dict = <factory>, addr: str = ' ', aid: str = ' ')#

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: str = ' '#
aid: str = ' '#
eligible_obligations_lambda()#
market_id: str = 'market'#
market_mechanism: str = 'pay_as_clear'#
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 = 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[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.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_id (str) – the id 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.datetime, numbers.Number]#
accepted_volume: Number | dict[datetime.datetime, numbers.Number]#
agent_id: str#
bid_id: str#
end_time: datetime#
node: str#
only_hours: OnlyHours | None#
price: Number#
start_time: datetime#
volume: Number | dict[datetime.datetime, numbers.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[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.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.outputs module#

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, db_engine=None, export_csv_path: str = '', save_frequency_hours: int = None, learning_mode: bool = False, evaluation_mode: bool = False, additional_kpis: dict[str, assume.common.outputs.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.

  • db_engine – 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.

  • 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.

check_for_tensors(data: Series)#

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

Parameters:

data (pandas.Series) – The data to be checked.

del_similar_runs()#

Deletes all similar runs from the database based on the simulation ID. This ensures that we overwrite simulations results when restarting one. Please note that a simulation which you also want to keep need to be assigned anew ID.

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()#

Retrieves the total reward for each learning unit.

Returns:

The total reward for each learning unit.

Return type:

np.array

handle_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.

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, pandas.core.frame.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.

write_market_dispatch(data: any)#

Writes the planned dispatch of the units after the market clearing to a CSV and database.

Parameters:

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

write_market_orders(market_orders: any, market_id: str)#

Writes market orders to the corresponding data frame.

Parameters:

  • market_orders (any) – The market orders.

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

write_market_results(market_meta: dict)#

Writes market results to the corresponding data frame.

Parameters:

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

write_rl_params(rl_params: dict)#

Writes the RL parameters such as reward, regret, and profit to the corresponding data frame.

Parameters:

rl_params (dict) – The RL parameters.

write_unit_dispatch(data: any)#

Writes the actual dispatch of the units to a CSV and database.

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 stores it in the database and CSV.

Parameters:

unit_info (dict) – The unit information.

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.

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

use_portfolio_opt#

Whether to use portfolio optimization.

Type:

bool

portfolio_strategy#

The portfolio strategy.

Type:

BaseStrategy

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:
async add_unit(unit: BaseUnit) → None#

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]#

Formulates the bid to the market according to the bidding strategy of the each unit individually.

Parameters:

  • market (MarketConfig) – The market to formulate bids for.

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

Returns:

The orderbook that is submitted as a bid to the market.

Return type:

OrderBook

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

Formulates the bid to the market according to the bidding strategy of the unit operator.

Parameters:

  • market (MarketConfig) – The market to formulate bids for.

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

Returns:

The orderbook that is submitted as a bid to the market.

Return type:

OrderBook

Note

Placeholder for future portfolio optimization.

get_actual_dispatch(product_type: str, last: datetime) → tuple[pandas.core.frame.DataFrame, list[pandas.core.frame.DataFrame]]#

Retrieves the actual 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) – the last date until which the dispatch was already sent

Returns:

market_dispatch and unit_dispatch dataframes

Return type:

tuple[pd.DataFrame, list[pd.DataFrame]]

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

Handles the data request received from other agents.

Parameters:
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:
handle_registration_feedback(content: RegistrationMessage, meta: MetaDict) → None#

Handles the feedback received from a market regarding registration.

Parameters:
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#

Register a market.

Parameters:

market (MarketConfig) – The market to register.

set_unit_dispatch(orderbook: list[assume.common.market_objects.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 submit_bids(opening: OpeningMessage, meta: MetaDict) → None#

Formulates an orderbook and sends it to the market.

Parameters:

Note

This function will accomodate the portfolio optimization in the future.

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.

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

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(products_index: DatetimeIndex, marketconfig: MarketConfig) → None#

Sends the current rl_strategy update to the output agent.

Parameters:
write_to_learning(products_index: DatetimeIndex, marketconfig: MarketConfig, obs_dim: int, act_dim: int, device: str, learning_unit_count: int) → None#

Writes learning results to the learning agent.

Parameters:

  • products_index (pandas.DatetimeIndex) – The index of all products.

  • marketconfig (MarketConfig) – The market configuration.

  • obs_dim (int) – The observation dimension.

  • act_dim (int) – The action dimension.

  • device (str) – The device used for learning.

  • learning_unit_count (int) – The count of learning units.

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.

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.datetime2timestamp(datetime: datetime)#
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:
Returns:

List of available products.

Return type:

list[MarketProduct]

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

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

Parameters:

orderbook (Orderbook) – The orderbook.

Returns:

The index containing all start times of orders in orderbook and all inbetween.

Return type:

pd.DatetimeIndex

assume.common.utils.get_test_demand_orders(power: ndarray)#

Get test demand orders.

Parameters:

power (numpy.ndarray) – Power array.

Returns:

DataFrame of demand orders.

Return type:

pandas.DataFrame

Examples

>>> power = np.array([100, 200, 150])
>>> get_test_demand_orders(power)
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.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: list[assume.common.market_objects.Order])#

Separate orders with several hours into single hour orders.

Parameters:

orderbook (Orderbook) – The orderbook.

Returns:

The updated orderbook.

Return type:

list

Notes

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

assume.common.utils.timestamp2datetime(timestamp: float)#
assume.common.utils.visualize_orderbook(order_book: list[assume.common.market_objects.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:
assume.common.grid_utils.add_generators(network: Network, generators: DataFrame) → None#

Add generators normally to the grid

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

Add loads normally to the grid

Parameters:
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, pandas.core.frame.DataFrame])#

Generates the pypsa grid from a grid dictionary. Does not add the generators, as they are added in different ways, depending on wether 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

Module contents#

Contents