File size: 6,891 Bytes
88bc772 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | from dataclasses import dataclass
from typing import Dict, Optional, Any, Tuple, List
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
@dataclass()
class DaySolution:
"""
Represents the solution of the optimization for a single day.
date : The date of the optimization day.
input : The input data used for the optimization
schedule : The resulting schedule from the optimization
status : The status of the optimization
solver : The solver name used for the optimization.
"""
date: pd.Timestamp
input: pd.DataFrame
schedule: pd.DataFrame
status: str
solver: str
def plot_results(self, config: Optional[Dict[str, str]] = None):
"""
Create a multi-panel plot showing:
Reserve bids (FCR, aFRR UP/DOWN) and SoC
Energy prices over time
Reserve capacity prices over time
config : Configuration dictionary containing column names for prices.
"""
inp = self.input.copy()
sch = self.schedule.copy()
sch = sch.reindex(inp.index)
c_price_e = config["columns"]["energy"]
c_price_fcr = config["columns"]["fcr"]
c_price_up = config["columns"]["afrr_up"]
c_price_down = config["columns"]["afrr_down"]
c_r_fcr = "r_fcr_mw"
c_r_up = "r_afrr_up_mw"
c_r_down = "r_afrr_down_mw"
c_soc = "soc_mwh"
x = inp.index
if len(x) >= 2:
dt_seconds = float(np.median(np.diff(x.view("int64")) / 1e9))
else:
dt_seconds = 900.0
width_days = (dt_seconds / 86400.0) * 0.85
r_fcr = sch[c_r_fcr].to_numpy(dtype=float)
r_up = sch[c_r_up].to_numpy(dtype=float)
r_down = sch[c_r_down].to_numpy(dtype=float)
soc = sch[c_soc].to_numpy(dtype=float)
def nan_to_zero(a: np.ndarray) -> np.ndarray:
return np.nan_to_num(a, nan=0.0)
r_fcr = nan_to_zero(r_fcr)
r_up = nan_to_zero(r_up)
r_down = nan_to_zero(r_down)
fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(11, 9), sharex=True)
ax1.set_title(f"Reserve Bids and SoC — {self.date.date()} ({self.status}, {self.solver})")
x_num = mdates.date2num(x.to_pydatetime())
w = width_days
ax1.bar(x_num - w/3, r_fcr, width=w/3, label="FCR bid (MW)")
ax1.bar(x_num, r_down, width=w/3, label="aFRR DOWN bid (MW)")
ax1.bar(x_num + w/3, r_up, width=w/3, label="aFRR UP bid (MW)")
ax1.set_ylabel("Bid size (MW)")
ax1.grid(True, alpha=0.3)
ax1b = ax1.twinx()
ax1b.plot(x, soc, linewidth=2.0, label="SoC")
ax1b.set_ylabel("SoC (MWh)")
h1, l1 = ax1.get_legend_handles_labels()
h2, l2 = ax1b.get_legend_handles_labels()
ax1.legend(h1 + h2, l1 + l2, loc="upper right", frameon=True)
ax2.plot(x, inp[c_price_e].to_numpy(dtype=float), label="Spot / DA Price")
ax2.set_ylabel("Price (€/MWh)")
ax2.grid(True, alpha=0.3)
ax2.legend(loc="upper left", frameon=True)
ax2.set_title("Energy Prices Over Time")
ax3.plot(x, inp[c_price_fcr].to_numpy(dtype=float), label="FCR Price")
ax3.plot(x, inp[c_price_up].to_numpy(dtype=float), label="aFRR UP Price")
ax3.plot(x, inp[c_price_down].to_numpy(dtype=float), label="aFRR DOWN Price")
ax3.set_ylabel("Reserve price (€/MW/interval)")
ax3.grid(True, alpha=0.3)
ax3.legend(loc="upper right", frameon=True)
ax3.set_title("Reserve Capacity Prices Over Time")
ax3.xaxis.set_major_locator(mdates.HourLocator(interval=2))
ax3.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M"))
fig.autofmt_xdate(rotation=0)
fig.tight_layout()
fig.show()
def plot_global_results(solutions: List[DaySolution], config: Dict[str, Any]):
"""
Concatenate all daily solutions and plot the full horizon results.
solutions : List of daily solutions to concatenate and plot.
config : Configuration dictionary containing column names for prices.
"""
full_input = pd.concat([s.input for s in solutions])
full_schedule = pd.concat([s.schedule for s in solutions])
full_input.sort_index(inplace=True)
full_schedule.sort_index(inplace=True)
c_price_e = config["columns"]["energy"]
c_price_fcr = config["columns"]["fcr"]
c_price_up = config["columns"]["afrr_up"]
c_price_down = config["columns"]["afrr_down"]
x = full_input.index
dt_seconds = (x[1] - x[0]).total_seconds()
width_days = (dt_seconds / 86400.0) * 0.85
r_fcr = full_schedule["r_fcr_mw"].reindex(x).fillna(0.0).to_numpy()
r_up = full_schedule["r_afrr_up_mw"].reindex(x).fillna(0.0).to_numpy()
r_down = full_schedule["r_afrr_down_mw"].reindex(x).fillna(0.0).to_numpy()
soc = full_schedule["soc_mwh"].reindex(x).to_numpy()
x_num = mdates.date2num(x.to_pydatetime())
w = width_days
fig, (ax_fcr, ax_afrr, ax_energy, ax_prices) = plt.subplots(
4, 1, figsize=(12, 12), sharex=True
)
start_str = x[0].date()
end_str = x[-1].date()
fig.suptitle(f"Global Optimization Results: {start_str} to {end_str}", y=0.995)
ax_fcr.bar(x_num, r_fcr, width=w, label="FCR bid (MW)")
ax_fcr.set_ylabel("FCR (MW)")
ax_fcr.grid(True, alpha=0.3)
ax_fcr.legend(loc="upper left")
ax_fcr_soc = ax_fcr.twinx()
ax_fcr_soc.plot(x, soc, "k", linewidth=1.2, label="SoC")
ax_fcr_soc.set_ylabel("SoC (MWh)")
ax_fcr_soc.legend(loc="upper right")
ax_afrr.bar(x_num, r_down, color='tab:green', width=w, label="aFRR DOWN bid (MW)")
ax_afrr.bar(x_num, r_up, color='tab:orange', width=w, bottom=r_down, label="aFRR UP bid (MW)")
ax_afrr.set_ylabel("aFRR (MW)")
ax_afrr.grid(True, alpha=0.3)
ax_afrr.legend(loc="upper left")
ax_afrr_soc = ax_afrr.twinx()
ax_afrr_soc.plot(x, soc, "k", linewidth=1.2, label="SoC")
ax_afrr_soc.set_ylabel("SoC (MWh)")
ax_energy.plot(x, full_input[c_price_e].to_numpy(dtype=float), label="Spot / DA Price")
ax_energy.set_ylabel("€/MWh")
ax_energy.grid(True, alpha=0.3)
ax_energy.legend(loc="upper left")
ax_prices.plot(x, full_input[c_price_fcr].to_numpy(dtype=float), label="FCR Price")
ax_prices.plot(x, full_input[c_price_up].to_numpy(dtype=float), label="aFRR UP Price")
ax_prices.plot(x, full_input[c_price_down].to_numpy(dtype=float), label="aFRR DOWN Price")
ax_prices.set_ylabel("€/MW/interval")
ax_prices.grid(True, alpha=0.3)
ax_prices.legend(loc="upper left")
locator = mdates.AutoDateLocator()
formatter = mdates.ConciseDateFormatter(locator)
ax_prices.xaxis.set_major_locator(locator)
ax_prices.xaxis.set_major_formatter(formatter)
fig.tight_layout(rect=[0, 0, 1, 0.985])
plt.show() |