#################################################################################
# The Institute for the Design of Advanced Energy Systems Integrated Platform
# Framework (IDAES IP) was produced under the DOE Institute for the
# Design of Advanced Energy Systems (IDAES).
#
# Copyright (c) 2018-2024 by the software owners: The Regents of the
# University of California, through Lawrence Berkeley National Laboratory,
# National Technology & Engineering Solutions of Sandia, LLC, Carnegie Mellon
# University, West Virginia University Research Corporation, et al.
# All rights reserved. Please see the files COPYRIGHT.md and LICENSE.md
# for full copyright and license information.
#################################################################################
import os
from abc import ABC, abstractmethod
import datetime
import pandas as pd
import pyomo.environ as pyo
from pyomo.opt.base.solvers import OptSolver
from pyomo.common.dependencies import attempt_import
from idaes.apps.grid_integration.utils import convert_marginal_costs_to_actual_costs
import idaes.logger as idaeslog
egret, egret_avail = attempt_import("egret")
if egret_avail:
from egret.model_library.transmission import tx_utils
_logger = idaeslog.getLogger(__name__)
class AbstractBidder(ABC):
"""
The abstract class for all the bidder and self-schedulers.
"""
@abstractmethod
def update_day_ahead_model(self, **kwargs):
"""
Update the day-ahead model (advance timesteps) with necessary parameters in kwargs.
Arguments:
kwargs: necessary profiles to update the underlying model. {stat_name: [...]}
Returns:
None
"""
@abstractmethod
def update_real_time_model(self, **kwargs):
"""
Update the real-time model (advance timesteps) with necessary parameters in kwargs.
Arguments:
kwargs: necessary profiles to update the underlying model. {stat_name: [...]}
Returns:
None
"""
@abstractmethod
def compute_day_ahead_bids(self, date, hour, **kwargs):
"""
Solve the model to bid/self-schedule into the day-ahead market. After solving,
record the schedule from the solve.
Arguments:
date: current simulation date
hour: current simulation hour
**kwargs: other information to record
Returns:
None
"""
@abstractmethod
def compute_real_time_bids(self, date, hour, **kwargs):
"""
Solve the model to bid/self-schedule into the real-time market. After solving,
record the schedule from the solve.
Arguments:
date: current simulation date
hour: current simulation hour
**kwargs: other information to record
Returns:
None
"""
@abstractmethod
def write_results(self, path):
"""
This methods writes the saved results into an csv file.
Arguments:
path: the path to write the results.
Return:
None
"""
@abstractmethod
def formulate_DA_bidding_problem(self):
"""
Formulate the day-ahead bidding optimization problem by adding necessary
parameters, constraints, and objective function.
Arguments:
None
Returns:
None
"""
@abstractmethod
def formulate_RT_bidding_problem(self):
"""
Formulate the real-time bidding optimization problem by adding necessary
parameters, constraints, and objective function.
Arguments:
None
Returns:
None
"""
@abstractmethod
def record_bids(self, bids, model, date, hour):
"""
This function records the bids (schedule) and the details in the
underlying bidding model.
Arguments:
bids: the obtained bids for this date
model: the model we obtained bids from
date: the date we bid into
hour: the hour we bid into
Returns:
None
"""
@property
@abstractmethod
def generator(self):
return "AbstractGenerator"
def _check_inputs(self):
"""
Check if the inputs to construct the tracker is valid. If not raise errors.
"""
self._check_bidding_model_object()
self._check_n_scenario()
self._check_solver()
def _check_bidding_model_object(self):
"""
Check if tracking model object has the necessary methods and attributes.
"""
method_list = ["populate_model", "update_model"]
attr_list = ["power_output", "total_cost", "model_data"]
msg = "Bidding model object does not have a "
for m in method_list:
obtained_m = getattr(self.bidding_model_object, m, None)
if obtained_m is None:
raise AttributeError(
msg
+ f"{m}() method. The bidder object needs the users to implement this method in their model object."
)
for attr in attr_list:
obtained_attr = getattr(self.bidding_model_object, attr, None)
if obtained_attr is None:
raise AttributeError(
msg
+ f"{attr} property. The bidder object needs the users to specify this property in their model object."
)
def _check_n_scenario(self):
"""
Check if the number of LMP scenarios is an integer and greater than 0.
"""
# check if it is an integer
if not isinstance(self.n_scenario, int):
raise TypeError(
f"The number of LMP scenarios should be an integer, but a {type(self.n_scenario).__name__} was given."
)
if self.n_scenario <= 0:
raise ValueError(
f"The number of LMP scenarios should be greater than zero, but {self.n_scenario} was given."
)
def _check_solver(self):
"""
Check if provides solver is a valid Pyomo solver object.
"""
if not isinstance(self.solver, OptSolver):
raise TypeError(
f"The provided solver {self.solver} is not a valid Pyomo solver."
)
class StochasticProgramBidder(AbstractBidder):
"""
Template class for bidders that use scenario-based stochastic programs.
"""
def __init__(
self,
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
n_scenario,
solver,
forecaster,
real_time_underbid_penalty,
):
"""
Initializes the stochastic bidder object.
Arguments:
bidding_model_object: the model object for bidding
day_ahead_horizon: number of time periods in the day-ahead bidding problem
real_time_horizon: number of time periods in the real-time bidding problem
n_scenario: number of uncertain LMP scenarios
solver: a Pyomo mathematical programming solver object
forecaster: an initialized LMP forecaster object
real_time_underbid_penalty: penalty for RT power bid that's less than DA power bid, non-negative
Returns:
None
"""
self.bidding_model_object = bidding_model_object
self.day_ahead_horizon = day_ahead_horizon
self.real_time_horizon = real_time_horizon
self.n_scenario = n_scenario
self.solver = solver
self.forecaster = forecaster
self.real_time_underbid_penalty = real_time_underbid_penalty
self._check_inputs()
self.generator = self.bidding_model_object.model_data.gen_name
# day-ahead model
self.day_ahead_model = self.formulate_DA_bidding_problem()
self.real_time_model = self.formulate_RT_bidding_problem()
# declare a list to store results
self.bids_result_list = []
def _set_up_bidding_problem(self, horizon):
"""
Set up the base stochastic programming bidding problems.
Arguments:
horizon: number of time periods in the bidding problem
Returns:
pyomo.core.base.PyomoModel.ConcreteModel: base bidding model
"""
model = pyo.ConcreteModel()
model.SCENARIOS = pyo.Set(initialize=range(self.n_scenario))
model.fs = pyo.Block(model.SCENARIOS)
for i in model.SCENARIOS:
self.bidding_model_object.populate_model(model.fs[i], horizon)
self._save_power_outputs(model)
self._add_bidding_params(model)
self._add_bidding_vars(model)
self._add_bidding_objective(model)
return model
def formulate_DA_bidding_problem(self):
"""
Set up the day-ahead stochastic programming bidding problems.
Returns:
pyomo.core.base.PyomoModel.ConcreteModel: base bidding model
"""
model = self._set_up_bidding_problem(self.day_ahead_horizon)
self._add_DA_bidding_constraints(model)
# do not relax the DA offering UB
for i in model.SCENARIOS:
model.fs[i].real_time_underbid_power.fix(0)
return model
def formulate_RT_bidding_problem(self):
"""
Set up the real-time stochastic programming bidding problems.
Returns:
pyomo.core.base.PyomoModel.ConcreteModel: base bidding model
"""
model = self._set_up_bidding_problem(self.real_time_horizon)
self._add_RT_bidding_constraints(model)
# relax the DA offering UB
for i in model.SCENARIOS:
model.fs[i].real_time_underbid_power.unfix()
return model
def _save_power_outputs(self, model):
"""
Create references of the power output variable in each price scenario
block.
Arguments:
None
Returns:
None
"""
for i in model.SCENARIOS:
# get the power output
power_output_name = self.bidding_model_object.power_output
model.fs[i].power_output_ref = pyo.Reference(
getattr(model.fs[i], power_output_name)
)
return
def _add_bidding_params(self, model):
"""
Add necessary bidding parameters to the model, i.e., market energy price.
Arguments:
model: bidding model
Returns:
None
"""
for i in model.SCENARIOS:
time_index = model.fs[i].power_output_ref.index_set()
model.fs[i].day_ahead_energy_price = pyo.Param(
time_index, initialize=0, mutable=True
)
model.fs[i].real_time_energy_price = pyo.Param(
time_index, initialize=0, mutable=True
)
model.fs[i].real_time_underbid_penalty = pyo.Param(
initialize=self.real_time_underbid_penalty, mutable=True
)
return
def _add_bidding_vars(self, model):
"""
Add necessary bidding parameters to the model, i.e., market energy price.
Arguments:
model: bidding model
Returns:
None
"""
def relaxed_day_ahead_power_ub_rule(fs, t):
return (
fs.power_output_ref[t] + fs.real_time_underbid_power[t]
>= fs.day_ahead_power[t]
)
for i in model.SCENARIOS:
time_index = model.fs[i].power_output_ref.index_set()
model.fs[i].day_ahead_power = pyo.Var(
time_index, initialize=0, within=pyo.NonNegativeReals
)
model.fs[i].real_time_underbid_power = pyo.Var(
time_index, initialize=0, within=pyo.NonNegativeReals
)
model.fs[i].day_ahead_power_ub = pyo.Constraint(
time_index, rule=relaxed_day_ahead_power_ub_rule
)
return
def _add_bidding_objective(self, model):
"""
Add objective function to the model, i.e., maximizing the expected profit
of the energy system.
Arguments:
model: bidding model
Returns:
None
"""
# declare an empty objective
model.obj = pyo.Objective(expr=0, sense=pyo.maximize)
for k in model.SCENARIOS:
time_index = model.fs[k].power_output_ref.index_set()
# currently .total_cost is a tuple of 2 items
# the first item is the name of the cost expression
# the second item is the weight for the cost
cost_name = self.bidding_model_object.total_cost[0]
cost = getattr(model.fs[k], cost_name)
weight = self.bidding_model_object.total_cost[1]
for t in time_index:
model.obj.expr += (
model.fs[k].day_ahead_energy_price[t]
* model.fs[k].day_ahead_power[t]
+ model.fs[k].real_time_energy_price[t]
* (model.fs[k].power_output_ref[t] - model.fs[k].day_ahead_power[t])
- weight * cost[t]
- model.fs[k].real_time_underbid_penalty
* model.fs[k].real_time_underbid_power[t]
)
return
def _compute_bids(
self,
day_ahead_price,
real_time_energy_price,
date,
hour,
model,
power_var_name,
energy_price_param_name,
market,
):
"""
Solve the model to bid into the markets. After solving, record the bids from the solve.
Arguments:
day_ahead_price: day-ahead price forecasts needed to solve the bidding problem
real_time_energy_price: real-time price forecasts needed to solve the bidding problem
date: current simulation date
hour: current simulation hour
model: bidding model
power_var_name: the name of the power output (str)
energy_price_param_name: the name of the energy price forecast params (str)
market: the market name (str), e.g., Day-ahead, real-time
Returns:
dict: the obtained bids
"""
# update the price forecasts
self._pass_price_forecasts(model, day_ahead_price, real_time_energy_price)
self.solver.solve(model, tee=True)
bids = self._assemble_bids(
model,
power_var_name=power_var_name,
energy_price_param_name=energy_price_param_name,
hour=hour,
)
self.record_bids(bids, model=model, date=date, hour=hour, market=market)
return bids
def compute_day_ahead_bids(self, date, hour=0):
"""
Solve the model to bid into the day-ahead market. After solving, record
the bids from the solve.
Arguments:
date: current simulation date
hour: current simulation hour
Returns:
dict: the obtained bids
"""
(
day_ahead_price,
real_time_energy_price,
) = self.forecaster.forecast_day_ahead_and_real_time_prices(
date=date,
hour=hour,
bus=self.bidding_model_object.model_data.bus,
horizon=self.day_ahead_horizon,
n_samples=self.n_scenario,
)
return self._compute_bids(
day_ahead_price=day_ahead_price,
real_time_energy_price=real_time_energy_price,
date=date,
hour=hour,
model=self.day_ahead_model,
power_var_name="day_ahead_power",
energy_price_param_name="day_ahead_energy_price",
market="Day-ahead",
)
def compute_real_time_bids(
self, date, hour, realized_day_ahead_prices, realized_day_ahead_dispatches
):
"""
Solve the model to bid into the real-time market. After solving, record
the bids from the solve.
Arguments:
date: current simulation date
hour: current simulation hour
Returns:
dict: the obtained bids
"""
real_time_energy_price = self.forecaster.forecast_real_time_prices(
date=date,
hour=hour,
bus=self.bidding_model_object.model_data.bus,
horizon=self.real_time_horizon,
n_samples=self.n_scenario,
)
self._pass_realized_day_ahead_dispatches(realized_day_ahead_dispatches, hour)
self._pass_realized_day_ahead_prices(realized_day_ahead_prices, date, hour)
return self._compute_bids(
day_ahead_price=None,
real_time_energy_price=real_time_energy_price,
date=date,
hour=hour,
model=self.real_time_model,
power_var_name="power_output_ref",
energy_price_param_name="real_time_energy_price",
market="Real-time",
)
def _pass_realized_day_ahead_prices(self, realized_day_ahead_prices, date, hour):
"""
Pass the realized day-ahead prices into model parameters.
Arguments:
realized_day_ahead_prices: realized day-ahead prices
date: current simulation date
hour: current simulation hour
Returns:
None
"""
time_index = self.real_time_model.fs[0].day_ahead_energy_price.index_set()
# forecast the day-ahead price, if not enough realized data
if len(realized_day_ahead_prices[hour:]) < len(time_index):
forecasts = self.forecaster.forecast_day_ahead_prices(
date=date + datetime.timedelta(days=1),
hour=0,
bus=self.bidding_model_object.model_data.bus,
horizon=self.day_ahead_horizon,
n_samples=self.n_scenario,
)
for s in self.real_time_model.SCENARIOS:
for t in time_index:
try:
price = realized_day_ahead_prices[t + hour]
except IndexError:
self.real_time_model.fs[s].day_ahead_energy_price[t] = forecasts[s][
(t + hour) - 24
]
else:
self.real_time_model.fs[s].day_ahead_energy_price[t] = price
def _pass_realized_day_ahead_dispatches(self, realized_day_ahead_dispatches, hour):
"""
Pass the realized day-ahead dispatches into model and fix the corresponding variables.
Arguments:
realized_day_ahead_dispatches: realized day-ahead dispatches
hour: current simulation hour
Returns:
None
"""
time_index = self.real_time_model.fs[0].day_ahead_power.index_set()
for s in self.real_time_model.SCENARIOS:
for t in time_index:
try:
dispatch = realized_day_ahead_dispatches[t + hour]
except IndexError:
self.real_time_model.fs[s].day_ahead_power[t].unfix()
# unrelax the DA offering UB
self.real_time_model.fs[s].real_time_underbid_power[t].fix(0)
else:
self.real_time_model.fs[s].day_ahead_power[t].fix(dispatch)
# relax the DA offering UB
self.real_time_model.fs[s].real_time_underbid_power[t].unfix()
def update_day_ahead_model(self, **kwargs):
"""
This method updates the parameters in the day-ahead model based on the implemented profiles.
Arguments:
kwargs: the newly implemented stats. {stat_name: [...]}
Returns:
None
"""
self._update_model(self.day_ahead_model, **kwargs)
def update_real_time_model(self, **kwargs):
"""
This method updates the parameters in the real-time model based on the implemented profiles.
Arguments:
kwargs: the newly implemented stats. {stat_name: [...]}
Returns:
None
"""
self._update_model(self.real_time_model, **kwargs)
def _update_model(self, model, **kwargs):
"""
Update the flowsheets in all the price scenario blocks to advance time
step.
Arguments:
model: bidding model
kwargs: necessary profiles to update the underlying model. {stat_name: [...]}
Returns:
None
"""
for i in model.SCENARIOS:
self.bidding_model_object.update_model(b=model.fs[i], **kwargs)
return
def record_bids(self, bids, model, date, hour, market):
"""
This function records the bids and the details in the underlying bidding model.
Arguments:
bids: the obtained bids for this date.
model: bidding model
date: the date we bid into
hour: the hour we bid into
Returns:
None
"""
# record bids
self._record_bids(bids, date, hour, Market=market)
# record the details of bidding model
for i in model.SCENARIOS:
self.bidding_model_object.record_results(
model.fs[i], date=date, hour=hour, Scenario=i, Market=market
)
return
def _pass_price_forecasts(self, model, day_ahead_price, real_time_energy_price):
"""
Pass the price forecasts into model parameters.
Arguments:
day_ahead_price: day-ahead price forecasts needed to solve the bidding problem
real_time_energy_price: real-time price forecasts needed to solve the bidding problem
Returns:
None
"""
for i in model.SCENARIOS:
time_index = model.fs[i].real_time_energy_price.index_set()
if day_ahead_price is not None:
for t, p in zip(time_index, day_ahead_price[i]):
model.fs[i].day_ahead_energy_price[t] = p
for t, p in zip(time_index, real_time_energy_price[i]):
model.fs[i].real_time_energy_price[t] = p
return
def write_results(self, path):
"""
This methods writes the saved operation stats into an csv file.
Arguments:
path: the path to write the results.
Return:
None
"""
_logger.info("Saving bidding results to disk.")
pd.concat(self.bids_result_list).to_csv(
os.path.join(path, "bidder_detail.csv"), index=False
)
self.bidding_model_object.write_results(
path=os.path.join(path, "bidding_model_detail.csv")
)
return
@property
def generator(self):
return self._generator
@generator.setter
def generator(self, name):
self._generator = name
[docs]
class SelfScheduler(StochasticProgramBidder):
"""
Wrap a model object to self schedule into the market using stochastic programming.
"""
def __init__(
self,
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
n_scenario,
solver,
forecaster,
real_time_underbid_penalty=10000,
fixed_to_schedule=False,
):
"""
Initializes the stochastic self-scheduler object.
Arguments:
bidding_model_object: the model object for bidding
day_ahead_horizon: number of time periods in the day-ahead bidding problem
real_time_horizon: number of time periods in the real-time bidding problem
n_scenario: number of uncertain LMP scenarios
solver: a Pyomo mathematical programming solver object
forecaster: an initialized LMP forecaster object
fixed_to_schedule: If True, force market simulator to give the same schedule.
Returns:
None
"""
super().__init__(
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
n_scenario,
solver,
forecaster,
real_time_underbid_penalty,
)
self.fixed_to_schedule = fixed_to_schedule
def _add_DA_bidding_constraints(self, model):
"""
Add bidding constraints to the model, i.e., power outputs in the first
stage need to be the same across all the scenarios.
Arguments:
model: bidding model
Returns:
None
"""
# nonanticipativity constraints
def day_ahead_bidding_constraints_rule(model, s1, s2, t):
if s1 == s2:
return pyo.Constraint.Skip
return model.fs[s1].day_ahead_power[t] == model.fs[s2].day_ahead_power[t]
time_index = model.fs[model.SCENARIOS.first()].power_output_ref.index_set()
model.day_ahead_bidding_constraints = pyo.Constraint(
model.SCENARIOS,
model.SCENARIOS,
time_index,
rule=day_ahead_bidding_constraints_rule,
)
return
def _add_RT_bidding_constraints(self, model):
"""
Add bidding constraints to the model, i.e., power outputs in the first
stage need to be the same across all the scenarios.
Arguments:
model: bidding model
Returns:
None
"""
# nonanticipativity constraints
def real_time_bidding_constraints_rule(model, s1, s2, t):
if s1 == s2:
return pyo.Constraint.Skip
return model.fs[s1].power_output_ref[t] == model.fs[s2].power_output_ref[t]
time_index = model.fs[model.SCENARIOS.first()].power_output_ref.index_set()
model.real_time_bidding_constraints = pyo.Constraint(
model.SCENARIOS,
model.SCENARIOS,
time_index,
rule=real_time_bidding_constraints_rule,
)
return
def _assemble_bids(self, model, power_var_name, energy_price_param_name, hour):
"""
This methods extract the bids out of the stochastic programming model and
organize them into self-schedule bids.
For thermal generators, startup times and costs are set to 0. And the bid price for power outside of p_min and p_max are 0.
Arguments:
model: bidding model
power_var_name: the name of the power output (str)
energy_price_param_name: the name of the energy price forecast params (str)
hour: current simulation hour
Returns:
dict: the bid we computed.
"""
bids = {}
is_thermal = self.bidding_model_object.model_data.generator_type == "thermal"
power_output_var = getattr(model.fs[0], power_var_name)
time_index = power_output_var.index_set()
for t_idx in time_index:
t = t_idx + hour
bids[t] = {}
bids[t][self.generator] = {
"p_max": round(pyo.value(power_output_var[t_idx]), 4),
"p_min": self.bidding_model_object.model_data.p_min,
}
if self.fixed_to_schedule:
bids[t][self.generator]["p_min"] = bids[t][self.generator]["p_max"]
bids[t][self.generator]["fixed_commitment"] = (
1 if bids[t][self.generator]["p_min"] > 0 else 0
)
if is_thermal:
bids[t][self.generator]["min_up_time"] = 0
bids[t][self.generator]["min_down_time"] = 0
bids[t][self.generator]["startup_fuel"] = [
(bids[t][self.generator]["min_down_time"], 0)
]
bids[t][self.generator]["startup_cost"] = [
(bids[t][self.generator]["min_down_time"], 0)
]
if is_thermal:
bids[t][self.generator]["p_cost"] = [
(bids[t][self.generator]["p_min"], 0),
(bids[t][self.generator]["p_max"], 0),
]
bids[t][self.generator]["startup_capacity"] = bids[t][self.generator][
"p_min"
]
bids[t][self.generator]["shutdown_capacity"] = bids[t][self.generator][
"p_min"
]
return bids
def _record_bids(self, bids, date, hour, **kwargs):
"""
This function records the bids (schedule) we computed for the given date into a
DataFrame and temporarily stores the DataFrame in an instance attribute
list called bids_result_list.
Arguments:
bids: the obtained bids (schedule) for this date.
date: the date we bid into
hour: the hour we bid into
Returns:
None
"""
df_list = []
for t in bids:
for g in bids[t]:
result_dict = {}
result_dict["Generator"] = g
result_dict["Date"] = date
if hour is not None:
result_dict["Hour"] = hour
result_dict["Horizon"] = t
result_dict["Bid Power [MW]"] = bids[t][g].get("p_max")
result_dict["Bid Min Power [MW]"] = bids[t][g].get("p_min")
for k, v in kwargs.items():
result_dict[k] = v
result_df = pd.DataFrame.from_dict(result_dict, orient="index")
df_list.append(result_df.T)
# save the result to object property
# wait to be written when simulation ends
self.bids_result_list.append(pd.concat(df_list))
[docs]
class Bidder(StochasticProgramBidder):
"""
Wrap a model object to bid into the market using stochastic programming.
"""
def __init__(
self,
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
n_scenario,
solver,
forecaster,
real_time_underbid_penalty=10000,
):
"""
Initializes the bidder object.
Arguments:
bidding_model_object: the model object for bidding
day_ahead_horizon: number of time periods in the day-ahead bidding problem
real_time_horizon: number of time periods in the real-time bidding problem
n_scenario: number of uncertain LMP scenarios
solver: a Pyomo mathematical programming solver object
forecaster: an initialized LMP forecaster object
real_time_underbid_penalty: penalty for RT power bid that's less than DA power bid, non-negative
Returns:
None
"""
super().__init__(
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
n_scenario,
solver,
forecaster,
real_time_underbid_penalty,
)
def _add_DA_bidding_constraints(self, model):
"""
Add bidding constraints to the model, i.e., the bid curves need to be
nondecreasing.
Arguments:
model: bidding model
Returns:
None
"""
def day_ahead_bidding_constraints_rule(model, s1, s2, t):
if s1 == s2:
return pyo.Constraint.Skip
return (
model.fs[s1].day_ahead_power[t] - model.fs[s2].day_ahead_power[t]
) * (
model.fs[s1].day_ahead_energy_price[t]
- model.fs[s2].day_ahead_energy_price[t]
) >= 0
time_index = model.fs[model.SCENARIOS.first()].power_output_ref.index_set()
model.day_ahead_bidding_constraints = pyo.Constraint(
model.SCENARIOS,
model.SCENARIOS,
time_index,
rule=day_ahead_bidding_constraints_rule,
)
return
def _add_RT_bidding_constraints(self, model):
"""
Add bidding constraints to the model, i.e., the bid curves need to be
nondecreasing.
Arguments:
model: bidding model
Returns:
None
"""
def real_time_bidding_constraints_rule(model, s1, s2, t):
if s1 == s2:
return pyo.Constraint.Skip
return (
model.fs[s1].power_output_ref[t] - model.fs[s2].power_output_ref[t]
) * (
model.fs[s1].real_time_energy_price[t]
- model.fs[s2].real_time_energy_price[t]
) >= 0
time_index = model.fs[model.SCENARIOS.first()].power_output_ref.index_set()
model.real_time_bidding_constraints = pyo.Constraint(
model.SCENARIOS,
model.SCENARIOS,
time_index,
rule=real_time_bidding_constraints_rule,
)
return
def _assemble_bids(self, model, power_var_name, energy_price_param_name, hour):
"""
This methods extract the bids out of the stochastic programming model and
organize them into ( MWh, $ ) pairs.
Arguments:
model: bidding model
power_var_name: the name of the power output (str)
energy_price_param_name: the name of the energy price forecast params (str)
hour: current simulation hour
Returns:
bids: the bid we computed.
"""
bids = {}
gen = self.generator
for i in model.SCENARIOS:
power_output_var = getattr(model.fs[i], power_var_name)
energy_price_param = getattr(model.fs[i], energy_price_param_name)
time_index = power_output_var.index_set()
for t in time_index:
if t not in bids:
bids[t] = {}
if gen not in bids[t]:
bids[t][gen] = {}
power = round(pyo.value(power_output_var[t]), 2)
marginal_cost = round(pyo.value(energy_price_param[t]), 2)
# if power lower than pmin, e.g., power = 0, we need to skip this
# solution, because Prescient is not expecting any power output lower
# than pmin in the bids
if power < self.bidding_model_object.model_data.p_min:
continue
elif power in bids[t][gen]:
bids[t][gen][power] = min(bids[t][gen][power], marginal_cost)
else:
bids[t][gen][power] = marginal_cost
for t in time_index:
# always include pmin in the cost curve, but include the other points if required
if self.bidding_model_object.model_data.include_default_p_cost:
p_cost_add = self.bidding_model_object.model_data.p_cost
else:
p_cost_add = self.bidding_model_object.model_data.p_cost[0:1]
for power, marginal_cost in p_cost_add:
if round(power, 2) not in bids[t][gen]:
bids[t][gen][power] = marginal_cost
pmin = self.bidding_model_object.model_data.p_min
# sort the curves by power
bids[t][gen] = dict(sorted(bids[t][gen].items()))
# make sure the curve is nondecreasing
pre_power = pmin
for power, marginal_cost in bids[t][gen].items():
# ignore pmin, because min load cost is special
if pre_power == pmin:
pre_power = power
continue
bids[t][gen][power] = max(bids[t][gen][power], bids[t][gen][pre_power])
pre_power = power
# calculate the actual cost
bids[t][gen] = convert_marginal_costs_to_actual_costs(
list(bids[t][gen].items())
)
# check if bids are convex
for t in bids:
for gen in bids[t]:
temp_curve = {
"data_type": "cost_curve",
"cost_curve_type": "piecewise",
"values": bids[t][gen],
}
try:
tx_utils.validate_and_clean_cost_curve(
curve=temp_curve,
curve_type="cost_curve",
p_min=min([p[0] for p in bids[t][gen]]),
p_max=max([p[0] for p in bids[t][gen]]),
gen_name=gen,
t=t,
)
except NameError:
raise RuntimeError(
"'egret' must be installed to use this functionality"
)
# create full bids: this includes info in addition to costs
full_bids = {}
for t_idx in bids:
t = t_idx + hour
full_bids[t] = {}
for gen in bids[t_idx]:
full_bids[t][gen] = {}
full_bids[t][gen]["p_cost"] = bids[t_idx][gen]
full_bids[t][gen]["p_min"] = min([p[0] for p in bids[t_idx][gen]])
full_bids[t][gen]["p_max"] = max([p[0] for p in bids[t_idx][gen]])
full_bids[t][gen]["p_min_agc"] = min([p[0] for p in bids[t_idx][gen]])
full_bids[t][gen]["p_max_agc"] = max([p[0] for p in bids[t_idx][gen]])
full_bids[t][gen]["startup_capacity"] = full_bids[t][gen]["p_min"]
full_bids[t][gen]["shutdown_capacity"] = full_bids[t][gen]["p_min"]
fixed_commitment = getattr(
self.bidding_model_object.model_data, "fixed_commitment", None
)
if fixed_commitment is not None:
full_bids[t][gen]["fixed_commitment"] = fixed_commitment
return full_bids
def _record_bids(self, bids, date, hour, **kwargs):
"""
This method records the bids we computed for the given date into a
DataFrame. This DataFrame has the following columns: gen, date, hour,
power 1, ..., power n, price 1, ..., price n. And concatenate the
DataFrame into a class property 'bids_result_list'. The methods then
temporarily stores the DataFrame in an instance attribute list called
bids_result_list.
Arguments:
bids: the obtained bids for this date.
date: the date we bid into
hour: the hour we bid into
Returns:
None
"""
df_list = []
for t in bids:
for gen in bids[t]:
result_dict = {}
result_dict["Generator"] = gen
result_dict["Date"] = date
result_dict["Hour"] = t
for k, v in kwargs.items():
result_dict[k] = v
pair_cnt = len(bids[t][gen]["p_cost"])
for idx, (power, cost) in enumerate(bids[t][gen]["p_cost"]):
result_dict[f"Power {idx} [MW]"] = power
result_dict[f"Cost {idx} [$]"] = cost
# place holder, in case different len of bids
while pair_cnt < self.n_scenario:
result_dict[f"Power {pair_cnt} [MW]"] = None
result_dict[f"Cost {pair_cnt} [$]"] = None
pair_cnt += 1
result_df = pd.DataFrame.from_dict(result_dict, orient="index")
df_list.append(result_df.T)
# save the result to object property
# wait to be written when simulation ends
self.bids_result_list.append(pd.concat(df_list))
return
class ParametrizedBidder(AbstractBidder):
"""
Create a parameterized bidder.
Bid the resource at different prices.
"""
def __init__(
self,
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
solver,
forecaster,
):
"""
Arguments:
bidding_model_object: pyomo model object,
the IES model object for bidding.
day_ahead_horizon: int,
number of time periods in the day-ahead bidding problem.
real_time_horizon: int,
number of time periods in the real-time bidding problem.
solver: a Pyomo mathematical programming solver object,
solver for solving the bidding problem. In this class we do not need a solver.
forecaster: forecaster object,
the forecaster to predict the generator/IES capacity factor.
"""
self.bidding_model_object = bidding_model_object
self.day_ahead_horizon = day_ahead_horizon
self.real_time_horizon = real_time_horizon
self.solver = solver
self.forecaster = forecaster
# Because ParameterizedBidder is inherited from the AbstractBidder, and we want to
# use the _check_inputs() function to check the solver and bidding_model_object.
# We must have the self.scenario attribute. In this case, I set the self.n_scenario = 1 when initializing.
# However, self.n_scenario will never be used in this class.
self.n_scenario = 1
self._check_inputs()
self.generator = self.bidding_model_object.model_data.gen_name
self.bids_result_list = []
@property
def generator(self):
return self._generator
@generator.setter
def generator(self, name):
self._generator = name
def formulate_DA_bidding_problem(self):
"""
No need to formulate a DA bidding problem here.
Arguments:
None
Returns:
None
"""
pass
def formulate_RT_bidding_problem(self):
"""
No need to formulate a RT bidding problem here.
Arguments:
None
Returns:
None
"""
pass
def compute_day_ahead_bids(self, date, hour=0):
raise NotImplementedError
def compute_real_time_bids(
self,
date,
hour,
realized_day_ahead_prices,
realized_day_ahead_dispatches,
tracker_profile,
):
raise NotImplementedError
def update_day_ahead_model(self, **kwargs):
"""
No need to update the RT bidding problem here.
Arguments:
None
Returns:
None
"""
pass
def update_real_time_model(self, **kwargs):
"""
No need to update the RT bidding problem here.
Arguments:
None
Returns:
None
"""
pass
def record_bids(self, bids: dict, model, date: str, hour: int, market):
"""
This function records the bids and the details in the underlying bidding model.
The detailed bids of each time step at day-ahead and real-time planning horizon will
be recorded at bidder_detail.csv
Arguments:
bids: dictionary,
the obtained bids for this date. Keys are time step t, example as following:
bids = {t: {gen: {p_cost: float, p_max: float, p_min: float, startup_capacity: float, shutdown_capacity: float}}}
model: pyomo model object,
our bidding model.
date: str,
the date we bid into.
hour: int,
the hour we bid into.
market: str,
the market we participate.
Returns:
None
"""
# record bids
self._record_bids(bids, date, hour, Market=market)
return
def _record_bids(self, bids: dict, date: str, hour: int, **kwargs):
"""
Record the bis of each time period.
Arguments:
bids: dictionary,
the obtained bids for this date. Keys are time step t, example as following:
bids = {t: {gen: {p_cost: float, p_max: float, p_min: float, startup_capacity: float, shutdown_capacity: float}}}
date: str,
the date we bid into.
hour: int,
the hour we bid into.
Returns:
None
"""
df_list = []
for t in bids:
for gen in bids[t]:
result_dict = {}
result_dict["Generator"] = gen
result_dict["Date"] = date
result_dict["Hour"] = t
for k, v in kwargs.items():
result_dict[k] = v
num_bid_pairs = len(bids[t][gen]["p_cost"])
for idx, (power, cost) in enumerate(bids[t][gen]["p_cost"]):
result_dict[f"Power {idx} [MW]"] = power
result_dict[f"Cost {idx} [$]"] = cost
result_df = pd.DataFrame.from_dict(result_dict, orient="index")
df_list.append(result_df.T)
# save the result to object property
# wait to be written when simulation ends
self.bids_result_list.append(pd.concat(df_list))
return
def write_results(self, path):
"""
This methods writes the saved operation stats into an csv file.
Arguments:
path: str or Pathlib object,
the path to write the results.
Return:
None
"""
_logger.info("Saving bidding results to disk.")
pd.concat(self.bids_result_list).to_csv(
os.path.join(path, "bidder_detail.csv"), index=False
)
return
[docs]
class PEMParametrizedBidder(ParametrizedBidder):
"""
Renewable (PV or Wind) + PEM bidder that uses parameterized bid curves.
"""
def __init__(
self,
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
solver,
forecaster,
renewable_mw,
pem_marginal_cost,
pem_mw,
real_time_bidding_only=False,
):
"""
Arguments:
bidding_model_object: pyomo model object,
the IES model object for bidding.
day_ahead_horizon: int,
number of time periods in the day-ahead bidding problem.
real_time_horizon: int,
number of time periods in the real-time bidding problem.
solver: a Pyomo mathematical programming solver object,
solver for solving the bidding problem. In this class we do not need a solver.
forecaster: forecaster object,
the forecaster to predict the generator/IES capacity factor.
renewable_mw: int or float,
maximum renewable energy system capacity.
pem_marginal_cost: int or float,
cost/MW, above which all available wind energy will be sold to grid;
below which, make hydrogen and sell remainder of wind to grid.
pem_mw: int or float,
maximum PEM capacity limits how much energy is bid at the `pem_marginal_cost`.
real_time_bidding_only: bool,
if True, do real-time bidding only.
"""
super().__init__(
bidding_model_object,
day_ahead_horizon,
real_time_horizon,
solver,
forecaster,
)
self.renewable_mw = renewable_mw
self.renewable_marginal_cost = 0
self.pem_marginal_cost = pem_marginal_cost
self.pem_mw = pem_mw
self.real_time_bidding_only = real_time_bidding_only
self._check_power()
def _check_power(self):
"""
Check the power of PEM should not exceed the power of renewables
"""
if self.pem_mw >= self.renewable_mw:
raise ValueError(f"The power of PEM is greater than the renewable power.")
[docs]
def compute_day_ahead_bids(self, date: str, hour=0):
"""
DA Bid: from 0 MW to (Wind Resource - PEM capacity) MW, bid $0/MWh.
from (Wind Resource - PEM capacity) MW to Wind Resource MW, bid 'pem_marginal_cost'
If Wind resource at some time is less than PEM capacity, then reduce to available resource
Arguments:
date: str,
the date we bid into.
hour: int,
the hour we bid into.
Returns:
full_bids: dictionary,
the obtained bids. Keys are time step t, example as following:
bids = {t: {gen: {p_cost: float, p_max: float, p_min: float, startup_capacity: float, shutdown_capacity: float}}}
"""
gen = self.generator
# Forecast the day-ahead wind generation
forecast = self.forecaster.forecast_day_ahead_capacity_factor(
date, hour, gen, self.day_ahead_horizon
)
full_bids = {}
for t_idx in range(self.day_ahead_horizon):
da_wind = forecast[t_idx] * self.renewable_mw
grid_wind = max(0, da_wind - self.pem_mw)
# grid wind are bidded at marginal cost = 0
# The rest of the power is bidded at the pem marginal cost
if grid_wind == 0:
bids = [(0, 0), (da_wind, self.pem_marginal_cost)]
else:
bids = [(0, 0), (grid_wind, 0), (da_wind, self.pem_marginal_cost)]
cost_curve = convert_marginal_costs_to_actual_costs(bids)
temp_curve = {
"data_type": "cost_curve",
"cost_curve_type": "piecewise",
"values": cost_curve,
}
tx_utils.validate_and_clean_cost_curve(
curve=temp_curve,
curve_type="cost_curve",
p_min=0,
p_max=max([p[0] for p in cost_curve]),
gen_name=gen,
t=t_idx,
)
t = t_idx + hour
full_bids[t] = {}
full_bids[t][gen] = {}
full_bids[t][gen]["p_cost"] = cost_curve
full_bids[t][gen]["p_min"] = 0
full_bids[t][gen]["p_max"] = da_wind
full_bids[t][gen]["startup_capacity"] = da_wind
full_bids[t][gen]["shutdown_capacity"] = da_wind
self._record_bids(full_bids, date, hour, Market="Day-ahead")
return full_bids
[docs]
def compute_real_time_bids(
self, date, hour, realized_day_ahead_dispatches, realized_day_ahead_prices
):
"""
RT Bid: from 0 MW to (Wind Resource - PEM capacity) MW, bid $0/MWh.
from (Wind Resource - PEM capacity) MW to Wind Resource MW, bid 'pem_marginal_cost'
Arguments:
date: str,
the date we bid into
hour: int,
the hour we bid into
Returns:
full_bids: dictionary,
the obtained bids. Keys are time step t, example as following:
bids = {t: {gen: {p_cost: float, p_max: float, p_min: float, startup_capacity: float, shutdown_capacity: float}}}
"""
gen = self.generator
forecast = self.forecaster.forecast_real_time_capacity_factor(
date, hour, gen, self.real_time_horizon
)
full_bids = {}
for t_idx in range(self.real_time_horizon):
rt_wind = forecast[t_idx] * self.renewable_mw
# if we participate in both DA and RT market
if not self.real_time_bidding_only:
try:
da_dispatch = realized_day_ahead_dispatches[t_idx + hour]
except IndexError:
# When having indexerror, it must be the period that we are looking ahead. It is ok to set da_dispatch to 0
da_dispatch = 0
# if we only participates in the RT market, then we do not consider the DA commitment
else:
da_dispatch = 0
avail_rt_wind = max(0, rt_wind - da_dispatch)
grid_wind = max(0, avail_rt_wind - self.pem_mw)
if avail_rt_wind == 0:
bids = [(0, 0), (0, 0)]
else:
if grid_wind == 0:
bids = [(0, 0), (avail_rt_wind, self.pem_marginal_cost)]
else:
bids = [
(0, 0),
(grid_wind, 0),
(avail_rt_wind, self.pem_marginal_cost),
]
cost_curve = convert_marginal_costs_to_actual_costs(bids)
temp_curve = {
"data_type": "cost_curve",
"cost_curve_type": "piecewise",
"values": cost_curve,
}
tx_utils.validate_and_clean_cost_curve(
curve=temp_curve,
curve_type="cost_curve",
p_min=0,
p_max=max([p[0] for p in cost_curve]),
gen_name=gen,
t=t_idx,
)
t = t_idx + hour
full_bids[t] = {}
full_bids[t][gen] = {}
full_bids[t][gen]["p_cost"] = cost_curve
full_bids[t][gen]["p_min"] = 0
full_bids[t][gen]["p_max"] = max([p[0] for p in cost_curve])
full_bids[t][gen]["startup_capacity"] = rt_wind
full_bids[t][gen]["shutdown_capacity"] = rt_wind
self._record_bids(full_bids, date, hour, Market="Real-time")
return full_bids
