Du weißt jetzt, wie man formuliert, löst, bietet und Unsicherheit handhabt. Diese letzte Lektion zeigt, wie alle Teile in einem Produktionssystem zusammenhängen — die Architektur, die du bei der Arbeit bauen wirst.
┌──────────────────────────────────────────────────────────────────┐
│ BATTERIE-OPTIMIERER-SYSTEM │
├──────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌─────────────────┐ │
│ │ DATEN- │ │ OPTIMIERER- │ │ AUSFÜHRUNGS- │ │
│ │ SCHICHT │────▶│ ENGINE │────▶│ SCHICHT │ │
│ └─────────────┘ └──────────────┘ └─────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ • Preisprognosen • Day-Ahead-Modell • Gebotseinreichung│
│ • BMS-Telemetrie • Intraday-Reopt. • Order-Management │
│ • Marktdaten-Feed • FCR-Co-Optimierung • Positionsverfolgung│
│ • FCR-Auktionsergeb. • Szenarioanalyse • Fahrplan → BMS │
│ │
├──────────────────────────────────────────────────────────────────┤
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ MONITORING & LOGGING │ │
│ │ • P&L-Tracking • Prognosegenauigkeit • Solver-Metriken │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘"""
daten_schicht.py — Sammelt und speichert alle Eingaben die der Optimierer braucht
"""
from dataclasses import dataclass
from datetime import datetime
import pandas as pd
@dataclass
class BatterieZustand:
"""Echtzeit-Batteriezustand vom BMS"""
soc_mwh: float # Aktueller Ladezustand
soc_pct: float # SoC in Prozent
p_max_laden: float # Aktuelle max. Ladeleistung (ggf. deratiert)
p_max_entladen: float # Aktuelle max. Entladeleistung
temperatur_c: float # Zelltemperatur
soh_pct: float # State of Health
zeitstempel: datetime
@dataclass
class MarktEingaben:
"""Alle Marktdaten für einen Optimierungslauf"""
da_preis_prognose: pd.Series # 24h Preisprognose (€/MWh)
id_preis_prognose: pd.Series # Intraday-Preisprognose
fcr_preise: pd.Series # FCR-Kapazitätspreise pro Block
bestehende_positionen: pd.Series # Bereits gebundene MW pro Stunde
gate_closure: datetime # Wann Gebote eingereicht sein müssen
@dataclass
class OptimierungsParameter:
"""Batterieparameter (aus Konfiguration, nicht Echtzeit)"""
p_max: float # MW Nennleistung
e_max: float # MWh Nennkapazität
eta: float # Gesamtwirkungsgrad
soc_min_pct: float # Minimaler SoC (Bruchteil)
soc_max_pct: float # Maximaler SoC (Bruchteil)
deg_kosten: float # €/MWh Degradationskosten
rampe_max: float # MW/h Rampengrenze
"""
optimierer_engine.py — Kernlogik der Optimierung
"""
import pyomo.environ as pyo
import numpy as np
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class OptimierungsErgebnis:
fahrplan: pd.DataFrame
erloes_eur: float
fcr_erloes_eur: float
degradationskosten_eur: float
solver_zeit_s: float
status: str
class BatterieOptimierer:
"""
Einheitlicher Optimierer der handhabt:
- Day-Ahead-Fahrplanung
- Intraday-Reoptimierung
- FCR-Co-Optimierung
"""
def __init__(self, parameter: OptimierungsParameter):
self.parameter = parameter
self.solver = pyo.SolverFactory('highs')
def optimiere_day_ahead(self,
preis_prognose: List[float],
fcr_preise: List[float],
anfangs_soc: float) -> OptimierungsErgebnis:
"""Volle Day-Ahead-Optimierung mit FCR-Co-Optimierung."""
p = self.parameter
T = len(preis_prognose)
eta_sqrt = np.sqrt(p.eta)
model = pyo.ConcreteModel()
model.T = pyo.RangeSet(0, T - 1)
model.B = pyo.RangeSet(0, len(fcr_preise) - 1)
# Variablen
model.p_l = pyo.Var(model.T, bounds=(0, p.p_max))
model.p_e = 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))
# ... (Nebenbedingungen wie in Lektionen 3 + 6) ...
ergebnis = self.solver.solve(model, tee=False)
return OptimierungsErgebnis(
fahrplan=self._fahrplan_extrahieren(model, T, preis_prognose),
erloes_eur=pyo.value(model.ziel),
# ... usw.
)
def reoptimiere_intraday(self,
aktueller_soc: float,
bestehende_positionen: Dict[int, float],
preis_prognose: Dict[int, float],
verbleibende_stunden: List[int]) -> pd.DataFrame:
"""Intraday-Reoptimierung, gibt Delta-Trades zurück."""
# (Implementierung aus Lektion 5)
pass
"""
ausfuehrung.py — Übersetzt Optimierer-Ausgabe in Börsenorders
"""
from dataclasses import dataclass
from typing import List
from enum import Enum
class OrderRichtung(Enum):
KAUF = "KAUF"
VERKAUF = "VERKAUF"
class OrderTyp(Enum):
LIMIT = "LIMIT"
MARKET = "MARKET"
BLOCK = "BLOCK"
@dataclass
class BoersenOrder:
produkt_id: str # z.B. "DE_stündlich_2026-07-18_17"
richtung: OrderRichtung
menge_mw: float
preis_eur_mwh: float
order_typ: OrderTyp
class GebotsGenerator:
"""Konvertiert Optimierer-Fahrplan zu börsenfertigen Orders."""
def __init__(self, strategie: str = 'preisnehmer'):
self.strategie = strategie
def erzeuge_da_gebote(self, fahrplan: pd.DataFrame,
liefertag: str) -> List[BoersenOrder]:
orders = []
for _, zeile in fahrplan.iterrows():
if zeile['laden_mw'] > 0.1:
orders.append(BoersenOrder(
produkt_id=f"DE_stündlich_{liefertag}_{int(zeile['stunde']):02d}",
richtung=OrderRichtung.KAUF,
menge_mw=round(zeile['laden_mw'], 1),
preis_eur_mwh=3000.0 if self.strategie == 'preisnehmer'
else zeile['gebotspreis'],
order_typ=OrderTyp.LIMIT,
))
elif zeile['entladen_mw'] > 0.1:
orders.append(BoersenOrder(
produkt_id=f"DE_stündlich_{liefertag}_{int(zeile['stunde']):02d}",
richtung=OrderRichtung.VERKAUF,
menge_mw=round(zeile['entladen_mw'], 1),
preis_eur_mwh=-500.0 if self.strategie == 'preisnehmer'
else zeile['gebotspreis'],
order_typ=OrderTyp.LIMIT,
))
return orders
"""
orchestrierer.py — Täglicher Ablauf-Workflow
"""
from datetime import datetime, timedelta
class TaeglichAblauf:
def __init__(self, optimierer, daten_schicht, ausfuehrung):
self.optimierer = optimierer
self.daten = daten_schicht
self.ausfuehrung = ausfuehrung
def day_ahead_ausfuehren(self):
"""Aufgerufen ~10:00 D-1, vor Gate Closure um 12:00."""
# 1. Eingaben sammeln
batterie = self.daten.hole_batterie_zustand()
markt = self.daten.hole_markt_eingaben(horizont='day_ahead')
# 2. Optimieren
ergebnis = self.optimierer.optimiere_day_ahead(
preis_prognose=markt.da_preis_prognose.tolist(),
fcr_preise=markt.fcr_preise.tolist(),
anfangs_soc=batterie.soc_mwh,
)
# 3. Gebote erzeugen und einreichen
gebote = self.ausfuehrung.erzeuge_da_gebote(
ergebnis.fahrplan,
liefertag=(datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
)
self.ausfuehrung.gebote_einreichen(gebote)
# 4. Loggen
print(f"[{datetime.now()}] DA-Gebote eingereicht. Erwarteter Erlös: €{ergebnis.erloes_eur:.0f}")
def intraday_reopt_ausfuehren(self):
"""Aufgerufen alle 15 Min. während des Liefertags."""
batterie = self.daten.hole_batterie_zustand()
positionen = self.daten.hole_bestehende_positionen()
prognose = self.daten.hole_intraday_prognose()
verbleibend = self.daten.hole_verbleibende_handelbare_stunden()
if not verbleibend:
return # Nichts mehr zu handeln
deltas = self.optimierer.reoptimiere_intraday(
aktueller_soc=batterie.soc_mwh,
bestehende_positionen=positionen,
preis_prognose=prognose,
verbleibende_stunden=verbleibend,
)
if not deltas.empty:
orders = self.ausfuehrung.erzeuge_intraday_orders(deltas)
self.ausfuehrung.orders_einreichen(orders)
print(f"[{datetime.now()}] Intraday-Orders: {len(orders)} eingereicht")
Du musst die Performance verfolgen um zu wissen, ob dein Optimierer funktioniert:
# Schlüsselmetrik: Capture Ratio (Abschöpfungsquote)
# = realisierter_erloes / perfekter_voraussicht_erloes
# Gut: > 0.7. Exzellent: > 0.85.
# Wenn < 0.6, hat deine Prognose oder dein Optimierer ein Problem.Warum trennt der Orchestrierer die „Optimierer-Engine" von der „Ausführungsschicht"?
Was ist die „Capture Ratio" und warum ist sie die wichtigste Performance-Metrik?
Warum muss der Optimierer negative Strompreise handhaben können?
Open Energy Modelling Initiative — Gemeinschaft von Forschern und Praktikern die Open-Source-Energiesystemmodelle bauen. Dem Forum beitreten für praktische Ratschläge zur Produktions-Deployment.
Rechtliches: Impressum