English version

Gold Layer & Bereitstellung nach außen

Diese Lektion behandelt die letzte Stufe: die Prognoseausgabe im Gold Layer strukturieren, Konfidenzintervalle berechnen, von der Turbine auf Parkebene aggregieren und die Vorhersagen an nachgelagerte Systeme liefern (Handelstische, Fahrplanmanagement, Netzbetreiber). Am Ende weißt du, wie du die Gold-Tabelle entwirfst und über welche Integrationsmuster du sie bereitstellst.

Was Gold liefern muss

Deine nachgelagerten Verbraucher interessieren sich nicht für Bronze, Silver oder Features. Sie brauchen genau eines: eine verlässliche Prognose mit einem klaren Vertrag.

Schema der Gold-Tabelle

CREATE TABLE prod.gold.forecast_ts (
  asset_id            STRING NOT NULL,
  asset_type          STRING NOT NULL,        -- "turbine" oder "wind_park"
  timestamp           TIMESTAMP NOT NULL,     -- Gültigkeitszeitpunkt der Prognose
  forecast_power_kw   DOUBLE NOT NULL,
  confidence_low_kw   DOUBLE,                 -- P10 (10. Perzentil)
  confidence_high_kw  DOUBLE,                 -- P90 (90. Perzentil)
  capacity_kw         DOUBLE,                 -- für die Berechnung des Kapazitätsfaktors
  capacity_factor     DOUBLE,                 -- forecast_power / capacity
  forecast_ref_time   TIMESTAMP NOT NULL,     -- welcher Wettermodell-Lauf
  model_version       STRING NOT NULL,        -- URI des MLflow-Modells
  produced_at         TIMESTAMP NOT NULL,     -- wann diese Zeile berechnet wurde
  source_model        STRING                  -- verwendetes Wettermodell
)
USING DELTA
PARTITIONED BY (asset_id)
TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true'
);

Aggregation Turbine → Park

Das ML-Modell prognostiziert je Turbine. Nachgelagerte Systeme brauchen oft Aggregate auf Parkebene. Berechne beides und lege es in derselben Tabelle ab:

from pyspark.sql import functions as F

def aggregate_to_park(turbine_forecasts, dim_assets):
    """
    Aggregate turbine-level forecasts to park level.
    Returns park-level rows to append alongside turbine rows in Gold.
    """
    # Join, um den übergeordneten Park zu ermitteln
    with_park = turbine_forecasts.join(
        dim_assets.select("asset_id", "parent_asset_id", "capacity_kw"),
        on="asset_id",
        how="inner"
    ).filter(F.col("parent_asset_id").isNotNull())  # nur Turbinen

    # Aggregation je Park × Zeitstempel
    park_forecast = with_park.groupBy(
        F.col("parent_asset_id").alias("asset_id"),
        "timestamp",
        "forecast_ref_time",
        "model_version",
        "produced_at",
        "source_model"
    ).agg(
        F.sum("forecast_power_kw").alias("forecast_power_kw"),
        F.sum("confidence_low_kw").alias("confidence_low_kw"),
        F.sum("confidence_high_kw").alias("confidence_high_kw"),
        F.sum("capacity_kw").alias("capacity_kw"),
    ).withColumn(
        "asset_type", F.lit("wind_park")
    ).withColumn(
        "capacity_factor", 
        F.col("forecast_power_kw") / F.col("capacity_kw")
    )

    return park_forecast

Warum werden Konfidenzintervalle summiert?

P10/P90 über Turbinen hinweg zu summieren ist eine Vereinfachung. Streng genommen sind Konfidenzintervalle auf Portfolioebene schmaler als die Summe der Einzelintervalle (Diversifikationseffekt). Für eine erste Iteration ist Summieren konservativ (die Unsicherheit wird überschätzt). Für eine Verfeinerung modellierst du die Unsicherheit direkt auf Parkebene.

Konfidenzintervalle

Punktprognosen reichen nicht. Handel und Fahrplanmanagement brauchen eine Quantifizierung der Unsicherheit. Dafür gibt es zwei Wege:

Variante A: Quantilsregression

# Separate Modelle für P10, P50 und P90 trainieren
# LightGBM unterstützt das von Haus aus:
model_p10 = lgb.LGBMRegressor(objective="quantile", alpha=0.10, **params)
model_p50 = lgb.LGBMRegressor(objective="quantile", alpha=0.50, **params)
model_p90 = lgb.LGBMRegressor(objective="quantile", alpha=0.90, **params)

Variante B: Prognoseintervalle aus Residuen

# Historische Prognosefehler zur Schätzung der Intervalle nutzen
residuals = actuals - predictions  # aus dem Validierungsset
p10_offset = np.percentile(residuals, 10)
p90_offset = np.percentile(residuals, 90)
# Bei der Inferenz: confidence_low = prediction + p10_offset
#                   confidence_high = prediction + p90_offset

Muster für die Bereitstellung

Wie greifen nachgelagerte Systeme auf die Gold-Daten zu?

Tabelle zur Prognosegüte

Gold sollte außerdem eine Qualitäts- bzw. Monitoring-Tabelle enthalten, die Prognosen mit Ist-Werten vergleicht:

-- Prognosegüte materialisieren, sobald Ist-Werte eintreffen
CREATE OR REPLACE TABLE prod.gold.forecast_accuracy AS
SELECT
  f.asset_id,
  f.timestamp,
  f.forecast_power_kw,
  p.power_kw AS actual_power_kw,
  f.forecast_power_kw - p.power_kw AS error_kw,
  ABS(f.forecast_power_kw - p.power_kw) AS abs_error_kw,
  f.model_version,
  f.forecast_ref_time,
  f.source_model,
  -- Lag der Ist-Wert innerhalb des Konfidenzbandes?
  CASE WHEN p.power_kw BETWEEN f.confidence_low_kw AND f.confidence_high_kw
       THEN TRUE ELSE FALSE END AS within_confidence
FROM prod.gold.forecast_ts f
JOIN prod.silver.power_ts p
  ON f.asset_id = p.asset_id AND f.timestamp = p.timestamp
WHERE f.asset_type = 'turbine'
  AND p.power_kw IS NOT NULL;

Prüfe dein Verständnis

Primärquelle

Wie geht es weiter

Die Pipeline steht durchgängig. Die nächste Lektion behandelt Workflows & Orchestrierung — wie du alle Stufen zu einem produktionsreifen Workflow mit Abhängigkeiten, Retries, Alerting und Zeitplan verdrahtest.

Fragen zum Aufsetzen von Delta Sharing, zu KPIs für die Prognosegüte oder zum Umgang mit negativen Prognosewerten (physikalisch unmöglich)? Frag deinen Lehrer.

Rechtliches: Impressum