You now know how to formulate, solve, bid, and handle uncertainty. This final lesson shows how all the pieces connect in a production system — the architecture you'll build at work.
┌──────────────────────────────────────────────────────────────────┐
│ BATTERY OPTIMIZER SYSTEM │
├──────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌─────────────────┐ │
│ │ DATA │ │ OPTIMIZER │ │ EXECUTION │ │
│ │ LAYER │────▶│ ENGINE │────▶│ LAYER │ │
│ └─────────────┘ └──────────────┘ └─────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ • Price forecasts • Day-ahead model • Bid submission │
│ • BMS telemetry • Intraday reopt • Order management│
│ • Market data feed • FCR co-optimization • Position tracking│
│ • FCR auction results • Scenario analysis • Schedule to BMS │
│ │
├──────────────────────────────────────────────────────────────────┤
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ MONITORING & LOGGING │ │
│ │ • P&L tracking • Forecast accuracy • Solver metrics │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘"""
data_layer.py — Collects and stores all inputs the optimizer needs
"""
from dataclasses import dataclass
from datetime import datetime
import pandas as pd
@dataclass
class BatteryState:
"""Real-time battery state from BMS"""
soc_mwh: float # Current state of charge
soc_pct: float # SoC as percentage
p_max_charge: float # Current max charge (may be derated)
p_max_discharge: float # Current max discharge
temperature_c: float # Cell temperature
soh_pct: float # State of health
timestamp: datetime
@dataclass
class MarketInputs:
"""All market data for one optimization run"""
da_price_forecast: pd.Series # 24h price forecast (€/MWh)
id_price_forecast: pd.Series # Intraday price forecast
fcr_prices: pd.Series # FCR capacity prices per block
existing_positions: pd.Series # Already committed MW per hour
gate_closure: datetime # When bids must be submitted
@dataclass
class OptimizerParams:
"""Battery parameters (from config, not real-time)"""
p_max: float # MW nameplate
e_max: float # MWh nameplate
eta: float # Round-trip efficiency
soc_min_pct: float # Minimum SoC (fraction)
soc_max_pct: float # Maximum SoC (fraction)
deg_cost: float # €/MWh degradation cost
ramp_max: float # MW/h ramp limit
"""
optimizer_engine.py — Core optimization logic
"""
import pyomo.environ as pyo
import numpy as np
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class OptimizationResult:
schedule: pd.DataFrame
revenue_eur: float
fcr_revenue_eur: float
degradation_cost_eur: float
solver_time_s: float
status: str
class BatteryOptimizer:
"""
Unified optimizer that handles:
- Day-ahead scheduling
- Intraday reoptimization
- FCR co-optimization
"""
def __init__(self, params: OptimizerParams):
self.params = params
self.solver = pyo.SolverFactory('highs')
def optimize_day_ahead(self,
price_forecast: List[float],
fcr_prices: List[float],
initial_soc: float) -> OptimizationResult:
"""Full day-ahead optimization with FCR co-optimization."""
p = self.params
T = len(price_forecast)
eta_sqrt = np.sqrt(p.eta)
model = pyo.ConcreteModel()
model.T = pyo.RangeSet(0, T - 1)
model.B = pyo.RangeSet(0, len(fcr_prices) - 1)
# Variables
model.p_c = pyo.Var(model.T, bounds=(0, p.p_max))
model.p_d = pyo.Var(model.T, bounds=(0, p.p_max))
model.soc = pyo.Var(model.T, bounds=(p.soc_min_pct * p.e_max,
p.soc_max_pct * p.e_max))
model.y = pyo.Var(model.T, within=pyo.Binary)
model.fcr = pyo.Var(model.B, bounds=(0, p.p_max))
# ... (constraints as in Lessons 3 + 6) ...
# [Full implementation follows same pattern]
result = self.solver.solve(model, tee=False)
return OptimizationResult(
schedule=self._extract_schedule(model, T, price_forecast),
revenue_eur=pyo.value(model.obj),
# ... etc
)
def reoptimize_intraday(self,
current_soc: float,
existing_positions: Dict[int, float],
price_forecast: Dict[int, float],
remaining_hours: List[int]) -> pd.DataFrame:
"""Intraday reoptimization, returns delta trades."""
# (Implementation from Lesson 5)
pass
"""
execution.py — Translates optimizer output into exchange orders
"""
from dataclasses import dataclass
from typing import List
from enum import Enum
class OrderDirection(Enum):
BUY = "BUY"
SELL = "SELL"
class OrderType(Enum):
LIMIT = "LIMIT"
MARKET = "MARKET"
BLOCK = "BLOCK"
@dataclass
class ExchangeOrder:
product_id: str # e.g., "DE_hourly_2026-07-18_17"
direction: OrderDirection
quantity_mw: float
price_eur_mwh: float
order_type: OrderType
class BidGenerator:
"""Converts optimizer schedule to exchange-ready orders."""
def __init__(self, strategy: str = 'price_taker'):
self.strategy = strategy
def generate_da_bids(self, schedule: pd.DataFrame,
delivery_date: str) -> List[ExchangeOrder]:
orders = []
for _, row in schedule.iterrows():
if row['charge_mw'] > 0.1:
orders.append(ExchangeOrder(
product_id=f"DE_hourly_{delivery_date}_{int(row['hour']):02d}",
direction=OrderDirection.BUY,
quantity_mw=round(row['charge_mw'], 1),
price_eur_mwh=3000.0 if self.strategy == 'price_taker'
else row['bid_price'],
order_type=OrderType.LIMIT,
))
elif row['discharge_mw'] > 0.1:
orders.append(ExchangeOrder(
product_id=f"DE_hourly_{delivery_date}_{int(row['hour']):02d}",
direction=OrderDirection.SELL,
quantity_mw=round(row['discharge_mw'], 1),
price_eur_mwh=-500.0 if self.strategy == 'price_taker'
else row['bid_price'],
order_type=OrderType.LIMIT,
))
return orders
"""
orchestrator.py — Daily scheduling workflow
"""
from datetime import datetime, timedelta
import schedule
import time
class DailyWorkflow:
def __init__(self, optimizer, data_layer, execution):
self.optimizer = optimizer
self.data = data_layer
self.execution = execution
def run_day_ahead(self):
"""Called at ~10:00 D-1, before gate closure at 12:00."""
# 1. Gather inputs
battery = self.data.get_battery_state()
market = self.data.get_market_inputs(horizon='day_ahead')
# 2. Optimize
result = self.optimizer.optimize_day_ahead(
price_forecast=market.da_price_forecast.tolist(),
fcr_prices=market.fcr_prices.tolist(),
initial_soc=battery.soc_mwh,
)
# 3. Generate and submit bids
bids = self.execution.generate_da_bids(
result.schedule,
delivery_date=(datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
)
self.execution.submit_bids(bids)
# 4. Log
print(f"[{datetime.now()}] DA bids submitted. Expected revenue: €{result.revenue_eur:.0f}")
def run_intraday_reopt(self):
"""Called every 15 min during delivery day."""
battery = self.data.get_battery_state()
positions = self.data.get_existing_positions()
forecast = self.data.get_intraday_forecast()
remaining = self.data.get_remaining_tradeable_hours()
if not remaining:
return # Nothing left to trade
deltas = self.optimizer.reoptimize_intraday(
current_soc=battery.soc_mwh,
existing_positions=positions,
price_forecast=forecast,
remaining_hours=remaining,
)
if not deltas.empty:
orders = self.execution.generate_intraday_orders(deltas)
self.execution.submit_orders(orders)
print(f"[{datetime.now()}] Intraday orders: {len(orders)} submitted")
# --- Scheduling ---
# workflow = DailyWorkflow(optimizer, data_layer, execution)
# schedule.every().day.at("10:00").do(workflow.run_day_ahead)
# schedule.every(15).minutes.do(workflow.run_intraday_reopt)
You must track performance to know if your optimizer is working:
# Key metric: capture ratio
# = realized_revenue / perfect_foresight_revenue
# Good: > 0.7. Excellent: > 0.85.
# If < 0.6, your forecast or optimizer has a problem.Why does the orchestrator separate the "optimizer engine" from the "execution layer"?
What is "capture ratio" and why is it the most important performance metric?
Why must the optimizer handle negative electricity prices?
Open Energy Modelling Initiative — community of researchers and practitioners building open-source energy system models. Join the forum for practical advice on production deployment.
Rechtliches: Impressum