Skip to content

Contracts API

Contracts

Defines the "data contracts": the schemas defining the precise shape of each data source, the units each column carries, and the sign convention each column's values follow.

The contracts package also owns the thin configuration layer that sits beside those schemas: the cross-validation (CV) fold config and the class_target/import_class pair that turns a class into a _target_ string and back. A CV fold is one train-then-validate split of the history, and _target_ is the key a YAML config file uses to name the Python class to build. Both the fold config and the class-path pair are model-agnostic and need nothing heavier than pydantic and PyYAML.

Two further modules sit here because every package needs them and neither is specific to machine learning (ML). contracts.settings holds Settings, the single source of every setting the pipeline reads from its environment: the data paths, the object-store credentials, the MLflow tracking URI, and the four settings for Sentry, the error-reporting service, all resolved from the environment and the workspace .env and reached through the cached get_settings(). The Settings docstring names the four Sentry settings. contracts.uri holds the local-or-remote path helpers those settings fields need, because a data-location field may be a local path or an s3:// URI, and pathlib mangles a URI.

Light enough for any component to import

This package is designed to be lightweight. It defines the shape of the data using Patito and Polars, plus the settings and object-store path helpers those shapes are read and written through (deltalake and obstore), but it contains no ML-specific logic and no ML or orchestration dependency such as MLflow, XGBoost, or Dagster. That light dependency footprint is what lets any component in the system (e.g., a data ingestion script or a dashboard) import these schemas without bringing in the entire ML stack.

Key data contracts

The five schemas below are the ones most callers touch. The package defines four more — EffectiveCapacity, Metrics, EligibleTimeSeries, and H3GridWeights — for nine in all.

  • PowerTimeSeries: Half-hourly power observations in MW (megawatts) or MVA (megavolt-amperes) per time_series_id, as received from National Grid Electricity Distribution (NGED), the distribution network operator whose network this project forecasts.
  • TimeSeriesMetadata: Substation and customer meter metadata, including lat/lon, H3 index (the identifier of one cell of the H3 hexagonal grid the weather is aggregated onto), substation_type (Primary, BSP, GSP, EHV Customer, or HV Customer — the field's own description expands the voltage acronyms), and time_series_type (PV, Wind, BESS, Disaggregated Demand, and 18 others — LIST_OF_TIME_SERIES_TYPES holds all 22, expands BESS and PV, and names the seven types that appear in the V1 trial area).
  • Nwp: Numerical weather prediction (NWP) data from the European Centre for Medium-Range Weather Forecasts (ECMWF) ensemble (ENS), in physical units (Float32), on disk and in memory alike. An ensemble forecast runs the weather model many times over, and each run is one ensemble member, so the spread across the members is what expresses the forecast's uncertainty; ECMWF_ENS_ENSEMBLE_MEMBERS gives the count and the control/perturbed split. The on-disk copy is rounded to a 13-bit significand, which keeps a fixed number of significant figures rather than a fixed number of decimal places, and is laid out so that compression works well and a query can skip whole Parquet row groups — both by delta_store.nwp, which states the resulting error bound.
  • AllFeatures: The final joined dataset passed to ML models. Primary key is (time_series_id, power_fcst_init_time, valid_time[, ensemble_member]), where the square brackets mark ensemble_member as a key column only when the frame carries one row per ensemble member. Includes NWP weather variables, power lag/rolling features (the power observed a given number of hours earlier, and its rolling mean over a given number of hours), and datetime features. time_series_type is the one metadata column it can carry, and only when a feature set asks for it.
  • PowerForecast: ML model output schema. power_fcst is in MW (active power) or MVA (apparent power), with the unit given per time_series_id in TimeSeriesMetadata. The forecast is of the power available, meaning what the series would have exported or drawn had no active-network-management instruction been in force, so a consumer that wants the flow a capped generator will actually be allowed applies the cap itself. A planned change will normalise the forecast to [−1, +1] for NGED to multiply by that series' effective_capacity, in the same MW or MVA, to recover a power — see Forecast Building Blocks. Includes power_fcst_model_name, power_fcst_model_version, power_fcst_init_time, nwp_init_time, valid_time, time_series_id, and ensemble_member.

Sign convention

Sign convention depends on substation_type in TimeSeriesMetadata, whose five values (BSP, EHV Customer, GSP, HV Customer, Primary) partition into two behavioural cases:

  • Substations (BSP, GSP, Primary): positive = power flowing towards end-users; negative = excess generation flowing back upstream, into the electricity network above the substation.
  • Customer meters (EHV Customer, HV Customer): positive = the customer is sending power to NGED's distribution network; negative = the customer is drawing power from NGED's distribution network. A customer meter can sit at a demand site or a generation site, so this case is not "generators only".

The convention describes a direction, so it applies only where units is MW. A series metered in MVA reports the magnitude of the flow and cannot see direction, so reverse power flow appears as a rise rather than as a change of sign. A negative value is then a fault between the meter and us rather than an export. In the trial area, 10 sites are metered in apparent power — see apparent power (MVA) metering for the "bouncing off zero" behaviour apparent-power metering produces.

Design principles

  • The contract is the authoritative account of what the data means. It says what the data should be, not what some current code path happens to produce. So when code and contract disagree, the code is the first suspect. A null the contract forbids usually means an upstream join kept a row it should have dropped, or a caller passed input it should have rejected. Widening a field to | None, or relaxing a range, so that a failing validate() passes buries that defect in the one place the rest of the system trusts. Fix the code instead. Change the contract only when you can say what the data now means and why that meaning is right. Get the change agreed before making it, including a widening that looks like a formality — every reader of contracts is relying on it to still mean what it said yesterday.
  • Column naming: Column names are snake_case throughout, with any acronym or SI unit lower-cased inside the name (effective_capacity_mw, h3_res_5, area_wkt). Acronyms and SI units keep their standard capitalisation everywhere else — in a column's values (MW, MVA, BSP, PV), in class names, and in prose, where "DER" (distributed energy resource) and "MW" (megawatts) are both uppercase.
  • Semantic checks: Range validation should be generous — the aim is to catch physically impossible values (e.g., 1 GW from a 1 MW solar farm), not possible-but-unlikely values.
  • Datetime ranges: Timestamps on the columns where external data enters — PowerTimeSeries.time, Nwp.init_time and Nwp.valid_time — are bounded to [MIN_PLAUSIBLE_DATETIME, MAX_PLAUSIBLE_DATETIME] (2000-01-01 to 2100-01-01, inclusive), which rejects a corrupt feed or an epoch-unit mix-up without ever excluding a real reading. The check lives in each model's validate override via check_datetime_bounds, because Patito silently ignores ge and le — the greater-than-or-equal and less-than-or-equal bounds a field declares — on a datetime field. Patito derives its bounds checks from the JSON Schema minimum/maximum keywords, which JSON Schema (the validation standard, not NGED's JSON feed) defines for numbers only. Columns on our own output schemas (PowerForecast, EffectiveCapacity, AllFeatures, Metrics) have not opted in: they are computed from already-bounded inputs rather than received from outside.
  • Degrade, don't abort, at an ingestion boundary: validate() stays strict everywhere — it is also used as a hard assertion in tests and R&D code, where a raise-on-violation contract must not silently change. But a single malformed row from an external feed should not abort ingestion of every other well-formed row in the batch, so PowerTimeSeries.drop_implausible_rows() filters out rows with an out-of-range or minute-misaligned time before validate() runs, returning the survivors plus a count of what was dropped. Only the NGED JSON ingestion path (nged_data.read_nged_json) calls it; the duplicate/sortedness checks in validate() are never relaxed, because those indicate a bug in our own pipeline rather than malformed external data.
  • Repair a known upstream fault at the ingestion boundary: NGED's feed stamped every reading half an hour late until 08:30 UTC on 26 March 2026 (POWER_TIMESTAMPS_CORRECTED_BEFORE), so PowerTimeSeries.correct_late_timestamps() moves every reading stamped before that instant back by 30 minutes. The correct_late_timestamps docstring says why the repair has to run before drop_implausible_rows().
  • No lookahead bias: AllFeatures carries power_fcst_init_time (when we make the forecast) as a distinct field from nwp_init_time (when the NWP model ran). Power lag features are nullified by _nullify_leaky_lags() when the lag is shorter than or equal to the forecast lead time — the lead time being the gap between power_fcst_init_time and valid_time. A short lag is the dangerous case: a reading taken fewer hours before the target time than the lead time had not yet happened when the forecast was made, so using it would be reading the future.

contracts.common

Shared building blocks for the Patito schemas.

This module holds the canonical UTC dtype, the plausible-datetime bounds and the checks that enforce them, the quantile levels agreed with NGED for the delivery tables and their p{level} labels, the shared time_series_id field factory, and validate_schema. validate_schema checks a frame's columns and dtypes against a model without materialising it, that is, without reading any of the frame's rows.

Attributes

UTC_DATETIME_DTYPE = pl.Datetime(time_unit='us', time_zone='UTC') module-attribute

MIN_PLAUSIBLE_DATETIME = datetime(2000, 1, 1, tzinfo=UTC) module-attribute

The earliest timestamp a bounded datetime column may carry (inclusive).

A column is bounded when its model's validate passes it to check_datetime_bounds; the constant says nothing about columns that have not opted in.

NGED telemetry cannot predate the instrumentation that produced the telemetry, and the ECMWF archive we ingest begins later still. No legitimate row is therefore older than 2000-01-01. The bound is also deliberately far later than 1847: Europe/London ran on local mean time at UTC−0:01:15 until then, so a pre-1848 timestamp produces a sub-minute UTC offset and a nonsensical value for every local-time feature. Enforcing this bound is what guarantees the local-time features in ml_core never see a sub-minute UTC offset.

MAX_PLAUSIBLE_DATETIME = datetime(2100, 1, 1, tzinfo=UTC) module-attribute

The latest timestamp a bounded datetime column may carry (inclusive).

The maximum is a fixed date rather than an offset from the current time, so validation never depends on the wall clock. A frame that validated when it was written still validates when it is read back years later, and tests need no clock control. A fixed far-future bound still catches an epoch-unit mix-up, where Unix milliseconds read as seconds land tens of thousands of years in the future — a plausible failure on any path that converts a numeric timestamp rather than parsing an ISO-8601 string. The bound deliberately does not catch a small clock skew that ships tomorrow's data as today's. Clock skew is a monitoring concern, not a contract concern.

DELIVERY_QUANTILES = (0.01, 0.02, 0.05, 0.1, 0.2, 0.35, 0.5, 0.65, 0.8, 0.9, 0.95, 0.98, 0.99) module-attribute

The 13 quantile levels agreed with NGED for the delivery tables.

The levels are deliberately tail-heavy, because NGED is far more interested in the tails than the shoulders. This tuple is the single source of truth for every quantile-indexed artefact — the pinball-loss metric_param labels today, and the percentile columns of the delivery-table representations (Representations 2 and 3) when those representations land in v0.5.

Functions:

check_datetime_bounds(dataframe, column, *more_columns)

Raise ValueError if any timestamp lies outside the plausible-datetime range.

Call this from a Patito model's validate override, after super().validate(). It exists because Patito silently ignores ge/le on a datetime field. Patito derives its bounds checks from the Pydantic JSON schema's minimum/maximum keywords, and JSON Schema defines those two keywords for numbers only. The JSON schema therefore never receives the Ge/Le annotations Pydantic builds from a datetime field's ge/le arguments, and no check is ever generated. (ge/le on a numeric field works normally, which is why PowerTimeSeries.power can state its bounds on the field itself.)

Parameters:

Name Type Description Default
dataframe DataFrame

An already-validated frame. Every named column must be a datetime column.

 required
column str

Name of a datetime column to bound.

 required
*more_columns str

Names of any further datetime columns to bound.

 ()

Raises:

Type Description
ValueError

If any value is before MIN_PLAUSIBLE_DATETIME or after MAX_PLAUSIBLE_DATETIME. Nulls are ignored — absence is not malformedness — and an empty frame always passes.
Source code in packages/contracts/src/contracts/common.py 
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
def check_datetime_bounds(dataframe: pl.DataFrame, column: str, *more_columns: str) -> None:
    """Raise ``ValueError`` if any timestamp lies outside the plausible-datetime range.

    Call this from a Patito model's ``validate`` override, after ``super().validate()``. It exists
    because Patito **silently ignores** ``ge``/``le`` on a datetime field. Patito derives its bounds
    checks from the Pydantic JSON schema's ``minimum``/``maximum`` keywords, and JSON Schema defines
    those two keywords for numbers only. The JSON schema therefore never receives the ``Ge``/``Le``
    annotations Pydantic builds from a datetime field's ``ge``/``le`` arguments, and no check is
    ever generated. (``ge``/``le`` on a *numeric* field works normally, which is why
    ``PowerTimeSeries.power`` can state its bounds on the field itself.)

    Args:
        dataframe: An already-validated frame. Every named column must be a datetime column.
        column: Name of a datetime column to bound.
        *more_columns: Names of any further datetime columns to bound.

    Raises:
        ValueError: If any value is before `MIN_PLAUSIBLE_DATETIME` or after
            `MAX_PLAUSIBLE_DATETIME`. Nulls are ignored — absence is not malformedness — and
            an empty frame always passes.
    """
    columns = (column, *more_columns)
    extremes = dataframe.select(
        *(pl.col(name).min().alias(f"min_{name}") for name in columns),
        *(pl.col(name).max().alias(f"max_{name}") for name in columns),
    ).row(0, named=True)
    for name in columns:
        _raise_if_outside_plausible_range(name, extremes[f"min_{name}"], extremes[f"max_{name}"])

split_by_datetime_plausibility(dataframe, column)

Partition dataframe into (plausible, implausible) rows by column.

A row is implausible when its column value is before MIN_PLAUSIBLE_DATETIME or after MAX_PLAUSIBLE_DATETIME — the same bounds check_datetime_bounds enforces. Nulls are always plausible (absence is not malformedness).

Use this at an ingestion boundary, to drop-and-report malformed external rows rather than abort the whole batch. Where a hard assertion is wanted instead — inside a Patito model's validate, for example — use check_datetime_bounds.

Parameters:

Name Type Description Default
dataframe DataFrame

Any frame; column must be a datetime column.

 required
column str

Name of the datetime column to test.

 required

Returns:

Type Description
tuple[DataFrame, DataFrame]

(plausible, implausible), each keeping dataframe's row order and schema.
Source code in packages/contracts/src/contracts/common.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def split_by_datetime_plausibility(
    dataframe: pl.DataFrame, column: str
) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Partition ``dataframe`` into ``(plausible, implausible)`` rows by ``column``.

    A row is implausible when its ``column`` value is before `MIN_PLAUSIBLE_DATETIME` or
    after `MAX_PLAUSIBLE_DATETIME` — the same bounds `check_datetime_bounds` enforces.
    Nulls are always plausible (absence is not malformedness).

    Use this at an ingestion boundary, to drop-and-report malformed external rows rather than
    abort the whole batch. Where a hard assertion is wanted instead — inside a Patito model's
    ``validate``, for example — use `check_datetime_bounds`.

    Args:
        dataframe: Any frame; ``column`` must be a datetime column.
        column: Name of the datetime column to test.

    Returns:
        ``(plausible, implausible)``, each keeping ``dataframe``'s row order and schema.
    """
    is_implausible = pl.col(column).is_not_null() & (
        (pl.col(column) < MIN_PLAUSIBLE_DATETIME) | (pl.col(column) > MAX_PLAUSIBLE_DATETIME)
    )
    return dataframe.filter(~is_implausible), dataframe.filter(is_implausible)

quantile_label(quantile)

Return the canonical p{level} label for a quantile, e.g. 0.05 → "p5".

The label format matches the percentile column names agreed with NGED for the delivery tables (see DELIVERY_QUANTILES).

Source code in packages/contracts/src/contracts/common.py 
155
156
157
158
159
160
161
def quantile_label(quantile: float) -> str:
    """Return the canonical ``p{level}`` label for a quantile, e.g. ``0.05`` → ``"p5"``.

    The label format matches the percentile column names agreed with NGED for the delivery tables
    (see ``DELIVERY_QUANTILES``).
    """
    return f"p{round(quantile * 100)}"

validate_schema(model, df)

Validate a Polars DataFrame's or LazyFrame's schema against a Patito model.

Raises DataFrameValidationError on failure. On LazyFrames this materializes no data; it just calls collect_schema().

Source code in packages/contracts/src/contracts/common.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def validate_schema(model: type[pt.Model], df: pl.DataFrame | pl.LazyFrame) -> None:
    """Validate a Polars DataFrame's or LazyFrame's schema against a Patito model.

    Raises `DataFrameValidationError` on failure. On LazyFrames this materializes no data; it
    just calls `collect_schema()`.
    """
    # Get actual schema
    actual_schema = dict(df.collect_schema()) if isinstance(df, pl.LazyFrame) else dict(df.schema)

    # Check for missing columns
    missing_cols = set(model.dtypes.keys()) - set(actual_schema.keys())
    if missing_cols:
        error = ErrorWrapper(
            MissingColumnsError(f"Missing columns: {missing_cols}"), loc=tuple(missing_cols)
        )
        raise DataFrameValidationError([error], model)

    # Check for dtype mismatches
    errors = []
    for col, expected_dtype in model.dtypes.items():
        if col in actual_schema:
            actual_dtype = actual_schema[col]
            if actual_dtype != expected_dtype:
                errors.append(
                    ErrorWrapper(
                        ColumnDTypeError(
                            f"Column '{col}' expected {expected_dtype}, got {actual_dtype}"
                        ),
                        loc=(col,),
                    )
                )

    # Raise the native Patito exception if errors were found
    if errors:
        raise DataFrameValidationError(errors, model)

contracts.power_schemas

Contracts for NGED's power telemetry and for the forecasts we make from it.

The half-hourly PowerTimeSeries observations as they arrive from NGED, the TimeSeriesMetadata roster describing each series, the PowerForecast schema every model emits, and the EffectiveCapacity estimate the metrics pipeline divides the mean absolute error by, to express that error as a fraction of the series' capacity.

Attributes

POWER_TIMESTAMPS_CORRECTED_BEFORE = datetime(2026, 3, 26, 8, 30, tzinfo=UTC) module-attribute

NGED's power timestamps are half an hour late before this instant, and correct from this instant onwards.

A reading whose timestamp T falls before this instant is the mean over (T - 60 min, T - 30 min], not the (T - 30 min, T] the time field states. This instant is NGED's own account of where the fault stops, not a changepoint fitted from the readings. NGED's archive still carries the late timestamps: the earlier readings have not been republished. Three independent measurements are consistent with NGED's account — two against the sun's own position on a clear day, the third against satellite irradiance: https://openclimatefix.github.io/nged-substation-forecast/studies/beam-diffuse-split/#the-power-timestamps-before-26-march-2026-are-half-an-hour-late.

The repair applies to every time_series_id. NGED report that they convert every series in the trial area through one code path, so no series can have escaped the fault. That report is what the fleet-wide scope rests on: the published measurements cover the six metered solar farms only, because each measurement needs a series whose output follows the sun. A substation load profile has no equivalent.

LIST_OF_TIME_SERIES_TYPES = ('BESS', 'Biofuel', 'CHP', 'Data Centre', 'Disaggregated Demand', 'Energy from Waste', 'EV Charging', 'Geothermal', 'Hydro', 'Hydrogen Electrolysis', 'Industrial Demand', 'Mixed (Demand)', 'Mixed (Generation)', 'Other (Demand)', 'Other (Generation)', 'Other (Storage)', 'Peaking Plant', 'PV', 'Rail', 'Raw Flow', 'Synchronous Condenser', 'Wind') module-attribute

All time-series type values used in NGED data.

Types present in the V1 trial area: BESS, Biofuel, Disaggregated Demand, Other (Generation), PV, Raw Flow, Wind.

Notes:

  • BESS: Battery energy storage system.
  • Disaggregated Demand: In the trial area, exclusively associated with "Primary" substations. All "Primary" substations in the trial area have their TimeSeriesType set to "Disaggregated Demand". Indicates that NGED have already removed metered generation connected to that primary.
  • PV: Photovoltaic (solar).
  • Raw Flow: Used for bulk supply point (BSP) and grid supply point (GSP) substations.

FoldId = str module-attribute

Fold identifier for PowerForecast.fold_id.

A CV fold's id is a short label defined in conf/cv/default.yaml (e.g. "mid_2025_to_mid_2026"); fold identity is config-driven, never hard-coded here. "live" is the reserved sentinel for a production forecast that belongs to no CV fold.

Classes

DropImplausibleRowsResult

Bases: NamedTuple

Result of PowerTimeSeries.drop_implausible_rows.

Source code in packages/contracts/src/contracts/power_schemas.py 
26
27
28
29
30
class DropImplausibleRowsResult(NamedTuple):
    """Result of ``PowerTimeSeries.drop_implausible_rows``."""

    survivors: pl.DataFrame
    n_dropped: int
Attributes
survivors instance-attribute
n_dropped instance-attribute

PowerTimeSeries

Bases: Model

Half-hourly power observations (MW or MVA), one row per (time_series_id, time).

Source code in packages/contracts/src/contracts/power_schemas.py 
 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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class PowerTimeSeries(pt.Model):
    """Half-hourly power observations (MW or MVA), one row per (time_series_id, time)."""

    time_series_id: int = _get_time_series_id_dtype()

    time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "End time of the 30-minute observation period (all NGED data is already half-hourly)."
            " A value before `POWER_TIMESTAMPS_CORRECTED_BEFORE` is NGED's own timestamp moved 30"
            " minutes earlier by `correct_late_timestamps`. NGED's feed stamped every reading half"
            " an hour late until that instant, and has stamped every reading correctly since."
            f" Must fall between {MIN_PLAUSIBLE_DATETIME:%Y-%m-%d} and"
            f" {MAX_PLAUSIBLE_DATETIME:%Y-%m-%d} (enforced by `validate`, not by the field, because"
            " Patito ignores `ge`/`le` on datetime fields — see `check_datetime_bounds`)."
        ),
    )

    power: float = pt.Field(
        dtype=pl.Float32,
        ge=-1000,
        le=1000,
        description=(
            "Average power (MW or MVA) over the preceding 30-minute period. Unit defined in "
            "TimeSeriesMetadata."
            " This is what the meter recorded, so an hour in which the network operator capped"
            " a generator's export reads as the capped value rather than as the power that was"
            " available. `PowerForecast.power_fcst` is defined the other way round."
            " Sign convention depends on `substation_type` in `TimeSeriesMetadata`, and describes"
            " a direction, so it applies only where `units` is `MW`. A series metered in `MVA`"
            " reports the magnitude of the flow and cannot see direction, so reverse power flow"
            " appears as a rise rather than as a change of sign. A negative value is then a fault"
            " between the meter and us rather than an export. See the Sign convention section in"
            " this package's README.md, also published at"
            " https://openclimatefix.github.io/nged-substation-forecast/roadmap/forecast-building-blocks/#sign-convention."
        ),
    )

    @classmethod
    def validate(  # ty: ignore[invalid-method-override]
        cls,
        dataframe: pl.DataFrame,
        columns: Sequence[str] | None = None,
        allow_missing_columns: bool = False,
        allow_superfluous_columns: bool = False,
        drop_superfluous_columns: bool = False,
    ) -> pt.DataFrame[Self]:
        """Validate the given dataframe, ensuring time is plausible, at :00 or :30, and unique."""
        validated_df = super().validate(
            dataframe=dataframe,
            columns=columns,
            allow_missing_columns=allow_missing_columns,
            allow_superfluous_columns=allow_superfluous_columns,
            drop_superfluous_columns=drop_superfluous_columns,
        )

        # Validate time falls in the plausible range (Patito ignores `ge`/`le` on datetime fields)
        check_datetime_bounds(validated_df, "time")

        # Validate time is at :00 or :30
        minutes = validated_df["time"].dt.minute()
        if not minutes.is_in([0, 30]).all():
            raise ValueError("time must be at the top or bottom of the hour (minute 00 or 30).")

        # Validate uniqueness of (time_series_id, time)
        if validated_df.select(["time_series_id", "time"]).is_duplicated().any():
            raise ValueError("Duplicate entries found for (time_series_id, time).")

        # Validate the time_series_id column is sorted
        if not validated_df["time_series_id"].is_sorted():
            raise ValueError("time_series_id is not sorted!")

        # Validate the time column is sorted (within each time_series_id group)
        if (
            not validated_df.group_by("time_series_id")
            .agg(pl.col("time").diff().min() > 0)["time"]
            .all()
        ):
            raise ValueError("the `time` column is not sorted!")

        return validated_df

    @classmethod
    def correct_late_timestamps(cls, dataframe: pl.DataFrame) -> pl.DataFrame:
        """Move every `time` before ``POWER_TIMESTAMPS_CORRECTED_BEFORE`` back by 30 minutes.

        NGED stamped this feed half an hour late until ``POWER_TIMESTAMPS_CORRECTED_BEFORE``, and
        have stamped the feed correctly since. The constant's docstring holds the evidence and the
        scope. Repairing at ingestion is what lets the `time` field mean the same half-hour for
        every row, so no consumer has to know which side of that instant a row was stamped on.

        Call ``correct_late_timestamps`` BEFORE ``drop_implausible_rows``, and only at a boundary
        that receives NGED's raw JSON. ``drop_implausible_rows`` has to judge the timestamp that
        will actually be stored. In the reverse order, a reading whose repaired timestamp falls
        outside the plausible range survives the drop and then raises out of ``validate``. One
        malformed external reading would then fail the whole ingest run.

        The repair cannot collide with an existing row or disturb the sort order. The repair shifts
        a contiguous prefix of each series by a constant. Every shifted timestamp lands more than
        30 minutes before the first unshifted timestamp. A repaired series has no reading at
        ``POWER_TIMESTAMPS_CORRECTED_BEFORE - 30 min``, because NGED never published that
        half-hour.

        Args:
            dataframe: A frame with a `time` column already cast to ``UTC_DATETIME_DTYPE``; need
                not yet be validated.

        Returns:
            `dataframe` with `time` corrected, and every other column untouched.
        """
        return dataframe.with_columns(
            time=pl.when(pl.col("time") < POWER_TIMESTAMPS_CORRECTED_BEFORE)
            .then(pl.col("time").dt.offset_by("-30m"))
            .otherwise(pl.col("time"))
        )

    @classmethod
    def drop_implausible_rows(cls, dataframe: pl.DataFrame) -> DropImplausibleRowsResult:
        """Drop rows with a malformed ``time``, returning ``(survivors, n_dropped)``.

        A row is dropped when its ``time`` lies outside the plausible datetime range, is null, or
        does not fall on the top or bottom of the hour (minute 00 or 30). The schema declares
        ``time`` non-nullable, so a null this early is already malformed. All three conditions
        indicate a malformed upstream reading — not a bug in our own pipeline — so under [inherent
        stability](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/)
        an ingestion boundary should degrade the batch rather than abort it entirely.

        Dropping rows exists alongside ``validate``, which stays strict. ``validate`` raises on
        two of the three conditions above: a ``time`` outside the plausible range, and a ``time``
        that is not on the top or bottom of the hour. The third condition, a null ``time``, is
        rejected earlier still, by the non-nullable field. ``validate`` is also used as a hard
        assertion in tests and R&D code, where a raise-on-violation contract must not silently
        change.

        Call this method BEFORE ``validate``, and only at a boundary that receives data from
        outside our system (e.g. NGED's raw JSON feed). The uniqueness and sortedness checks in
        ``validate`` are NOT relaxed here, and should keep raising. A duplicate row or an unsorted
        column indicates a bug in OUR pipeline rather than malformed external data.

        Args:
            dataframe: An already-cast frame with a ``time`` column; need not yet be validated.

        Returns:
            ``(survivors, n_dropped)``. ``survivors`` keeps ``dataframe``'s row order.
        """
        survivors, _ = split_by_datetime_plausibility(dataframe, "time")
        # `dt.minute()` is null for a null `time` and `.filter()` drops a row on a null predicate,
        # so this also drops the null `time`s the non-nullable schema forbids.
        survivors = survivors.filter(pl.col("time").dt.minute().is_in([0, 30]))
        # The dropped-row count is computed as a difference in `height` (a frame's row count),
        # rather than by summing the rejected partitions. Whatever a filter does with a null
        # predicate, a height difference counts every row that left exactly once.
        return DropImplausibleRowsResult(survivors, dataframe.height - survivors.height)

    # Define columns_to_sort_by as a ClassVar so Patito/Pydantic knows it is not a data field
    columns_to_sort_by: ClassVar[tuple[str, str]] = ("time_series_id", "time")
Attributes
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
time = pt.Field(dtype=UTC_DATETIME_DTYPE, description=f"End time of the 30-minute observation period (all NGED data is already half-hourly). A value before `POWER_TIMESTAMPS_CORRECTED_BEFORE` is NGED's own timestamp moved 30 minutes earlier by `correct_late_timestamps`. NGED's feed stamped every reading half an hour late until that instant, and has stamped every reading correctly since. Must fall between {MIN_PLAUSIBLE_DATETIME:%Y-%m-%d} and {MAX_PLAUSIBLE_DATETIME:%Y-%m-%d} (enforced by `validate`, not by the field, because Patito ignores `ge`/`le` on datetime fields — see `check_datetime_bounds`).") class-attribute instance-attribute
power = pt.Field(dtype=pl.Float32, ge=-1000, le=1000, description="Average power (MW or MVA) over the preceding 30-minute period. Unit defined in TimeSeriesMetadata. This is what the meter recorded, so an hour in which the network operator capped a generator's export reads as the capped value rather than as the power that was available. `PowerForecast.power_fcst` is defined the other way round. Sign convention depends on `substation_type` in `TimeSeriesMetadata`, and describes a direction, so it applies only where `units` is `MW`. A series metered in `MVA` reports the magnitude of the flow and cannot see direction, so reverse power flow appears as a rise rather than as a change of sign. A negative value is then a fault between the meter and us rather than an export. See the Sign convention section in this package's README.md, also published at https://openclimatefix.github.io/nged-substation-forecast/roadmap/forecast-building-blocks/#sign-convention.") class-attribute instance-attribute
columns_to_sort_by = ('time_series_id', 'time') class-attribute
Methods:
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the given dataframe, ensuring time is plausible, at :00 or :30, and unique.

Source code in packages/contracts/src/contracts/power_schemas.py 
 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
@classmethod
def validate(  # ty: ignore[invalid-method-override]
    cls,
    dataframe: pl.DataFrame,
    columns: Sequence[str] | None = None,
    allow_missing_columns: bool = False,
    allow_superfluous_columns: bool = False,
    drop_superfluous_columns: bool = False,
) -> pt.DataFrame[Self]:
    """Validate the given dataframe, ensuring time is plausible, at :00 or :30, and unique."""
    validated_df = super().validate(
        dataframe=dataframe,
        columns=columns,
        allow_missing_columns=allow_missing_columns,
        allow_superfluous_columns=allow_superfluous_columns,
        drop_superfluous_columns=drop_superfluous_columns,
    )

    # Validate time falls in the plausible range (Patito ignores `ge`/`le` on datetime fields)
    check_datetime_bounds(validated_df, "time")

    # Validate time is at :00 or :30
    minutes = validated_df["time"].dt.minute()
    if not minutes.is_in([0, 30]).all():
        raise ValueError("time must be at the top or bottom of the hour (minute 00 or 30).")

    # Validate uniqueness of (time_series_id, time)
    if validated_df.select(["time_series_id", "time"]).is_duplicated().any():
        raise ValueError("Duplicate entries found for (time_series_id, time).")

    # Validate the time_series_id column is sorted
    if not validated_df["time_series_id"].is_sorted():
        raise ValueError("time_series_id is not sorted!")

    # Validate the time column is sorted (within each time_series_id group)
    if (
        not validated_df.group_by("time_series_id")
        .agg(pl.col("time").diff().min() > 0)["time"]
        .all()
    ):
        raise ValueError("the `time` column is not sorted!")

    return validated_df
correct_late_timestamps(dataframe) classmethod

Move every time before POWER_TIMESTAMPS_CORRECTED_BEFORE back by 30 minutes.

NGED stamped this feed half an hour late until POWER_TIMESTAMPS_CORRECTED_BEFORE, and have stamped the feed correctly since. The constant's docstring holds the evidence and the scope. Repairing at ingestion is what lets the time field mean the same half-hour for every row, so no consumer has to know which side of that instant a row was stamped on.

Call correct_late_timestamps BEFORE drop_implausible_rows, and only at a boundary that receives NGED's raw JSON. drop_implausible_rows has to judge the timestamp that will actually be stored. In the reverse order, a reading whose repaired timestamp falls outside the plausible range survives the drop and then raises out of validate. One malformed external reading would then fail the whole ingest run.

The repair cannot collide with an existing row or disturb the sort order. The repair shifts a contiguous prefix of each series by a constant. Every shifted timestamp lands more than 30 minutes before the first unshifted timestamp. A repaired series has no reading at POWER_TIMESTAMPS_CORRECTED_BEFORE - 30 min, because NGED never published that half-hour.

Parameters:

Name Type Description Default
dataframe DataFrame

A frame with a time column already cast to UTC_DATETIME_DTYPE; need not yet be validated.

 required

Returns:

Type Description
DataFrame

dataframe with time corrected, and every other column untouched.
Source code in packages/contracts/src/contracts/power_schemas.py 
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
@classmethod
def correct_late_timestamps(cls, dataframe: pl.DataFrame) -> pl.DataFrame:
    """Move every `time` before ``POWER_TIMESTAMPS_CORRECTED_BEFORE`` back by 30 minutes.

    NGED stamped this feed half an hour late until ``POWER_TIMESTAMPS_CORRECTED_BEFORE``, and
    have stamped the feed correctly since. The constant's docstring holds the evidence and the
    scope. Repairing at ingestion is what lets the `time` field mean the same half-hour for
    every row, so no consumer has to know which side of that instant a row was stamped on.

    Call ``correct_late_timestamps`` BEFORE ``drop_implausible_rows``, and only at a boundary
    that receives NGED's raw JSON. ``drop_implausible_rows`` has to judge the timestamp that
    will actually be stored. In the reverse order, a reading whose repaired timestamp falls
    outside the plausible range survives the drop and then raises out of ``validate``. One
    malformed external reading would then fail the whole ingest run.

    The repair cannot collide with an existing row or disturb the sort order. The repair shifts
    a contiguous prefix of each series by a constant. Every shifted timestamp lands more than
    30 minutes before the first unshifted timestamp. A repaired series has no reading at
    ``POWER_TIMESTAMPS_CORRECTED_BEFORE - 30 min``, because NGED never published that
    half-hour.

    Args:
        dataframe: A frame with a `time` column already cast to ``UTC_DATETIME_DTYPE``; need
            not yet be validated.

    Returns:
        `dataframe` with `time` corrected, and every other column untouched.
    """
    return dataframe.with_columns(
        time=pl.when(pl.col("time") < POWER_TIMESTAMPS_CORRECTED_BEFORE)
        .then(pl.col("time").dt.offset_by("-30m"))
        .otherwise(pl.col("time"))
    )
drop_implausible_rows(dataframe) classmethod

Drop rows with a malformed time, returning (survivors, n_dropped).

A row is dropped when its time lies outside the plausible datetime range, is null, or does not fall on the top or bottom of the hour (minute 00 or 30). The schema declares time non-nullable, so a null this early is already malformed. All three conditions indicate a malformed upstream reading — not a bug in our own pipeline — so under inherent stability an ingestion boundary should degrade the batch rather than abort it entirely.

Dropping rows exists alongside validate, which stays strict. validate raises on two of the three conditions above: a time outside the plausible range, and a time that is not on the top or bottom of the hour. The third condition, a null time, is rejected earlier still, by the non-nullable field. validate is also used as a hard assertion in tests and R&D code, where a raise-on-violation contract must not silently change.

Call this method BEFORE validate, and only at a boundary that receives data from outside our system (e.g. NGED's raw JSON feed). The uniqueness and sortedness checks in validate are NOT relaxed here, and should keep raising. A duplicate row or an unsorted column indicates a bug in OUR pipeline rather than malformed external data.

Parameters:

Name Type Description Default
dataframe DataFrame

An already-cast frame with a time column; need not yet be validated.

 required

Returns:

Type Description
DropImplausibleRowsResult

(survivors, n_dropped). survivors keeps dataframe's row order.
Source code in packages/contracts/src/contracts/power_schemas.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
@classmethod
def drop_implausible_rows(cls, dataframe: pl.DataFrame) -> DropImplausibleRowsResult:
    """Drop rows with a malformed ``time``, returning ``(survivors, n_dropped)``.

    A row is dropped when its ``time`` lies outside the plausible datetime range, is null, or
    does not fall on the top or bottom of the hour (minute 00 or 30). The schema declares
    ``time`` non-nullable, so a null this early is already malformed. All three conditions
    indicate a malformed upstream reading — not a bug in our own pipeline — so under [inherent
    stability](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/)
    an ingestion boundary should degrade the batch rather than abort it entirely.

    Dropping rows exists alongside ``validate``, which stays strict. ``validate`` raises on
    two of the three conditions above: a ``time`` outside the plausible range, and a ``time``
    that is not on the top or bottom of the hour. The third condition, a null ``time``, is
    rejected earlier still, by the non-nullable field. ``validate`` is also used as a hard
    assertion in tests and R&D code, where a raise-on-violation contract must not silently
    change.

    Call this method BEFORE ``validate``, and only at a boundary that receives data from
    outside our system (e.g. NGED's raw JSON feed). The uniqueness and sortedness checks in
    ``validate`` are NOT relaxed here, and should keep raising. A duplicate row or an unsorted
    column indicates a bug in OUR pipeline rather than malformed external data.

    Args:
        dataframe: An already-cast frame with a ``time`` column; need not yet be validated.

    Returns:
        ``(survivors, n_dropped)``. ``survivors`` keeps ``dataframe``'s row order.
    """
    survivors, _ = split_by_datetime_plausibility(dataframe, "time")
    # `dt.minute()` is null for a null `time` and `.filter()` drops a row on a null predicate,
    # so this also drops the null `time`s the non-nullable schema forbids.
    survivors = survivors.filter(pl.col("time").dt.minute().is_in([0, 30]))
    # The dropped-row count is computed as a difference in `height` (a frame's row count),
    # rather than by summing the rejected partitions. Whatever a filter does with a null
    # predicate, a height difference counts every row that left exactly once.
    return DropImplausibleRowsResult(survivors, dataframe.height - survivors.height)

TimeSeriesMetadata

Bases: Model

One row per time series — a substation or a customer meter.

Carries the series' name, location, H3 index, and substation type.

Source code in packages/contracts/src/contracts/power_schemas.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
class TimeSeriesMetadata(pt.Model):
    """One row per time series — a substation or a customer meter.

    Carries the series' name, location, H3 index, and substation type.
    """

    time_series_id: int = _get_time_series_id_dtype(unique=True)

    time_series_name: str = pt.Field(
        dtype=pl.String,
        description="Human-readable name for the substation or asset.",
        examples=[
            "ALFORD 33 11kV S STN",
            "BAMBERS FARM WIND GENERATION MABLETHORPE 33kV S ST",
            "Leverton Solar Park",
        ],
    )

    time_series_type: str = pt.Field(
        dtype=pl.Enum(LIST_OF_TIME_SERIES_TYPES),
        description=(
            "Asset category (e.g. ‘PV’, ‘Wind’, ‘Disaggregated Demand’). See "
            "LIST_OF_TIME_SERIES_TYPES."
        ),
    )

    units: str = pt.Field(
        dtype=pl.Enum(["MW", "MVA"]),
        description=(
            "Power unit for this time series: ‘MW’ (active power) or ‘MVA’ (apparent power)."
        ),
    )

    licence_area: str = pt.Field(
        dtype=pl.Enum(["EMids"]),
        description="NGED licence area (for the trial area, this is always ‘EMids’).",
    )

    substation_number: int = pt.Field(
        dtype=pl.Int32,
        gt=0,
        lt=1_000_000,
        description=(
            "Each customer meter in the NGED trial area has its own "
            "substation_number (not one per physical substation)."
        ),
    )

    substation_type: str = pt.Field(
        dtype=pl.Enum(["BSP", "EHV Customer", "GSP", "HV Customer", "Primary"]),
        description=(
            "Substation voltage level / role: BSP, EHV Customer, GSP, HV Customer, or Primary."
            " BSP = bulk supply point. GSP = grid supply point. HV = high voltage."
            " EHV = extra high voltage."
        ),
    )

    latitude: float = pt.Field(
        dtype=pl.Float32,
        ge=49,
        le=61,  # UK latitude range
        description=(
            "Latitude in decimal degrees. For customer time series, gives the location of the "
            "substation, not the customer's site."
        ),
    )

    longitude: float = pt.Field(
        dtype=pl.Float32,
        ge=-9,
        le=2,  # UK longitude range
        description=(
            "Longitude in decimal degrees. For customer time series, gives the location of the "
            "substation, not the customer's site."
        ),
    )

    information: str | None = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        description="Free-text NGED notes field; always null in the V1 trial area.",
    )

    area_wkt: str | None = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        # Maps to the nested Area.WKT field in NGED's source JSON.
        description=(
            "Well-known text (WKT) polygon for the asset’s area. In the trial, only Primary"
            " substations have this. No customer site has a polygon yet. Where a customer site"
            " does have a polygon, the polygon refers to the area covered by the generator"
            " itself."
        ),
    )

    area_center_lat: float | None = pt.Field(
        dtype=pl.Float32,
        allow_missing=True,
        description=(
            "Centroid latitude of the area polygon. For customer sites, the area, where present, "
            "refers to the area covered by the generator itself."
        ),
    )

    area_center_lon: float | None = pt.Field(
        dtype=pl.Float32,
        allow_missing=True,
        description=(
            "Centroid longitude of the area polygon. For customer sites, the area, where present, "
            "refers to the area covered by the generator itself."
        ),
    )

    h3_res_5: int = pt.Field(
        dtype=pl.UInt64,
        description="H3 discrete spatial index at resolution 5.",
    )
Attributes
time_series_id = _get_time_series_id_dtype(unique=True) class-attribute instance-attribute
time_series_name = pt.Field(dtype=pl.String, description='Human-readable name for the substation or asset.', examples=['ALFORD 33 11kV S STN', 'BAMBERS FARM WIND GENERATION MABLETHORPE 33kV S ST', 'Leverton Solar Park']) class-attribute instance-attribute
time_series_type = pt.Field(dtype=pl.Enum(LIST_OF_TIME_SERIES_TYPES), description='Asset category (e.g. ‘PV’, ‘Wind’, ‘Disaggregated Demand’). See LIST_OF_TIME_SERIES_TYPES.') class-attribute instance-attribute
units = pt.Field(dtype=pl.Enum(['MW', 'MVA']), description='Power unit for this time series: ‘MW’ (active power) or ‘MVA’ (apparent power).') class-attribute instance-attribute
licence_area = pt.Field(dtype=pl.Enum(['EMids']), description='NGED licence area (for the trial area, this is always ‘EMids’).') class-attribute instance-attribute
substation_number = pt.Field(dtype=pl.Int32, gt=0, lt=1000000, description='Each customer meter in the NGED trial area has its own substation_number (not one per physical substation).') class-attribute instance-attribute
substation_type = pt.Field(dtype=pl.Enum(['BSP', 'EHV Customer', 'GSP', 'HV Customer', 'Primary']), description='Substation voltage level / role: BSP, EHV Customer, GSP, HV Customer, or Primary. BSP = bulk supply point. GSP = grid supply point. HV = high voltage. EHV = extra high voltage.') class-attribute instance-attribute
latitude = pt.Field(dtype=pl.Float32, ge=49, le=61, description="Latitude in decimal degrees. For customer time series, gives the location of the substation, not the customer's site.") class-attribute instance-attribute
longitude = pt.Field(dtype=pl.Float32, ge=-9, le=2, description="Longitude in decimal degrees. For customer time series, gives the location of the substation, not the customer's site.") class-attribute instance-attribute
information = pt.Field(dtype=pl.String, allow_missing=True, description='Free-text NGED notes field; always null in the V1 trial area.') class-attribute instance-attribute
area_wkt = pt.Field(dtype=pl.String, allow_missing=True, description='Well-known text (WKT) polygon for the asset’s area. In the trial, only Primary substations have this. No customer site has a polygon yet. Where a customer site does have a polygon, the polygon refers to the area covered by the generator itself.') class-attribute instance-attribute
area_center_lat = pt.Field(dtype=pl.Float32, allow_missing=True, description='Centroid latitude of the area polygon. For customer sites, the area, where present, refers to the area covered by the generator itself.') class-attribute instance-attribute
area_center_lon = pt.Field(dtype=pl.Float32, allow_missing=True, description='Centroid longitude of the area polygon. For customer sites, the area, where present, refers to the area covered by the generator itself.') class-attribute instance-attribute
h3_res_5 = pt.Field(dtype=pl.UInt64, description='H3 discrete spatial index at resolution 5.') class-attribute instance-attribute

PowerForecast

Bases: Model

Forecast data schema for an ensemble of deterministic forecasts.

Each ensemble member carries its own single-valued, or deterministic, forecast. The spread across the members is what expresses the forecast's uncertainty.

One row per time series, per forecast run, per target time, per ensemble member — the four columns of PRIMARY_KEY, declared below in that order.

Internal vs delivered schema (Milestone 1 report Table 1, p.28): three columns are INTERNAL-ONLY — experiment_name, fold_id, and ml_flow_experiment_id. They exist on this schema and on the internal power_forecasts Delta table, to support cross-validation and the leaderboard. They are NOT part of the power_forecast table delivered to NGED.

Source code in packages/contracts/src/contracts/power_schemas.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
class PowerForecast(pt.Model):
    """Forecast data schema for an ensemble of deterministic forecasts.

    Each ensemble member carries its own single-valued, or deterministic, forecast. The spread
    across the members is what expresses the forecast's uncertainty.

    One row per time series, per forecast run, per target time, per ensemble member — the four
    columns of ``PRIMARY_KEY``, declared below in that order.

    Internal vs delivered schema (Milestone 1 report Table 1, p.28): three columns are
    INTERNAL-ONLY — ``experiment_name``, ``fold_id``, and ``ml_flow_experiment_id``. They exist
    on this schema and on the internal ``power_forecasts`` Delta table, to support
    cross-validation and the leaderboard. They are NOT part of the ``power_forecast`` table
    delivered to NGED.
    """

    valid_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        constraints=pl.col("valid_time") > pl.col("power_fcst_init_time"),
        description=(
            "The target time this forecast is valid for. Constrained to be strictly after"
            " power_fcst_init_time: a row targeting a valid time at or before its own"
            " initialisation is an undeliverable hindcast row — at power_fcst_init_time that"
            " valid time is already observed. The live service only forecasts strictly future"
            " valid times and bulk-mode feature engineering drops hindcast rows at source, so"
            " a constraint violation here indicates a pipeline regression."
        ),
    )

    time_series_id: int = _get_time_series_id_dtype()

    ensemble_member: int = pt.Field(
        dtype=pl.Int8,
        description=(
            "Ensemble member index. Member 0 is the control numerical weather prediction (NWP)"
            " ensemble member."
        ),
    )

    ml_flow_experiment_id: int | None = pt.Field(
        dtype=pl.Int32,
        allow_missing=True,
        description=(
            "MLflow experiment ID; links to the MLflow experiment that produced this forecast."
        ),
    )

    nwp_init_time: datetime | None = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description=(
            "The datetime that the underlying weather forecast was initialised. "
            "Null for models that do not use NWP (e.g. persistence baselines)."
        ),
    )

    power_fcst_model_name: str = pt.Field(
        dtype=pl.String,  # String, not Categorical — see experiment_name below.
        description=(
            "Identifier for our ML-based power forecasting model. Model-family identity set by the "
            "BaseForecaster subclass (MODEL_NAME)."
        ),
    )

    # String (not Categorical): experiment_name/fold_id are the Delta partition columns, and
    # delta-rs stores dictionary-encoded columns as String anyway. String keeps those two columns
    # cast-free and lets predicate pushdown work. See the "Delta Lake dictionary-encoded columns"
    # section of the `polars-patito-gotchas` skill.
    experiment_name: str = pt.Field(
        dtype=pl.String,
        description=(
            "Per-experiment key identifying the experiment that produced this forecast."
            " Distinct from `power_fcst_model_name`, which is the model-family identity"
            " (`MODEL_NAME`); do not overload that with experiment identity."
            " Forecasts are partitioned in Delta by (experiment_name, fold_id)."
            " INTERNAL-ONLY: projected out of the `power_forecast` table delivered to NGED."
        ),
    )

    power_fcst_model_version: int = pt.Field(
        dtype=pl.Int16,
        description=(
            "Model version integer, bumped with each breaking change to the model implementation."
        ),
    )

    power_fcst_init_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "The datetime that the power forecast was initialised. This might be called `t0` in "
            "some other OCF projects."
        ),
    )

    power_fcst: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "The power forecast itself in units of MW (active power) or MVA (apparent power)."
            " The unit is defined in the `TimeSeriesMetadata` for this `time_series_id`."
            " This forecasts the power *available*: what the series would have exported or drawn"
            " had no active-network-management (ANM) instruction been in force. A generator on a"
            " flexible connection can be told to turn down when the local electricity network"
            " fills, and that instruction is a network decision rather than a weather outcome, so"
            " it falls outside what this column describes. Holding a delivered forecast down to a"
            " cap the operator has set is the consumer's own step, and the same assumption is what"
            " the `power_forecast` delivery block means by a normal running arrangement — see"
            " https://openclimatefix.github.io/nged-substation-forecast/roadmap/forecast-building-blocks/#the-idea."
            " `PowerTimeSeries.power` is the opposite convention: a metered value that already"
            " carries any curtailment the operator instructed."
            " Sign convention depends on `substation_type` in `TimeSeriesMetadata`, and describes"
            " a direction, so it applies only where `units` is `MW` — see the Sign convention"
            " section in this package's README.md, also published at"
            " https://openclimatefix.github.io/nged-substation-forecast/roadmap/forecast-building-blocks/#sign-convention."
            " Rows read back from the internal `power_forecasts` Delta table carry reduced"
            " precision: values are rounded to a 13-bit significand at write time"
            " (max relative error 2^-13 ≈ 1.2e-4, far below forecast error) to aid compression;"
            " see `delta_store.power_forecasts`."
            # PLANNED: We intend to change `power_fcst` to a normalised value in the range
            # [-1, +1], which NGED multiplies by a capacity to recover MW/MVA. That change follows
            # the delivery-contract design agreed with NGED in the Milestone 1 report. The switch
            # is planned for v0.5. The switch will use the static P99 `effective_capacity`
            # estimate that already exists — the same scalar the `metrics` pipeline already divides
            # by for normalised mean absolute error (NMAE). The switch therefore no longer waits
            # for a time-varying capacity estimate.
        ),
    )

    fold_id: FoldId = pt.Field(
        dtype=pl.String,  # String, not Categorical — see experiment_name above.
        description=(
            "Identifies the source of this forecast row.  "
            "For cross-validation runs, the value is the fold's label from conf/cv/default.yaml "
            "(e.g. 'mid_2025_to_mid_2026').  "
            "'live' means a production forecast with no associated CV fold.  "
            "All forecasts — CV and live — live in the same Delta table; "
            "filter on this column to select the population you need."
        ),
    )

    PRIMARY_KEY: ClassVar[tuple[str, ...]] = (
        "time_series_id",
        "power_fcst_init_time",
        "valid_time",
        "ensemble_member",
    )
    """At most one forecast per series, per init time, per target time, per ensemble member."""

    @classmethod
    def validate(  # ty: ignore[invalid-method-override]
        cls,
        dataframe: pl.DataFrame,
        columns: Sequence[str] | None = None,
        allow_missing_columns: bool = False,
        allow_superfluous_columns: bool = False,
        drop_superfluous_columns: bool = False,
    ) -> pt.DataFrame[Self]:
        """Validate the given dataframe, ensuring the primary key is unique.

        A duplicated primary key means either a join fanned out on the way here or the same rows
        were written twice, and both corrupt the metrics computed from them. It is our own bug
        rather than the outside world misbehaving, so this raises rather than degrading — see
        <https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/>.
        """
        validated_df = super().validate(
            dataframe=dataframe,
            columns=columns,
            allow_missing_columns=allow_missing_columns,
            allow_superfluous_columns=allow_superfluous_columns,
            drop_superfluous_columns=drop_superfluous_columns,
        )

        # `n_unique`, not `is_duplicated().any()`: the two expressions are equivalent here (every
        # primary-key column is non-nullable) but `is_duplicated` materialises a per-row mask,
        # costing ~5x the peak memory on a predict-sized frame.
        pk_cols = list(cls.PRIMARY_KEY)
        if validated_df.select(pk_cols).n_unique() != validated_df.height:
            raise ValueError(
                f"Duplicate entries found for primary key columns: {pk_cols}. "
                "Either an upstream join fanned out or these rows were written twice."
            )

        return validated_df
Attributes
valid_time = pt.Field(dtype=UTC_DATETIME_DTYPE, constraints=pl.col('valid_time') > pl.col('power_fcst_init_time'), description='The target time this forecast is valid for. Constrained to be strictly after power_fcst_init_time: a row targeting a valid time at or before its own initialisation is an undeliverable hindcast row — at power_fcst_init_time that valid time is already observed. The live service only forecasts strictly future valid times and bulk-mode feature engineering drops hindcast rows at source, so a constraint violation here indicates a pipeline regression.') class-attribute instance-attribute
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
ensemble_member = pt.Field(dtype=pl.Int8, description='Ensemble member index. Member 0 is the control numerical weather prediction (NWP) ensemble member.') class-attribute instance-attribute
ml_flow_experiment_id = pt.Field(dtype=pl.Int32, allow_missing=True, description='MLflow experiment ID; links to the MLflow experiment that produced this forecast.') class-attribute instance-attribute
nwp_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='The datetime that the underlying weather forecast was initialised. Null for models that do not use NWP (e.g. persistence baselines).') class-attribute instance-attribute
power_fcst_model_name = pt.Field(dtype=pl.String, description='Identifier for our ML-based power forecasting model. Model-family identity set by the BaseForecaster subclass (MODEL_NAME).') class-attribute instance-attribute
experiment_name = pt.Field(dtype=pl.String, description='Per-experiment key identifying the experiment that produced this forecast. Distinct from `power_fcst_model_name`, which is the model-family identity (`MODEL_NAME`); do not overload that with experiment identity. Forecasts are partitioned in Delta by (experiment_name, fold_id). INTERNAL-ONLY: projected out of the `power_forecast` table delivered to NGED.') class-attribute instance-attribute
power_fcst_model_version = pt.Field(dtype=pl.Int16, description='Model version integer, bumped with each breaking change to the model implementation.') class-attribute instance-attribute
power_fcst_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description='The datetime that the power forecast was initialised. This might be called `t0` in some other OCF projects.') class-attribute instance-attribute
power_fcst = pt.Field(dtype=pl.Float32, description="The power forecast itself in units of MW (active power) or MVA (apparent power). The unit is defined in the `TimeSeriesMetadata` for this `time_series_id`. This forecasts the power *available*: what the series would have exported or drawn had no active-network-management (ANM) instruction been in force. A generator on a flexible connection can be told to turn down when the local electricity network fills, and that instruction is a network decision rather than a weather outcome, so it falls outside what this column describes. Holding a delivered forecast down to a cap the operator has set is the consumer's own step, and the same assumption is what the `power_forecast` delivery block means by a normal running arrangement — see https://openclimatefix.github.io/nged-substation-forecast/roadmap/forecast-building-blocks/#the-idea. `PowerTimeSeries.power` is the opposite convention: a metered value that already carries any curtailment the operator instructed. Sign convention depends on `substation_type` in `TimeSeriesMetadata`, and describes a direction, so it applies only where `units` is `MW` — see the Sign convention section in this package's README.md, also published at https://openclimatefix.github.io/nged-substation-forecast/roadmap/forecast-building-blocks/#sign-convention. Rows read back from the internal `power_forecasts` Delta table carry reduced precision: values are rounded to a 13-bit significand at write time (max relative error 2^-13 ≈ 1.2e-4, far below forecast error) to aid compression; see `delta_store.power_forecasts`.") class-attribute instance-attribute
fold_id = pt.Field(dtype=pl.String, description="Identifies the source of this forecast row. For cross-validation runs, the value is the fold's label from conf/cv/default.yaml (e.g. 'mid_2025_to_mid_2026'). 'live' means a production forecast with no associated CV fold. All forecasts — CV and live — live in the same Delta table; filter on this column to select the population you need.") class-attribute instance-attribute
PRIMARY_KEY = ('time_series_id', 'power_fcst_init_time', 'valid_time', 'ensemble_member') class-attribute

At most one forecast per series, per init time, per target time, per ensemble member.

Methods:
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the given dataframe, ensuring the primary key is unique.

A duplicated primary key means either a join fanned out on the way here or the same rows were written twice, and both corrupt the metrics computed from them. It is our own bug rather than the outside world misbehaving, so this raises rather than degrading — see https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/.

Source code in packages/contracts/src/contracts/power_schemas.py 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
@classmethod
def validate(  # ty: ignore[invalid-method-override]
    cls,
    dataframe: pl.DataFrame,
    columns: Sequence[str] | None = None,
    allow_missing_columns: bool = False,
    allow_superfluous_columns: bool = False,
    drop_superfluous_columns: bool = False,
) -> pt.DataFrame[Self]:
    """Validate the given dataframe, ensuring the primary key is unique.

    A duplicated primary key means either a join fanned out on the way here or the same rows
    were written twice, and both corrupt the metrics computed from them. It is our own bug
    rather than the outside world misbehaving, so this raises rather than degrading — see
    <https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/>.
    """
    validated_df = super().validate(
        dataframe=dataframe,
        columns=columns,
        allow_missing_columns=allow_missing_columns,
        allow_superfluous_columns=allow_superfluous_columns,
        drop_superfluous_columns=drop_superfluous_columns,
    )

    # `n_unique`, not `is_duplicated().any()`: the two expressions are equivalent here (every
    # primary-key column is non-nullable) but `is_duplicated` materialises a per-row mask,
    # costing ~5x the peak memory on a predict-sized frame.
    pk_cols = list(cls.PRIMARY_KEY)
    if validated_df.select(pk_cols).n_unique() != validated_df.height:
        raise ValueError(
            f"Duplicate entries found for primary key columns: {pk_cols}. "
            "Either an upstream join fanned out or these rows were written twice."
        )

    return validated_df

EffectiveCapacity

Bases: Model

Effective capacity of each time series, at one or more half-hourly timesteps.

Effective capacity is an estimate of the power a site actually reaches, derived from its own observed history. Effective capacity is not a nameplate, firm, or connection-agreement rating.

Delivered to NGED as effective_capacity Delta table (Table 4 in the Milestone 1 report). This table is backward-looking only — it does not cover the forecast period.

v0.1 implementation: one row per time_series_id, time set to the end of the available observation history, effective_capacity_mw = P99 of abs(power) over the full observed history. The v0.1 estimate is a static scalar per series.

Planned upgrade (v0.7): replace the P99 scalar with a time-varying capacity estimate, giving one row per (time_series_id, time) half-hourly timestep. The candidate estimation methods are described at https://openclimatefix.github.io/nged-substation-forecast/techniques/convex-optimisation/ and https://openclimatefix.github.io/nged-substation-forecast/techniques/differentiable-physics/. This schema is unchanged. The effective_capacity asset body changes, and the metrics pipeline swaps its time_series_id-only normalised-mean-absolute-error-denominator join for a temporal as-of join, which matches each forecast row to the most recent capacity row at or before its timestamp. Do not pre-densify the v0.1 scalar into one row per half-hour — densifying a constant adds no information, and the as-of join handles sparse capacity rows naturally.

Source code in packages/contracts/src/contracts/power_schemas.py 
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
class EffectiveCapacity(pt.Model):
    """Effective capacity of each time series, at one or more half-hourly timesteps.

    Effective capacity is an estimate of the power a site actually reaches, derived from its own
    observed history. Effective capacity is not a nameplate, firm, or connection-agreement rating.

    Delivered to NGED as ``effective_capacity`` Delta table (Table 4 in the Milestone 1 report).
    This table is backward-looking only — it does not cover the forecast period.

    **v0.1 implementation:** one row per ``time_series_id``, ``time`` set to the end of the
    available observation history, ``effective_capacity_mw`` = P99 of ``abs(power)`` over the
    full observed history. The v0.1 estimate is a static scalar per series.

    **Planned upgrade (v0.7):** replace the P99 scalar with a time-varying capacity estimate,
    giving one row per ``(time_series_id, time)`` half-hourly timestep. The candidate estimation
    methods are described at
    <https://openclimatefix.github.io/nged-substation-forecast/techniques/convex-optimisation/>
    and
    <https://openclimatefix.github.io/nged-substation-forecast/techniques/differentiable-physics/>.
    This schema is unchanged. The ``effective_capacity`` asset body changes, and the ``metrics``
    pipeline swaps its ``time_series_id``-only normalised-mean-absolute-error-denominator join
    for a temporal as-of join, which matches each forecast row to the most recent capacity row at
    or before its timestamp. Do **not** pre-densify the v0.1 scalar into one row per half-hour —
    densifying a constant adds no information, and the as-of join handles sparse capacity rows
    naturally.
    """

    time_series_id: int = _get_time_series_id_dtype()

    time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "The half-hourly timestep this capacity estimate applies to. "
            "In v0.1, this is the end of the available observation history for that series."
        ),
    )

    effective_capacity_mw: float = pt.Field(
        dtype=pl.Float32,
        gt=0,
        description=(
            "OCF's estimate of the effective capacity of this asset at this timestep. "
            "Despite the column name, the value carries the series' own unit from "
            "`TimeSeriesMetadata.units`, so a series metered in MVA has its effective capacity "
            "in MVA. "
            "For generators: absorbs any persistent loss of capability, such as photovoltaic "
            "(PV) panel degradation or a partial inverter trip. The estimate ignores Active "
            "Network Management (ANM) curtailment — a wind farm ANM-capped at 5 MW with 10 MW "
            "physical capability has effective_capacity_mw = 10. "
            "For substations: the 99th percentile of observed absolute power flow, under normal "
            "running arrangement only. In v0.1 that percentile is taken over the full observed "
            "history, giving one static scalar per series. 'Switched' power (Table 5) should be "
            "added or subtracted when a switching event is in effect."
        ),
    )
Attributes
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
time = pt.Field(dtype=UTC_DATETIME_DTYPE, description='The half-hourly timestep this capacity estimate applies to. In v0.1, this is the end of the available observation history for that series.') class-attribute instance-attribute
effective_capacity_mw = pt.Field(dtype=pl.Float32, gt=0, description="OCF's estimate of the effective capacity of this asset at this timestep. Despite the column name, the value carries the series' own unit from `TimeSeriesMetadata.units`, so a series metered in MVA has its effective capacity in MVA. For generators: absorbs any persistent loss of capability, such as photovoltaic (PV) panel degradation or a partial inverter trip. The estimate ignores Active Network Management (ANM) curtailment — a wind farm ANM-capped at 5 MW with 10 MW physical capability has effective_capacity_mw = 10. For substations: the 99th percentile of observed absolute power flow, under normal running arrangement only. In v0.1 that percentile is taken over the full observed history, giving one static scalar per series. 'Switched' power (Table 5) should be added or subtracted when a switching event is in effect.") class-attribute instance-attribute

Functions:

contracts.weather_schemas

Contracts for numerical weather prediction (NWP) data.

The Nwp frame as stored and read, plus the run-completeness and data-quality reports that assess an ingested run of the European Centre for Medium-Range Weather Forecasts' ensemble forecast (ECMWF ENS).

Three words recur throughout this module and are defined nowhere else. A run is one execution of the weather model, and every row of one run shares an init_time. A slice is one (ensemble_member, valid_time) combination within a run, holding one value per H3 cell.

De-accumulation is the third word. ECMWF publishes precipitation and radiation as running totals accumulated since the start of the run, and Dynamical.org, who we download ECMWF ENS from, de-accumulates those fields into rates before we receive them. Nwp.deaccumulated_var_names names the three fields de-accumulation applies to, and records that all three are legitimately null at lead-0.

The rest of the vocabulary is defined where it is declared: the ensemble and its size on ECMWF_ENS_ENSEMBLE_MEMBERS, the forecast steps and lead-0 on ECMWF_ENS_LEAD_TIME_HOURS, and init_time and valid_time on the Nwp fields of those names.

Attributes

WeatherFeature = Literal['temperature_2m', 'dew_point_temperature_2m', 'wind_speed_10m', 'wind_direction_10m', 'wind_speed_100m', 'wind_direction_100m', 'pressure_surface', 'pressure_reduced_to_mean_sea_level', 'geopotential_height_500hpa', 'downward_long_wave_radiation_flux_surface', 'downward_short_wave_radiation_flux_surface', 'precipitation_surface', 'categorical_precipitation_type_surface'] module-attribute

ECMWF_ENS_ENSEMBLE_MEMBERS = frozenset(range(51)) module-attribute

The ensemble members every run of the ECMWF Integrated Forecasting System (IFS) ensemble carries: the control member plus 50 perturbed members, indexed 0-50 by Dynamical.org (matching Nwp.ensemble_member, which is 0-based).

The count is a fixed property of the ECMWF IFS ENS configuration, corroborated by our own data: the 2026-07-14 incident described in https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/ lost a forecast step for "50 of 51 ensemble members". The per-partition row arithmetic in https://openclimatefix.github.io/nged-substation-forecast/architecture/performance/ is 1671 H3 cells x 51 ensemble members x 85 native steps.

ECMWF_ENS_LEAD_TIME_HOURS = tuple(range(0, 145, 3)) + tuple(range(150, 361, 6)) module-attribute

The 85 native forecast steps of the ECMWF IFS ENS 15-day 0.25-degree dataset, in hours after init_time: 3-hourly out to 144 h (49 steps, lead-0 inclusive), then 6-hourly out to 360 h (36 steps) — i.e. init_time to init_time + 15 days inclusive.

Lead-0 is included: it is a real step, distinguished only by the de-accumulated variables being legitimately null there (see Nwp.deaccumulated_var_names).

ECMWF_ENS_H3_RESOLUTION = 5 module-attribute

The H3 spatial resolution the ecmwf_ens ingest pipeline grids ECMWF ENS data to, and the resolution power_time_series_and_metadata computes for every time series so the two line up on one grid. Every NWP model shares this one resolution today; per-model resolution (so a future NWP source can use a different resolution) is tracked in https://github.com/openclimatefix/nged-substation-forecast/issues/114.

Classes

NwpModelId

Bases: StrEnum

The NWP models we ingest, used as the nwp_model_id column's vocabulary.

Source code in packages/contracts/src/contracts/weather_schemas.py 
55
56
57
58
class NwpModelId(StrEnum):
    """The NWP models we ingest, used as the `nwp_model_id` column's vocabulary."""

    ECMWF_ENS_0_25_degree = auto()
Attributes
ECMWF_ENS_0_25_degree = auto() class-attribute instance-attribute

NwpVariableWhollyMissing

Bases: ValueError

A de-accumulated NWP variable is null in every slice beyond lead-0 of a run.

The field is wholesale absent rather than locally corrupt, so Nwp.validate rejects it. An all-null weather column would otherwise train and serve as a silently-degraded input for the full 15-day horizon. Falling back on the previous run is the better of the two.

A distinct exception type rather than a bare ValueError because the ecmwf_ens asset retries it. An upstream publication still in progress can present this way: a variable's chunks read as fill-value null until the worker writing them commits. Waiting is therefore a better first response than failing the partition.

The retry covers only the de-accumulated variables: an incomplete publication reaches this check only when the variables still unwritten are the de-accumulated ones. The nine instantaneous variables are non-nullable, so a frame missing one of those is rejected by base Patito validation first, with no retry. An all-null categorical_precipitation_type_surface passes base validation, because that column is nullable. The all-null column is rejected instead by _check_variables_that_were_introduced_after_start_of_dataset, and only for an init_time after 2024-11-12, because the column does not exist in the runs before then. See https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

Source code in packages/contracts/src/contracts/weather_schemas.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class NwpVariableWhollyMissing(ValueError):
    """A de-accumulated NWP variable is null in *every* slice beyond lead-0 of a run.

    The field is wholesale absent rather than locally corrupt, so `Nwp.validate` rejects it. An
    all-null weather column would otherwise train and serve as a silently-degraded input for the
    full 15-day horizon. Falling back on the previous run is the better of the two.

    A distinct exception type rather than a bare `ValueError` because the `ecmwf_ens` asset
    retries it. An upstream publication still in progress can present this way: a variable's
    chunks read as fill-value null until the worker writing them commits. Waiting is therefore a
    better first response than failing the partition.

    The retry covers only the de-accumulated variables: an incomplete publication reaches this
    check only when the variables still unwritten are the de-accumulated ones. The nine
    instantaneous variables are non-nullable, so a frame missing one of those is rejected by base
    Patito validation first, with no retry. An all-null `categorical_precipitation_type_surface`
    passes base validation, because that column is nullable. The all-null column is rejected
    instead by `_check_variables_that_were_introduced_after_start_of_dataset`, and only for an
    `init_time` after 2024-11-12, because the column does not exist in the runs before then. See
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """

Nwp

Bases: Model

Weather data schema for NWP forecasts.

Gridded ECMWF ENS ensemble weather, one row per (nwp_model_id, init_time, valid_time, ensemble_member, h3_index).

Stored on disk as plain Float32, rounded to a significand-bit budget by delta_store.nwp.write_nwp. Rounding the significand throws away the low-order bits of each value's mantissa, so what survives is a fixed number of significant figures rather than a fixed number of decimal places. See the delta_store.nwp module docstring for the physical format, and https://openclimatefix.github.io/nged-substation-forecast/architecture/performance/ for the measured table size and row counts.

validate is the fatal ingest gate. Two further checks are non-fatal. assess_nwp_quality reports the tolerated nulls in the de-accumulated variables, which are the known upstream ECMWF ENS corruption. assess_nwp_run_completeness reports a run that is missing whole members, steps, or cells. Neither fails the run, because each is the upstream provider misbehaving rather than a contract violation. Which patterns are fatal versus tolerated, and why, is documented at https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

Source code in packages/contracts/src/contracts/weather_schemas.py 
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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
class Nwp(pt.Model):
    """Weather data schema for NWP forecasts.

    Gridded ECMWF ENS ensemble weather, one row per (nwp_model_id, init_time, valid_time,
    ensemble_member, h3_index).

    Stored on disk as plain Float32, rounded to a significand-bit budget by
    `delta_store.nwp.write_nwp`. Rounding the significand throws away the low-order bits of each
    value's mantissa, so what survives is a fixed number of significant figures rather than a
    fixed number of decimal places. See the `delta_store.nwp` module docstring for the physical
    format, and
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/performance/> for the
    measured table size and row counts.

    `validate` is the fatal ingest gate. Two further checks are non-fatal. `assess_nwp_quality`
    reports the tolerated nulls in the de-accumulated variables, which are the known upstream
    ECMWF ENS corruption. `assess_nwp_run_completeness` reports a run that is missing whole
    members, steps, or cells. Neither fails the run, because each is the upstream provider
    misbehaving rather than a contract violation. Which patterns are fatal versus tolerated, and
    why, is documented at
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """

    # This column is `dtype=pl.String`, not `Enum`, because Delta cannot store an `Enum`. Under
    # `Enum`, an out-of-vocabulary value was physically inexpressible. A plain `str` column
    # enforces nothing at construction time: it can hold any string, with no error. The
    # `constraints=` check below is now the *only* defence, and it fires at `Nwp.validate()`, not
    # at construction: a frame built and never validated can carry an unrecognised `nwp_model_id`
    # all the way through. Every current caller validates, so the hole is latent rather than live
    # today.
    nwp_model_id: str = pt.Field(
        dtype=pl.String,
        constraints=pl.col("nwp_model_id").is_in([model.name for model in NwpModelId]),
        description="Which NWP model produced this row (e.g. 'ECMWF_ENS_0_25_degree').",
    )

    init_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description="When the NWP model run was initialised.",
    )

    valid_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "The time for which this NWP value is valid. Most variables (temperature, wind, "
            "pressure, geopotential) are instantaneous — they describe conditions at this moment. "
            "Precipitation and radiation (downward_long_wave_radiation_flux_surface, "
            "downward_short_wave_radiation_flux_surface, precipitation_surface) are period-ending "
            "rates: each value represents the average rate over the period that ends at valid_time "
            "(i.e. the preceding forecast step interval). Dynamical.org de-accumulates these from "
            "ECMWF's raw cumulative fields before we receive them. Resampling must honour that "
            "distinction — the step interval is 3 h out to lead 144 h and 6 h beyond it, so "
            "treating a period-ending value as instantaneous shifts it by up to 3 h. Conventions "
            "for every variable, and how much error the shift introduces: "
            "<https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>"
        ),
    )

    ensemble_member: int = pt.Field(
        dtype=pl.Int8,
        description="Ensemble member index (0-based).",
    )

    # `Int64`, not `UInt64`, because Delta has no unsigned integer types: the column is stored as
    # `int64` whatever we declare. Declaring `UInt64` only added a cast in `scan_delta`, and that
    # cast stopped Polars pushing `h3_index` filters into the Parquet scan.
    #
    # Signed is safe for the whole H3 index space, not just the values we happen to store. An H3
    # index is a 64-bit integer that packs an index mode, a resolution, and a cell address into
    # fixed ranges of bits. H3 reserves bit 63 of that 64-bit index as always zero. Bit 63 is not
    # a value bit in any index mode (cell, directed edge, vertex) at any resolution. Every H3
    # index is therefore below 2**63, which is exactly the range a signed `Int64` covers. A cell
    # index sets mode bits 59-62 to 1 and so tops out around 2**60, well inside that. Sampling
    # every resolution 0-15 agrees: the largest index seen is ~6.5e17 against `Int64`'s ~9.2e18
    # ceiling, and bit 63 is never set.
    #
    # The guarantee does not depend on the resolution. So this column cannot overflow if
    # `ECMWF_ENS_H3_RESOLUTION` is raised or lowered, nor if a future NWP model is given its own
    # resolution (issue 114). Resolution lives in bits 52-55 and leaves bit 63 alone.
    h3_index: int = pt.Field(
        dtype=pl.Int64,
        description=(
            "H3 cell index, at `ECMWF_ENS_H3_RESOLUTION` (every NWP model shares it today)."
        ),
    )

    temperature_2m: float = pt.Field(
        dtype=pl.Float32,
        description="Air temperature at 2 m above ground. Unit: degrees C.",
        ge=-100,
        le=100,
    )

    dew_point_temperature_2m: float = pt.Field(
        dtype=pl.Float32,
        description="Dew-point temperature at 2 m. Unit: degrees C.",
        ge=-100,
        le=100,
    )

    wind_speed_10m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind speed at 10 m. Unit: metres per second. This is the magnitude of the cell's"
            " *vector* mean wind, not the mean of its grid points' scalar speeds. See"
            " <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>"
        ),
        ge=0,
        # The 200 m/s ceiling is a corrupt-feed guard, not a meteorological limit. The highest
        # non-tornadic surface wind gust ever recorded is 113.2 m/s. That gust was measured on
        # Barrow Island during Tropical Cyclone Olivia on 10 April 1996. The World Meteorological
        # Organisation (WMO) ratified the record in its World Weather and Climate Extremes
        # Archive: <https://wmo.int/asu-map?map=Wind_028>. That record is a 3-second gust at a
        # single anemometer. This column holds an instantaneous wind averaged as a vector over a
        # whole H3 cell. A cell-mean vector wind is lower again than a 3-second gust at a point.
        # Only a unit error or a corrupt value can reach 200.
        le=200,
    )

    wind_direction_10m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind direction at 10 m. The angle where the wind is coming from. Degrees."
            " 0° is North; 90° is East. This is a *circular* quantity: 359° and 1° are two degrees"
            " apart, so interpolating, averaging, differencing or taking a quantile of it as if it"
            " were an ordinary number is wrong wherever the values straddle North. Convert to"
            " components first. See"
            " <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>"
        ),
        ge=0,
        le=360,
    )

    wind_speed_100m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind speed at 100 m. Unit: metres per second. The stored value is a vector mean,"
            " with the same caveat as `wind_speed_10m`: a turbine power curve responds to the"
            " scalar speed at each grid point, not to the cell's vector mean. A height of 100 m"
            " matters most for wind generation, being near a modern turbine's hub. See"
            " <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>"
        ),
        ge=0,
        le=200,  # The same corrupt-feed ceiling as `wind_speed_10m`; see the comment there.
    )

    wind_direction_100m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind direction at 100 m. The angle where the wind is coming from. Degrees. 0° is "
            "North; 90° is East. This is circular, with the same handling requirement as "
            "`wind_direction_10m`."
        ),
        ge=0,
        le=360,
    )

    pressure_surface: float = pt.Field(
        dtype=pl.Float32,
        description="Surface pressure. Unit: Pa.",
        ge=0,
        le=200_000,  # Max in 2 years of ECMWF ENS = 105_760 Pa
    )

    pressure_reduced_to_mean_sea_level: float = pt.Field(
        dtype=pl.Float32,
        description="Mean sea-level pressure. Unit: Pa.",
        ge=0,
        le=200_000,  # Max in 2 years of ECMWF ENS = 105_727 Pa
    )

    geopotential_height_500hpa: float = pt.Field(
        dtype=pl.Float32,
        description="Geopotential height of the 500 hPa pressure surface. Unit: m.",
        ge=0,
        le=10_000,  # Max in 2 years of ECMWF ENS = 6_030 m
    )

    # Precipitation and radiation variables are null for the first forecast step (lead time 0) in
    # ECMWF ENS. These variables accumulate over forecast steps in ECMWF's raw forecasts. We get
    # ECMWF ENS from Dynamical.org, and Dynamical.org de-accumulates these values before we
    # receive them. So the values stored in these three columns are true _rates_.
    downward_long_wave_radiation_flux_surface: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "Downward long-wave radiation flux at surface. Note that this variable is all-null for "
            "lead time 0. Unit: W m-2."
        ),
        ge=0,
        le=1500,  # Max in 2 years of ECMWF ENS = 445 W m-2
    )

    downward_short_wave_radiation_flux_surface: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "Downward short-wave (solar) radiation flux at surface. Note that this variable is"
            " all-null for lead time 0. Unit: W m-2."
        ),
        ge=0,
        le=1500,  # Max in 2 years of ECMWF ENS = 892 W m-2
    )

    precipitation_surface: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "Total precipitation rate at surface, de-accumulated by Dynamical. Note that this "
            "variable is all-null for lead time 0. Unit: kg m-2 s-1."
        ),
        ge=0,
        le=0.01,  # Max in 2 years of ECMWF ENS = 0.006 kg m-2 s-1
    )

    categorical_precipitation_type_surface: int | None = pt.Field(
        dtype=pl.Int16,
        ge=0,
        le=255,
        description=(
            "This field is always null for init_times on and before 2024-11-12, and populated from"
            " the 2024-11-13 00Z run onwards (confirmed by direct inspection of the source data)."
            " ECMWF's NaNs are converted to nulls at the ingest boundary, so the check tests for"
            " null rather than for NaN, and an Int16 column could not carry a NaN in any case."
            " Derived from ECMWF's `ptype` field. See https://codes.ecmwf.int/grib/param-db/260015"
            " 0=No precipitation; 1=Rain; 2=Thunderstorm; 3=Freezing rain; 4=Mixed/ice;"
            " 5=Snow; 6=Wet snow; 7=Mixture of rain and snow; 8=Ice pellets; 9=Graupel;"
            " 10=Hail; 11=Drizzle; 12=Freezing drizzle; 13=Hail (less than 5 mm);"
            " 14=Hail (greater than or equal to 5 mm);"
            " 15-191=Reserved; 192-254=Reserved for local use; 255=Missing"
        ),
    )

    # ClassVars: excluded from Patito/Pydantic model fields so they're not treated as data columns.
    categorical_var_names: ClassVar[frozenset[str]] = frozenset(
        {"categorical_precipitation_type_surface"}
    )

    deaccumulated_var_names: ClassVar[frozenset[str]] = frozenset(
        {
            "precipitation_surface",
            "downward_short_wave_radiation_flux_surface",
            "downward_long_wave_radiation_flux_surface",
        }
    )
    """The variables Dynamical.org de-accumulates from ECMWF's cumulative source fields to rates.

    All three are legitimately null at lead-0, and they share a de-accumulation step whose known
    upstream corruption leaves further nulls beyond it. Those are tolerated at ingest and reported
    by `assess_nwp_quality`; only a variable null in *every* slice beyond lead-0 is fatal.
    Which corruption patterns arrive, what the H3 aggregation absorbs before they reach a validated
    frame, and why the survivors are tolerated:
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#nulls-in-the-de-accumulated-variables-tolerated>.
    """

    # Columns that aren't NWP variables:
    _non_var_column_names: ClassVar[frozenset[str]] = frozenset(
        {"nwp_model_id", "init_time", "valid_time", "ensemble_member", "h3_index"}
    )

    @classmethod
    def all_weather_var_names(cls) -> frozenset[str]:
        """All meteorological variable field names (continuous + categorical)."""
        return frozenset(cls.model_fields) - cls._non_var_column_names

    @classmethod
    def continuous_var_names(cls) -> frozenset[str]:
        """Meteorological variable field names suitable for linear interpolation."""
        return cls.all_weather_var_names() - cls.categorical_var_names

    @classmethod
    def validate(
        cls,
        dataframe: pl.DataFrame,
        columns: Sequence[str] | None = None,
        allow_missing_columns: bool = False,
        allow_superfluous_columns: bool = False,
        drop_superfluous_columns: bool = False,
    ) -> pt.DataFrame[Self]:  # ty:ignore[invalid-method-override]
        """Validate the frame.

        Bounds `init_time`/`valid_time` to the plausible datetime range. Rejects a de-accumulated
        variable that is wholly missing, tolerating every smaller null pattern. Enforces
        uniqueness. Enforces the introduction-date rule for
        `categorical_precipitation_type_surface` — the column derived from ECMWF's `ptype` field,
        whose own description states the rule and the date.
        """
        validated_df = super().validate(
            dataframe=dataframe,
            columns=columns,
            allow_missing_columns=allow_missing_columns,
            allow_superfluous_columns=allow_superfluous_columns,
            drop_superfluous_columns=drop_superfluous_columns,
        )

        # Patito ignores `ge`/`le` on datetime fields, so the range check lives here.
        check_datetime_bounds(validated_df, "init_time", "valid_time")
        cls._check_no_wholly_missing_deaccumulated_variable(validated_df)
        cls._check_unique(validated_df)
        cls._check_variables_that_were_introduced_after_start_of_dataset(validated_df)
        return validated_df

    @classmethod
    def _check_no_wholly_missing_deaccumulated_variable(cls, dataframe: pt.DataFrame[Self]) -> None:
        """Reject a de-accumulated variable that carries no weather at all.

        That is, one that is null in *every* (ensemble_member, valid_time) slice beyond lead-0 of
        a run.

        Every smaller null pattern is *tolerated* and reported by `assess_nwp_quality` instead,
        so the check is a cliff rather than a slope: a run one slice short of empty lands with a
        warning. There is no tunable fraction — the test is that *nothing* survives.

        Why an absent column is worth discarding a run over:
        <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#a-wholly-missing-variable-and-instantaneous-nulls-fatal>.
        Why every smaller pattern is not — the slice arithmetic, and the interpolation argument that
        makes a tolerated slice survivable:
        <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#nulls-in-the-de-accumulated-variables-tolerated>.

        Two assumptions a caller must not make:

        - The judgement is made per `init_time`. A run whose column is empty is therefore caught
          even inside a frame holding other, healthy runs. The same per-`init_time` judgement has
          an unwanted consequence: this check does raise when a frame is filtered down to nothing
          *but* a wholly-null slice, which is indistinguishable from an empty column. That is
          true even though that same slice was deliberately landed — written into the stored
          table — when the whole run was validated. That raise is latent rather than live today.
          The production caller, `dynamical_data.ecmwf_ens.convert_to_polars`, validates one
          whole run. Reads go through `scan_delta`/`set_model`, which do not validate.
        - Raising is not the end of the partition. `NwpVariableWhollyMissing` is a distinct type
          because the `ecmwf_ens` asset retries it rather than failing outright.
        """
        slices_per_run = (
            dataframe.filter(pl.col("valid_time") > pl.col("init_time"))
            .group_by("init_time")
            .agg(n_slices=pl.struct("ensemble_member", "valid_time").n_unique())
        )
        wholly_missing = (
            _deaccumulated_null_breakdown(dataframe)
            .filter(pl.col("n_null") == pl.col("n_total"))
            .group_by("variable", "init_time")
            .agg(n_whole_null_slices=pl.len())
            # Per `init_time`: comparing a variable's wholly-null slice count against the frame's
            # *total* would let a wholly-missing run hide behind a healthy one in the same frame.
            .join(slices_per_run, on="init_time", how="inner")
            .filter(pl.col("n_whole_null_slices") == pl.col("n_slices"))
        )
        if wholly_missing.height:
            offenders = sorted(
                (row["variable"], str(row["init_time"]), row["n_slices"])
                for row in wholly_missing.iter_rows(named=True)
            )
            raise NwpVariableWhollyMissing(
                "De-accumulated NWP variable(s) null in every (ensemble_member, valid_time) slice "
                "beyond lead-0 of their run — the field is wholesale absent, not the "
                f"locally-corrupt scatter we tolerate. Offending (variable, init_time, n_slices): "
                f"{offenders}"
            )

    @classmethod
    def _check_unique(cls, dataframe: pt.DataFrame[Self]) -> None:
        if (
            dataframe.select(
                ["nwp_model_id", "init_time", "valid_time", "ensemble_member", "h3_index"]
            )
            .is_duplicated()
            .any()
        ):
            raise ValueError(
                "Duplicate entries found for "
                "(nwp_model_id, init_time, valid_time, ensemble_member, h3_index)."
            )

    @classmethod
    def _check_variables_that_were_introduced_after_start_of_dataset(
        cls, dataframe: pt.DataFrame[Self]
    ) -> None:
        """Check the introduction date of `categorical_precipitation_type_surface`.

        The column is all-null when init_time <= 2024-11-12, and is never null afterwards.

        Direct inspection of the source data confirms the dates. The 2024-11-13 00Z run — the run
        initialised at 00:00 UTC that day — is the first run with `ptype` populated, 0% null
        across all lead times. The 2024-11-12 run and every earlier run are 100% null.
        """
        threshold_date = datetime(2024, 11, 12, tzinfo=UTC)

        # Split the frame at the threshold date, so each half can be checked against the opposite
        # expectation: all-null before, never-null after.
        partition_col = "is_before_or_on_threshold"
        partitioned_df = dataframe.with_columns(
            (pl.col("init_time") <= threshold_date).alias(partition_col)
        )
        partitions = partitioned_df.partition_by(partition_col, as_dict=True)

        empty_df = pl.DataFrame(schema=dataframe.schema)
        before_or_on = partitions.get((True,), empty_df)
        after = partitions.get((False,), empty_df)

        # Check before or on threshold
        if not before_or_on["categorical_precipitation_type_surface"].is_null().all():
            raise ValueError(
                "categorical_precipitation_type_surface must be all null for "
                "init_time <= 2024-11-12"
            )

        # Check after threshold
        if after["categorical_precipitation_type_surface"].is_null().any():
            raise ValueError(
                "categorical_precipitation_type_surface must not be null for init_time > 2024-11-12"
            )

    @classmethod
    def scan_delta(
        cls,
        path: str | Path | None = None,
        storage_options: ObjectStoreOptions | None = None,
    ) -> pt.LazyFrame[Self]:
        """Lazily scan the NWP Delta table, typed and cast to this contract's dtypes.

        The table stores physical-unit `Float32` directly (see `delta_store.nwp.write_nwp`),
        so no rescale step is needed.

        Args:
            path: Path or URI of the ``nwp`` Delta table; defaults to
                ``get_settings().nwp_data_path`` (resolved lazily so importing this module needs
                no ``.env``).
            storage_options: delta-rs object-store options (credentials/endpoint) for a remote
                ``path``; defaults to ``get_settings().storage_options`` (empty for a local
                ``path``).

        Returns:
            A lazy frame over the ``nwp`` Delta table, set to the ``Nwp`` model and cast to its
            declared dtypes — one row per (NWP model, init time, ensemble member, valid time,
            H3 cell), with the weather variables already in physical units.
        """
        settings = get_settings()
        if path is None:
            path = settings.nwp_data_path
        if storage_options is None:
            storage_options = settings.storage_options
        return (
            pt.LazyFrame.from_existing(
                pl.scan_delta(path, storage_options=typeddict_to_dict(storage_options))
            )
            .set_model(cls)
            .cast()
        )
Attributes
nwp_model_id = pt.Field(dtype=pl.String, constraints=pl.col('nwp_model_id').is_in([model.name for model in NwpModelId]), description="Which NWP model produced this row (e.g. 'ECMWF_ENS_0_25_degree').") class-attribute instance-attribute
init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description='When the NWP model run was initialised.') class-attribute instance-attribute
valid_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description="The time for which this NWP value is valid. Most variables (temperature, wind, pressure, geopotential) are instantaneous — they describe conditions at this moment. Precipitation and radiation (downward_long_wave_radiation_flux_surface, downward_short_wave_radiation_flux_surface, precipitation_surface) are period-ending rates: each value represents the average rate over the period that ends at valid_time (i.e. the preceding forecast step interval). Dynamical.org de-accumulates these from ECMWF's raw cumulative fields before we receive them. Resampling must honour that distinction — the step interval is 3 h out to lead 144 h and 6 h beyond it, so treating a period-ending value as instantaneous shifts it by up to 3 h. Conventions for every variable, and how much error the shift introduces: <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>") class-attribute instance-attribute
ensemble_member = pt.Field(dtype=pl.Int8, description='Ensemble member index (0-based).') class-attribute instance-attribute
h3_index = pt.Field(dtype=pl.Int64, description='H3 cell index, at `ECMWF_ENS_H3_RESOLUTION` (every NWP model shares it today).') class-attribute instance-attribute
temperature_2m = pt.Field(dtype=pl.Float32, description='Air temperature at 2 m above ground. Unit: degrees C.', ge=-100, le=100) class-attribute instance-attribute
dew_point_temperature_2m = pt.Field(dtype=pl.Float32, description='Dew-point temperature at 2 m. Unit: degrees C.', ge=-100, le=100) class-attribute instance-attribute
wind_speed_10m = pt.Field(dtype=pl.Float32, description="Wind speed at 10 m. Unit: metres per second. This is the magnitude of the cell's *vector* mean wind, not the mean of its grid points' scalar speeds. See <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>", ge=0, le=200) class-attribute instance-attribute
wind_direction_10m = pt.Field(dtype=pl.Float32, description='Wind direction at 10 m. The angle where the wind is coming from. Degrees. 0° is North; 90° is East. This is a *circular* quantity: 359° and 1° are two degrees apart, so interpolating, averaging, differencing or taking a quantile of it as if it were an ordinary number is wrong wherever the values straddle North. Convert to components first. See <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>', ge=0, le=360) class-attribute instance-attribute
wind_speed_100m = pt.Field(dtype=pl.Float32, description="Wind speed at 100 m. Unit: metres per second. The stored value is a vector mean, with the same caveat as `wind_speed_10m`: a turbine power curve responds to the scalar speed at each grid point, not to the cell's vector mean. A height of 100 m matters most for wind generation, being near a modern turbine's hub. See <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>", ge=0, le=200) class-attribute instance-attribute
wind_direction_100m = pt.Field(dtype=pl.Float32, description='Wind direction at 100 m. The angle where the wind is coming from. Degrees. 0° is North; 90° is East. This is circular, with the same handling requirement as `wind_direction_10m`.', ge=0, le=360) class-attribute instance-attribute
pressure_surface = pt.Field(dtype=pl.Float32, description='Surface pressure. Unit: Pa.', ge=0, le=200000) class-attribute instance-attribute
pressure_reduced_to_mean_sea_level = pt.Field(dtype=pl.Float32, description='Mean sea-level pressure. Unit: Pa.', ge=0, le=200000) class-attribute instance-attribute
geopotential_height_500hpa = pt.Field(dtype=pl.Float32, description='Geopotential height of the 500 hPa pressure surface. Unit: m.', ge=0, le=10000) class-attribute instance-attribute
downward_long_wave_radiation_flux_surface = pt.Field(dtype=pl.Float32, description='Downward long-wave radiation flux at surface. Note that this variable is all-null for lead time 0. Unit: W m-2.', ge=0, le=1500) class-attribute instance-attribute
downward_short_wave_radiation_flux_surface = pt.Field(dtype=pl.Float32, description='Downward short-wave (solar) radiation flux at surface. Note that this variable is all-null for lead time 0. Unit: W m-2.', ge=0, le=1500) class-attribute instance-attribute
precipitation_surface = pt.Field(dtype=pl.Float32, description='Total precipitation rate at surface, de-accumulated by Dynamical. Note that this variable is all-null for lead time 0. Unit: kg m-2 s-1.', ge=0, le=0.01) class-attribute instance-attribute
categorical_precipitation_type_surface = pt.Field(dtype=pl.Int16, ge=0, le=255, description="This field is always null for init_times on and before 2024-11-12, and populated from the 2024-11-13 00Z run onwards (confirmed by direct inspection of the source data). ECMWF's NaNs are converted to nulls at the ingest boundary, so the check tests for null rather than for NaN, and an Int16 column could not carry a NaN in any case. Derived from ECMWF's `ptype` field. See https://codes.ecmwf.int/grib/param-db/260015 0=No precipitation; 1=Rain; 2=Thunderstorm; 3=Freezing rain; 4=Mixed/ice; 5=Snow; 6=Wet snow; 7=Mixture of rain and snow; 8=Ice pellets; 9=Graupel; 10=Hail; 11=Drizzle; 12=Freezing drizzle; 13=Hail (less than 5 mm); 14=Hail (greater than or equal to 5 mm); 15-191=Reserved; 192-254=Reserved for local use; 255=Missing") class-attribute instance-attribute
categorical_var_names = frozenset({'categorical_precipitation_type_surface'}) class-attribute
deaccumulated_var_names = frozenset({'precipitation_surface', 'downward_short_wave_radiation_flux_surface', 'downward_long_wave_radiation_flux_surface'}) class-attribute

The variables Dynamical.org de-accumulates from ECMWF's cumulative source fields to rates.

All three are legitimately null at lead-0, and they share a de-accumulation step whose known upstream corruption leaves further nulls beyond it. Those are tolerated at ingest and reported by assess_nwp_quality; only a variable null in every slice beyond lead-0 is fatal. Which corruption patterns arrive, what the H3 aggregation absorbs before they reach a validated frame, and why the survivors are tolerated: https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#nulls-in-the-de-accumulated-variables-tolerated.

Methods:
all_weather_var_names() classmethod

All meteorological variable field names (continuous + categorical).

Source code in packages/contracts/src/contracts/weather_schemas.py 
374
375
376
377
@classmethod
def all_weather_var_names(cls) -> frozenset[str]:
    """All meteorological variable field names (continuous + categorical)."""
    return frozenset(cls.model_fields) - cls._non_var_column_names
continuous_var_names() classmethod

Meteorological variable field names suitable for linear interpolation.

Source code in packages/contracts/src/contracts/weather_schemas.py 
379
380
381
382
@classmethod
def continuous_var_names(cls) -> frozenset[str]:
    """Meteorological variable field names suitable for linear interpolation."""
    return cls.all_weather_var_names() - cls.categorical_var_names
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the frame.

Bounds init_time/valid_time to the plausible datetime range. Rejects a de-accumulated variable that is wholly missing, tolerating every smaller null pattern. Enforces uniqueness. Enforces the introduction-date rule for categorical_precipitation_type_surface — the column derived from ECMWF's ptype field, whose own description states the rule and the date.

Source code in packages/contracts/src/contracts/weather_schemas.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
@classmethod
def validate(
    cls,
    dataframe: pl.DataFrame,
    columns: Sequence[str] | None = None,
    allow_missing_columns: bool = False,
    allow_superfluous_columns: bool = False,
    drop_superfluous_columns: bool = False,
) -> pt.DataFrame[Self]:  # ty:ignore[invalid-method-override]
    """Validate the frame.

    Bounds `init_time`/`valid_time` to the plausible datetime range. Rejects a de-accumulated
    variable that is wholly missing, tolerating every smaller null pattern. Enforces
    uniqueness. Enforces the introduction-date rule for
    `categorical_precipitation_type_surface` — the column derived from ECMWF's `ptype` field,
    whose own description states the rule and the date.
    """
    validated_df = super().validate(
        dataframe=dataframe,
        columns=columns,
        allow_missing_columns=allow_missing_columns,
        allow_superfluous_columns=allow_superfluous_columns,
        drop_superfluous_columns=drop_superfluous_columns,
    )

    # Patito ignores `ge`/`le` on datetime fields, so the range check lives here.
    check_datetime_bounds(validated_df, "init_time", "valid_time")
    cls._check_no_wholly_missing_deaccumulated_variable(validated_df)
    cls._check_unique(validated_df)
    cls._check_variables_that_were_introduced_after_start_of_dataset(validated_df)
    return validated_df
scan_delta(path=None, storage_options=None) classmethod

Lazily scan the NWP Delta table, typed and cast to this contract's dtypes.

The table stores physical-unit Float32 directly (see delta_store.nwp.write_nwp), so no rescale step is needed.

Parameters:

Name Type Description Default
path str | Path | None

Path or URI of the nwp Delta table; defaults to get_settings().nwp_data_path (resolved lazily so importing this module needs no .env).

 None
storage_options ObjectStoreOptions | None

delta-rs object-store options (credentials/endpoint) for a remote path; defaults to get_settings().storage_options (empty for a local path).

 None

Returns:

Type Description
LazyFrame[Self]

A lazy frame over the nwp Delta table, set to the Nwp model and cast to its
LazyFrame[Self]

declared dtypes — one row per (NWP model, init time, ensemble member, valid time,
LazyFrame[Self]

H3 cell), with the weather variables already in physical units.
Source code in packages/contracts/src/contracts/weather_schemas.py 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
@classmethod
def scan_delta(
    cls,
    path: str | Path | None = None,
    storage_options: ObjectStoreOptions | None = None,
) -> pt.LazyFrame[Self]:
    """Lazily scan the NWP Delta table, typed and cast to this contract's dtypes.

    The table stores physical-unit `Float32` directly (see `delta_store.nwp.write_nwp`),
    so no rescale step is needed.

    Args:
        path: Path or URI of the ``nwp`` Delta table; defaults to
            ``get_settings().nwp_data_path`` (resolved lazily so importing this module needs
            no ``.env``).
        storage_options: delta-rs object-store options (credentials/endpoint) for a remote
            ``path``; defaults to ``get_settings().storage_options`` (empty for a local
            ``path``).

    Returns:
        A lazy frame over the ``nwp`` Delta table, set to the ``Nwp`` model and cast to its
        declared dtypes — one row per (NWP model, init time, ensemble member, valid time,
        H3 cell), with the weather variables already in physical units.
    """
    settings = get_settings()
    if path is None:
        path = settings.nwp_data_path
    if storage_options is None:
        storage_options = settings.storage_options
    return (
        pt.LazyFrame.from_existing(
            pl.scan_delta(path, storage_options=typeddict_to_dict(storage_options))
        )
        .set_model(cls)
        .cast()
    )

NwpQualityReport dataclass

Non-fatal data-quality summary for one NWP run.

Carries the tolerated nulls that the de-accumulated variables still hold after the H3 spatial aggregation, beyond lead-0. ECMWF publishes on its own latitude/longitude grid, and several of those grid points fall inside each H3 cell, so a cell goes null only when every grid point contributing to it is null. Those tolerated nulls are usually whole (ensemble_member, valid_time) slices that arrived empty, plus the cells where upstream scatter — isolated missing grid points strewn across the raw grid — happened to take out every contributing grid point. The empty slices and the scattered cells are counted separately, because each warrants a different response. Neither fails the run. Only a variable that is null in every slice is fatal, and Nwp.validate rejects that variable before this report runs.

Read this report as "how much did we lose", not as "how corrupt was the feed". The aggregation absorbs most per-pixel upstream corruption before it reaches a cell, so this report is a poor proxy for the upstream null rate. dynamical_data.ecmwf_ens.upstream_nulls.UpstreamNullRate measures the upstream null rate on the raw grid, where that rate lives. The ecmwf_ens asset publishes this report and the upstream null rate on one check, but the two are not comparable as rates.

Source code in packages/contracts/src/contracts/weather_schemas.py 
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
@dataclass(frozen=True)
class NwpQualityReport:
    """Non-fatal data-quality summary for one NWP run.

    Carries the *tolerated* nulls that the de-accumulated variables still hold after the H3
    spatial aggregation, beyond lead-0. ECMWF publishes on its own latitude/longitude grid, and
    several of those grid points fall inside each H3 cell, so a cell goes null only when every
    grid point contributing to it is null. Those tolerated nulls are usually whole
    (ensemble_member, valid_time) slices that arrived empty, plus the cells where upstream
    scatter — isolated missing grid points strewn across the raw grid — happened to take out
    every contributing grid point. The empty slices and the scattered cells are counted
    separately, because each warrants a different response. Neither fails the run. Only a
    variable that is null in *every* slice is fatal, and `Nwp.validate` rejects that variable
    before this report runs.

    Read this report as "how much did we lose", not as "how corrupt was the feed". The
    aggregation absorbs most per-pixel upstream corruption before it reaches a cell, so this
    report is a poor proxy for the upstream null rate.
    `dynamical_data.ecmwf_ens.upstream_nulls.UpstreamNullRate` measures the upstream null rate on
    the raw grid, where that rate lives. The ``ecmwf_ens`` asset publishes this report and the
    upstream null rate on one check, but the two are not comparable as rates.
    """

    affected: pl.DataFrame
    """One row per affected (variable, init_time, ensemble_member, valid_time) slice, with
    ``n_null`` and ``n_total`` cell counts. Empty when the run is clean."""

    @property
    def n_null_cells(self) -> int:
        """Total null cells across all affected slices."""
        return int(self.affected["n_null"].sum()) if self.affected.height else 0

    @property
    def n_affected_slices(self) -> int:
        """Number of (variable, member, valid_time) slices carrying at least one null."""
        return self.affected.height

    @property
    def n_whole_null_slices(self) -> int:
        """Affected slices that arrived *entirely* null.

        The field is missing altogether for that one (variable, member, valid_time) slice.

        Worth watching separately from the scattered case below: a rising count is the shape a
        partial upstream publication takes. The count is exact, because a wholly-null slice
        reaches the cells intact however the grid points are aggregated.
        """
        if not self.affected.height:
            return 0
        return int((self.affected["n_null"] == self.affected["n_total"]).sum())

    @property
    def n_scattered_slices(self) -> int:
        """Affected slices carrying only *some* null cells.

        Expect this count to be small. Scattered upstream corruption is mostly absorbed by the H3
        aggregation, so a slice reaches this count only where the scatter took out every grid
        point of some cell. The count is *not* a measure of the upstream per-pixel null rate.
        """
        return self.n_affected_slices - self.n_whole_null_slices

    @property
    def is_healthy(self) -> bool:
        """True when the run has no unexpected nulls."""
        return self.affected.height == 0

    @property
    def affected_variables(self) -> tuple[str, ...]:
        """The de-accumulated variables carrying unexpected nulls, sorted."""
        if not self.affected.height:
            return ()
        return tuple(sorted(self.affected["variable"].unique().to_list()))
Attributes
affected instance-attribute

One row per affected (variable, init_time, ensemble_member, valid_time) slice, with n_null and n_total cell counts. Empty when the run is clean.

n_null_cells property

Total null cells across all affected slices.

n_affected_slices property

Number of (variable, member, valid_time) slices carrying at least one null.

n_whole_null_slices property

Affected slices that arrived entirely null.

The field is missing altogether for that one (variable, member, valid_time) slice.

Worth watching separately from the scattered case below: a rising count is the shape a partial upstream publication takes. The count is exact, because a wholly-null slice reaches the cells intact however the grid points are aggregated.

n_scattered_slices property

Affected slices carrying only some null cells.

Expect this count to be small. Scattered upstream corruption is mostly absorbed by the H3 aggregation, so a slice reaches this count only where the scatter took out every grid point of some cell. The count is not a measure of the upstream per-pixel null rate.

is_healthy property

True when the run has no unexpected nulls.

affected_variables property

The de-accumulated variables carrying unexpected nulls, sorted.

Methods:
__init__(affected)

NwpRunCompletenessReport dataclass

Run-level shape summary for one ingested NWP run.

Answers a single question: does the run carry every (member, step, cell) combination we expect?

Deliberately a report rather than an exception. A short run is the upstream provider misbehaving, not a contract violation. The never-raise-on-absent-input rule says we land what arrived and warn, rather than throwing away an otherwise-good run. The ecmwf_ens asset wraps this into a WARN, non-blocking AssetCheckResult, which warns an operator without stopping the pipeline. The asset also publishes the counts as materialisation metadata, the numbers Dagster records against each run of an asset.

Source code in packages/contracts/src/contracts/weather_schemas.py 
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
@dataclass(frozen=True)
class NwpRunCompletenessReport:
    """Run-level shape summary for one ingested NWP run.

    Answers a single question: does the run carry every (member, step, cell) combination we
    expect?

    Deliberately a *report* rather than an exception. A short run is the upstream provider
    misbehaving, not a contract violation. The [never-raise-on-absent-input
    rule](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/#the-rules)
    says we land what arrived and warn, rather than throwing away an otherwise-good run. The
    `ecmwf_ens` asset wraps this into a WARN, non-blocking `AssetCheckResult`, which warns an
    operator without stopping the pipeline. The asset also publishes the counts as materialisation
    metadata, the numbers Dagster records against each run of an asset.
    """

    init_times: tuple[datetime, ...]
    """Distinct `init_time` values in the frame. A whole run has exactly one."""

    n_rows: int
    """Rows in the frame."""

    expected_n_rows: int
    """`len(expected_ensemble_members) * len(expected_lead_time_hours) * expected_n_h3_cells` —
    the size of the complete grid."""

    n_ensemble_members: int
    """Distinct `ensemble_member` values observed."""

    n_valid_times: int
    """Distinct `valid_time` values observed."""

    n_h3_cells: int
    """Distinct `h3_index` values observed."""

    expected_n_h3_cells: int
    """Distinct `h3_index` values the H3 grid weights say this run should cover.

    The H3 grid weights are the `contracts.geo_schemas.H3GridWeights` table; every distinct
    `h3_index` in that table is a cell the run must cover."""

    valid_time_min: datetime | None
    """Earliest `valid_time`, or `None` for an empty frame."""

    valid_time_max: datetime | None
    """Latest `valid_time`, or `None` for an empty frame."""

    missing_ensemble_members: tuple[int, ...]
    """Expected members with no rows at all, sorted."""

    unexpected_ensemble_members: tuple[int, ...]
    """Observed members outside the expected set, sorted — the upstream ensemble changed shape."""

    missing_lead_time_hours: tuple[int, ...]
    """Expected forecast steps (hours after `init_time`) with no rows at all, sorted."""

    unexpected_valid_times: tuple[datetime, ...]
    """Observed `valid_time`s that are not on any expected forecast step, sorted."""

    @property
    def is_complete(self) -> bool:
        """True when the run carries the shape we expect.

        That is, when the frame is one run whose member set, forecast-step set, H3 cell *count*,
        and row count all match the expectation.

        Cells are compared by count, not by set: the report never receives the expected
        `h3_index` values, only how many there should be. Substituting one cell for another would
        therefore pass. The substitution is not a live gap, because the asset derives the
        expected count from the very H3 grid weights that
        `dynamical_data.ecmwf_ens.convert_to_polars`, the step that turns the downloaded weather
        into `Nwp` rows, joins against — see
        <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
        """
        return (
            len(self.init_times) == 1
            and not self.missing_ensemble_members
            and not self.unexpected_ensemble_members
            and not self.missing_lead_time_hours
            and not self.unexpected_valid_times
            and self.n_h3_cells == self.expected_n_h3_cells
            and self.n_rows == self.expected_n_rows
        )

    @property
    def h3_cell_shortfall(self) -> int:
        """Expected H3 cells minus observed.

        The difference is *net*, so it is negative if the run carries extra cells and zero if one
        cell was swapped for another.
        """
        return self.expected_n_h3_cells - self.n_h3_cells

    def describe(self) -> str:
        """One human-readable sentence naming every gap, for the asset check's description."""
        gaps = list(self._describe_gaps())
        if not gaps:
            return (
                f"Complete run: {self.n_ensemble_members} ensemble members x "
                f"{self.n_valid_times} forecast steps x {self.n_h3_cells} H3 cells "
                f"= {self.n_rows} rows."
            )
        return "Incomplete NWP run — " + "; ".join(gaps) + "."

    def _describe_gaps(self) -> Iterator[str]:
        """Yield one clause per detected gap, in decreasing order of diagnostic value."""
        if len(self.init_times) != 1:
            yield f"expected exactly one init_time, found {len(self.init_times)}"
        if self.missing_ensemble_members:
            yield f"missing ensemble member(s) {_abbreviate(self.missing_ensemble_members)}"
        if self.unexpected_ensemble_members:
            yield f"unexpected ensemble member(s) {_abbreviate(self.unexpected_ensemble_members)}"
        if self.missing_lead_time_hours:
            yield f"missing lead time(s) in hours {_abbreviate(self.missing_lead_time_hours)}"
        if self.unexpected_valid_times:
            yield f"unexpected valid_time(s) {_abbreviate(self.unexpected_valid_times)}"
        if self.n_h3_cells != self.expected_n_h3_cells:
            yield f"{self.n_h3_cells} H3 cells, expected {self.expected_n_h3_cells}"
        if self.n_rows != self.expected_n_rows:
            yield f"{self.n_rows} rows, expected {self.expected_n_rows}"
Attributes
init_times instance-attribute

Distinct init_time values in the frame. A whole run has exactly one.

n_rows instance-attribute

Rows in the frame.

expected_n_rows instance-attribute

len(expected_ensemble_members) * len(expected_lead_time_hours) * expected_n_h3_cells — the size of the complete grid.

n_ensemble_members instance-attribute

Distinct ensemble_member values observed.

n_valid_times instance-attribute

Distinct valid_time values observed.

n_h3_cells instance-attribute

Distinct h3_index values observed.

expected_n_h3_cells instance-attribute

Distinct h3_index values the H3 grid weights say this run should cover.

The H3 grid weights are the contracts.geo_schemas.H3GridWeights table; every distinct h3_index in that table is a cell the run must cover.

valid_time_min instance-attribute

Earliest valid_time, or None for an empty frame.

valid_time_max instance-attribute

Latest valid_time, or None for an empty frame.

missing_ensemble_members instance-attribute

Expected members with no rows at all, sorted.

unexpected_ensemble_members instance-attribute

Observed members outside the expected set, sorted — the upstream ensemble changed shape.

missing_lead_time_hours instance-attribute

Expected forecast steps (hours after init_time) with no rows at all, sorted.

unexpected_valid_times instance-attribute

Observed valid_times that are not on any expected forecast step, sorted.

is_complete property

True when the run carries the shape we expect.

That is, when the frame is one run whose member set, forecast-step set, H3 cell count, and row count all match the expectation.

Cells are compared by count, not by set: the report never receives the expected h3_index values, only how many there should be. Substituting one cell for another would therefore pass. The substitution is not a live gap, because the asset derives the expected count from the very H3 grid weights that dynamical_data.ecmwf_ens.convert_to_polars, the step that turns the downloaded weather into Nwp rows, joins against — see https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

h3_cell_shortfall property

Expected H3 cells minus observed.

The difference is net, so it is negative if the run carries extra cells and zero if one cell was swapped for another.

Methods:
__init__(init_times, n_rows, expected_n_rows, n_ensemble_members, n_valid_times, n_h3_cells, expected_n_h3_cells, valid_time_min, valid_time_max, missing_ensemble_members, unexpected_ensemble_members, missing_lead_time_hours, unexpected_valid_times)
describe()

One human-readable sentence naming every gap, for the asset check's description.

Source code in packages/contracts/src/contracts/weather_schemas.py 
773
774
775
776
777
778
779
780
781
782
def describe(self) -> str:
    """One human-readable sentence naming every gap, for the asset check's description."""
    gaps = list(self._describe_gaps())
    if not gaps:
        return (
            f"Complete run: {self.n_ensemble_members} ensemble members x "
            f"{self.n_valid_times} forecast steps x {self.n_h3_cells} H3 cells "
            f"= {self.n_rows} rows."
        )
    return "Incomplete NWP run — " + "; ".join(gaps) + "."

Functions:

assess_nwp_quality(dataframe)

Summarise the tolerated-but-noteworthy nulls in a validated NWP run.

Reports the nulls in the de-accumulated variables (precipitation/radiation) beyond lead-0 that Nwp.validate deliberately tolerates: whole (ensemble_member, valid_time) slices that arrived empty, and the cells where the upstream per-pixel corruption survived the H3 aggregation by taking out every grid point of a cell. Pure and Dagster-free (unit-testable in isolation); the ecmwf_ens asset wraps the result into a WARN AssetCheckResult. See https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

Source code in packages/contracts/src/contracts/weather_schemas.py 
667
668
669
670
671
672
673
674
675
676
677
def assess_nwp_quality(dataframe: pt.DataFrame[Nwp]) -> NwpQualityReport:
    """Summarise the tolerated-but-noteworthy nulls in a *validated* NWP run.

    Reports the nulls in the de-accumulated variables (precipitation/radiation) beyond lead-0
    that `Nwp.validate` deliberately tolerates: whole (ensemble_member, valid_time) slices that
    arrived empty, and the cells where the upstream per-pixel corruption survived the H3
    aggregation by taking out every grid point of a cell. Pure and Dagster-free (unit-testable in
    isolation); the ``ecmwf_ens`` asset wraps the result into a WARN ``AssetCheckResult``. See
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """
    return NwpQualityReport(affected=_deaccumulated_null_breakdown(dataframe))

assess_nwp_run_completeness(dataframe, expected_n_h3_cells, expected_ensemble_members=ECMWF_ENS_ENSEMBLE_MEMBERS, expected_lead_time_hours=ECMWF_ENS_LEAD_TIME_HOURS)

Summarise whether one ingested NWP run covers its full (member x step x cell) grid.

The ecmwf_ens asset calls this function. Nwp.validate deliberately does not, because validation runs on arbitrary frames: filtered test fixtures, pruned scans, and single-member training reads. Completeness is a property of one whole ingested run, so it would be false on every one of those frames. Pure and Dagster-free, so it is unit-testable in isolation.

Never raises on a validated Nwp frame — not on an empty one, not on a multi-init_time one, not on one whose valid_times are off-grid — so it cannot turn the warning path into a failure path (rule 7 of Inherent Stability). The "validated" qualifier matters. Nwp.validate guarantees the key columns are present and non-null, which is exactly what this function relies on. The production caller, the ecmwf_ens asset, only ever passes a frame that dynamical_data.ecmwf_ens.convert_to_polars has already validated.

Row counting is safe here despite Polars' 32-bit row index. This function runs on a single in-memory run: at V1 scale, 1671 cells x 51 members x 85 steps ~ 7.24M rows. That row count is nearly three orders of magnitude below the 2**32 ceiling, which the whole ~5.9B-row NWP Delta table would hit.

Parameters:

Name Type Description Default
dataframe DataFrame[Nwp]

One whole ingested NWP run, already through Nwp.validate.

 required
expected_n_h3_cells int

Distinct H3 cells the run should cover — pass h3_grid["h3_index"].n_unique() so the expectation tracks the H3 grid weights rather than a hard-coded number.

 required
expected_ensemble_members frozenset[int]

Members the run should carry. Defaults to the ECMWF ENS ensemble, the only NwpModelId we ingest today.

 ECMWF_ENS_ENSEMBLE_MEMBERS
expected_lead_time_hours tuple[int, ...]

Forecast steps, in hours after init_time, the run should carry. Defaults to the ECMWF ENS native step structure.

 ECMWF_ENS_LEAD_TIME_HOURS

Returns:

Type Description
NwpRunCompletenessReport

A NwpRunCompletenessReport. The report gives the observed row count against the
NwpRunCompletenessReport

expected complete-grid row count, the observed versus expected ensemble members, and the
NwpRunCompletenessReport

observed versus expected H3 cell count. The report also gives the earliest and latest
NwpRunCompletenessReport

valid_time, both None for an empty frame. Finally, missing_lead_time_hours lists the
NwpRunCompletenessReport

expected lead times that are missing and unexpected_valid_times lists the observed
NwpRunCompletenessReport

valid_times that are unexpected. Both lists are empty when the frame does not hold
NwpRunCompletenessReport

exactly one init_time.
Source code in packages/contracts/src/contracts/weather_schemas.py 
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
def assess_nwp_run_completeness(
    dataframe: pt.DataFrame[Nwp],
    expected_n_h3_cells: int,
    expected_ensemble_members: frozenset[int] = ECMWF_ENS_ENSEMBLE_MEMBERS,
    expected_lead_time_hours: tuple[int, ...] = ECMWF_ENS_LEAD_TIME_HOURS,
) -> NwpRunCompletenessReport:
    """Summarise whether one ingested NWP run covers its full (member x step x cell) grid.

    The `ecmwf_ens` asset calls this function. `Nwp.validate` deliberately does **not**, because
    validation runs on arbitrary frames: filtered test fixtures, pruned scans, and single-member
    training reads. Completeness is a property of one whole ingested run, so it would be false on
    every one of those frames. Pure and Dagster-free, so it is unit-testable in isolation.

    Never raises on a validated `Nwp` frame — not on an empty one, not on a multi-`init_time` one,
    not on one whose `valid_time`s are off-grid — so it cannot turn the warning path into a failure
    path (rule 7 of [Inherent
    Stability](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/#the-rules)).
    The "validated" qualifier matters. `Nwp.validate` guarantees the key columns are present and
    non-null, which is exactly what this function relies on. The production caller, the `ecmwf_ens`
    asset, only ever passes a frame that `dynamical_data.ecmwf_ens.convert_to_polars` has already
    validated.

    Row counting is safe here despite Polars' 32-bit row index. This function runs on a single
    in-memory run: at V1 scale, 1671 cells x 51 members x 85 steps ~ 7.24M rows. That row count is
    nearly three orders of magnitude below the 2**32 ceiling, which the whole ~5.9B-row NWP Delta
    table would hit.

    Args:
        dataframe: One whole ingested NWP run, already through `Nwp.validate`.
        expected_n_h3_cells: Distinct H3 cells the run should cover — pass
            `h3_grid["h3_index"].n_unique()` so the expectation tracks the H3 grid weights rather
            than a hard-coded number.
        expected_ensemble_members: Members the run should carry. Defaults to the ECMWF ENS
            ensemble, the only `NwpModelId` we ingest today.
        expected_lead_time_hours: Forecast steps, in hours after `init_time`, the run should carry.
            Defaults to the ECMWF ENS native step structure.

    Returns:
        A `NwpRunCompletenessReport`. The report gives the observed row count against the
        expected complete-grid row count, the observed versus expected ensemble members, and the
        observed versus expected H3 cell count. The report also gives the earliest and latest
        `valid_time`, both `None` for an empty frame. Finally, `missing_lead_time_hours` lists the
        expected lead times that are missing and `unexpected_valid_times` lists the observed
        `valid_time`s that are unexpected. Both lists are empty when the frame does not hold
        exactly one `init_time`.
    """
    init_times = tuple(sorted(dataframe["init_time"].unique().to_list()))
    observed_members = set(dataframe["ensemble_member"].unique().to_list())
    observed_valid_times = set(dataframe["valid_time"].unique().to_list())
    missing_lead_times, unexpected_valid_times = _lead_time_gaps(
        init_times, observed_valid_times, expected_lead_time_hours
    )
    return NwpRunCompletenessReport(
        init_times=init_times,
        n_rows=dataframe.height,
        expected_n_rows=(
            len(expected_ensemble_members) * len(expected_lead_time_hours) * expected_n_h3_cells
        ),
        n_ensemble_members=len(observed_members),
        n_valid_times=len(observed_valid_times),
        n_h3_cells=dataframe["h3_index"].n_unique(),
        expected_n_h3_cells=expected_n_h3_cells,
        valid_time_min=min(observed_valid_times, default=None),
        valid_time_max=max(observed_valid_times, default=None),
        missing_ensemble_members=tuple(sorted(expected_ensemble_members - observed_members)),
        unexpected_ensemble_members=tuple(sorted(observed_members - expected_ensemble_members)),
        missing_lead_time_hours=missing_lead_times,
        unexpected_valid_times=unexpected_valid_times,
    )

contracts.geo_schemas

Spatial contracts: the weights mapping each H3 cell onto the NWP grid.

Those weights drive the spatial aggregation of numerical weather prediction (NWP) data.

Classes

H3GridWeights

Bases: Model

Schema for the pre-computed H3 grid weights.

This contract defines the mapping between H3 hexagons and a regular latitude/longitude grid. The contract ensures type safety when spatial mapping data passes from generic geospatial utilities (like packages/geo) to dataset-specific ingestion pipelines (like packages/dynamical_data).

Source code in packages/contracts/src/contracts/geo_schemas.py 
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
class H3GridWeights(pt.Model):
    """Schema for the pre-computed H3 grid weights.

    This contract defines the mapping between H3 hexagons and a regular latitude/longitude grid.
    The contract ensures type safety when spatial mapping data passes from generic geospatial
    utilities (like `packages/geo`) to dataset-specific ingestion pipelines (like
    `packages/dynamical_data`).
    """

    h3_index: int = pt.Field(dtype=pl.UInt64, description="H3 cell index.")

    nwp_lat: float = pt.Field(
        dtype=pl.Float32,
        ge=-90,
        le=90,
        description="Latitude of the NWP grid box centre (decimal degrees).",
    )

    nwp_lon: float = pt.Field(
        dtype=pl.Float32,
        ge=-180,
        le=180,
        description="Longitude of the NWP grid box centre (decimal degrees).",
    )

    proportion: float = pt.Field(
        dtype=pl.Float32,
        ge=0,
        le=1,
        description="Fraction of the H3 hexagon that overlaps this NWP grid box (0 to 1).",
    )
Attributes
h3_index = pt.Field(dtype=pl.UInt64, description='H3 cell index.') class-attribute instance-attribute
nwp_lat = pt.Field(dtype=pl.Float32, ge=-90, le=90, description='Latitude of the NWP grid box centre (decimal degrees).') class-attribute instance-attribute
nwp_lon = pt.Field(dtype=pl.Float32, ge=-180, le=180, description='Longitude of the NWP grid box centre (decimal degrees).') class-attribute instance-attribute
proportion = pt.Field(dtype=pl.Float32, ge=0, le=1, description='Fraction of the H3 hexagon that overlaps this NWP grid box (0 to 1).') class-attribute instance-attribute

contracts.ml_schemas

Contracts for the machine-learning (ML) pipeline.

The feature names a model config may request, the joined AllFeatures frame handed to models, the eligible-time-series population, and the metrics schema.

Attributes

TimeFeature = Literal['local_time_of_day_sin', 'local_time_of_day_cos', 'local_time_of_year_sin', 'local_time_of_year_cos', 'local_day_of_week_sin', 'local_day_of_week_cos', 'local_day_of_week', 'local_utc_offset_minutes'] module-attribute

SafeInputBaseColumn = Literal['time_series_id', 'time_series_type', 'nwp_lead_time_hours', 'ensemble_member', 'power_fcst_init_time', 'nwp_init_time'] module-attribute

HORIZON_SLICES = ('all', 'intraday', 'day_ahead', 'short_medium_range', 'extended_range') module-attribute

Horizon slice labels: the four forecast ranges from the project report, plus "all" for the aggregate across every horizon.

Bands are left-closed intervals of lead time (valid_time − power_fcst_init_time), as implemented by ml_core.metrics:

  • "all": aggregates over all horizons and is always computed
  • "intraday": [0 h, 6 h)
  • "day_ahead": [6 h, 36 h)
  • "short_medium_range": [36 h, 168 h) — day 2 to day 7
  • "extended_range": [168 h, ∞) — day 8 onwards (unbounded above; in practice forecasts stop at the 14-day NWP horizon)

METRIC_NAMES = ('mae', 'nmae', 'rmse', 'mbe', 'crps', 'spread_skill_ratio', 'pinball_loss', 'mean_pinball_loss', 'picp', 'interval_width') module-attribute

Metric names currently implemented.

Full definitions, equations, and design rationale: https://openclimatefix.github.io/nged-substation-forecast/techniques/evaluation-metrics/

Deterministic (scored on the per-run ensemble mean):

  • "mae": mean absolute error (MW)
  • "nmae": normalised MAE (dimensionless; normalised by full-history effective capacity)
  • "rmse": root mean squared error (MW)
  • "mbe": mean bias error (MW; positive = over-prediction)

Probabilistic (scored on the ensemble members before the mean collapse):

  • "crps": fair (finite-ensemble-unbiased) continuous ranked probability score (MW). The fair form removes the finite-ensemble bias, so scores stay comparable across models with different ensemble sizes.
  • "spread_skill_ratio": Fortin-corrected root-mean-square (RMS) ensemble spread ÷ RMSE of the ensemble mean (dimensionless; 1.0 = well-calibrated, < 1 = underdispersed/overconfident).
  • "pinball_loss": quantile loss (MW) at the quantile named by metric_param.
  • "mean_pinball_loss": unweighted mean of "pinball_loss" over the 13 DELIVERY_QUANTILES (MW). Tail-heavy by construction, matching NGED's priorities.
  • "picp": prediction-interval coverage probability of the band named by metric_param (dimensionless fraction). The calibrated reference for empirical quantiles from a finite ensemble sits below the nominal coverage — e.g. ≈ 0.769, not 0.8, for p10_p90 at 51 members.
  • "interval_width": mean width (MW) of the band named by metric_param — the sharpness companion to "picp" (coverage is only impressive if the band is also narrow).

QUANTILE_METRIC_PARAMS = tuple(quantile_label(q) for q in DELIVERY_QUANTILES) module-attribute

metric_param labels for quantile-indexed metrics (pinball loss), e.g. "p1" … "p99".

Derived from DELIVERY_QUANTILES so that evaluation is always scored at exactly the quantile levels the product delivers.

BAND_METRIC_PARAMS = tuple(f'{quantile_label(q)}_{quantile_label(1 - q)}' for q in DELIVERY_QUANTILES if q < 0.5) module-attribute

metric_param labels for the symmetric prediction-interval bands (PICP, interval width).

One band per symmetric pair of DELIVERY_QUANTILES: "p1_p99", "p2_p98", "p5_p95", "p10_p90", "p20_p80", "p35_p65". Together the six bands trace a coverage curve from the 30% band to the 98% band.

METRIC_PARAMS = ('all', *QUANTILE_METRIC_PARAMS, *BAND_METRIC_PARAMS) module-attribute

Parameter values for parametric metrics.

  • "all": all scalar metrics with no extra parameter dimension (MAE, NMAE, RMSE, MBE, CRPS, spread-skill ratio, mean pinball loss).
  • Quantile labels ("p1" … "p99", from QUANTILE_METRIC_PARAMS): pinball loss.
  • Band labels ("p1_p99" … "p35_p65", from BAND_METRIC_PARAMS): PICP and interval width.

EVALUATION_SCOPES = ('leaderboard', 'production_monitoring', 'ad_hoc') module-attribute

Evaluation scopes that coexist in the forecast_metrics table.

  • "leaderboard": CV-fold leaderboard metrics
  • "production_monitoring": live trailing-window monitoring
  • "ad_hoc": one-off analyses (no MLflow run)

EvalScopeType = Literal['leaderboard', 'ad_hoc'] module-attribute

Type annotation for the evaluation scopes currently accepted by the metrics asset config.

Distinct from EVALUATION_SCOPES (the runtime tuple driving the Polars Enum): that tuple includes "production_monitoring" for future use; this Literal reflects only the scopes the asset actually handles today. Expand to match EVALUATION_SCOPES when Phase 8 lands.

TIME_SERIES_TYPE_SLICES = ('all', *LIST_OF_TIME_SERIES_TYPES) module-attribute

Values for the time_series_type metric slice.

Every time_series_type plus the sentinel "all" for the across-everything aggregate.

Classes

AllFeatures

Bases: Model

Final joined dataset ready for the ML model.

Weather features are kept in their physical units (e.g., degrees Celsius, m/s) to ensure precision during interpolation and feature engineering.

DYNAMIC FEATURES: In addition to the explicitly defined columns below, the pipeline supports dynamically generated features. You can request these dynamic features in your model config:

  • power_lag_{hours}h: The power value shifted by X hours (e.g., power_lag_24h).
  • temperature_2m_rolling_mean_{hours}h: Rolling average of temperature over X hours (e.g., temperature_2m_rolling_mean_6h).

Dynamic features are not explicitly typed as Patito fields below. Omitting the dynamic features from the field list is intentional, to allow infinite parameterisation (e.g., any lag hour). The two alternatives both cost more: generating the field definitions in code at import time (metaprogramming), or defining hundreds of static fields by hand. During feature engineering, the pipeline asserts that every requested dynamic feature is present.

Source code in packages/contracts/src/contracts/ml_schemas.py 
 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
class AllFeatures(pt.Model):
    """Final joined dataset ready for the ML model.

    Weather features are kept in their physical units (e.g., degrees Celsius, m/s) to ensure
    precision during interpolation and feature engineering.

    DYNAMIC FEATURES: In addition to the explicitly defined columns below, the pipeline supports
    dynamically generated features. You can request these dynamic features in your model config:

    * `power_lag_{hours}h`: The power value shifted by X hours (e.g., `power_lag_24h`).
    * `temperature_2m_rolling_mean_{hours}h`: Rolling average of temperature over X hours (e.g.,
      `temperature_2m_rolling_mean_6h`).

    Dynamic features are not explicitly typed as Patito fields below. Omitting the dynamic
    features from the field list is intentional, to allow infinite parameterisation (e.g., any
    lag hour). The two alternatives both cost more: generating the field definitions in code at
    import time (metaprogramming), or defining hundreds of static fields by hand. During feature
    engineering, the pipeline asserts that every requested dynamic feature is present.
    """

    valid_time: datetime = pt.Field(dtype=UTC_DATETIME_DTYPE)
    time_series_id: int = _get_time_series_id_dtype()
    time_series_type: str = pt.Field(
        dtype=pl.Enum(LIST_OF_TIME_SERIES_TYPES),
        allow_missing=True,
        description=(
            "The substation's category. Only emitted when a feature set requests it, and never "
            "null: the feature pipeline drops any time series with no row in the metadata."
        ),
    )
    ensemble_member: int | None = pt.Field(dtype=pl.Int8, allow_missing=True)

    power_fcst_init_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "When OCF's power forecast model was initialised. This might be called `t0` in other "
            "OCF projects."
        ),
    )

    nwp_init_time: datetime | None = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description="When the NWP run was initialised.",
    )

    power: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "The observed power, or null where there is no observation. Live inference feeds a "
            "dense spine that runs past the last observation, and training drops the null rows."
        ),
    )
    nwp_lead_time_hours: float | None = pt.Field(dtype=pl.Float32, allow_missing=True)

    # Weather features
    temperature_2m: float | None = _FEATURE_DTYPE
    dew_point_temperature_2m: float | None = _FEATURE_DTYPE
    wind_speed_10m: float | None = _FEATURE_DTYPE
    wind_direction_10m: float | None = _FEATURE_DTYPE
    wind_speed_100m: float | None = _FEATURE_DTYPE
    wind_direction_100m: float | None = _FEATURE_DTYPE
    pressure_surface: float | None = _FEATURE_DTYPE
    pressure_reduced_to_mean_sea_level: float | None = _FEATURE_DTYPE
    geopotential_height_500hpa: float | None = _FEATURE_DTYPE
    downward_long_wave_radiation_flux_surface: float | None = _FEATURE_DTYPE
    downward_short_wave_radiation_flux_surface: float | None = _FEATURE_DTYPE
    precipitation_surface: float | None = _FEATURE_DTYPE
    categorical_precipitation_type_surface: int | None = _FEATURE_DTYPE

    # Derived weather features
    windchill: float | None = _FEATURE_DTYPE

    # Temporal features. `local` means "in the local timezone", e.g. "Europe/London". We use `local`
    # as the main input feature, because it's the local time that mostly drives demand.
    local_utc_offset_minutes: int | None = pt.Field(
        dtype=pl.Int16,
        allow_missing=True,
        description=(
            "Offset of the local time zone from UTC, in minutes (0 or 60 for GB). Minutes keep "
            "sub-hour zones distinct: India's +5:30 is 330 and Nepal's +5:45 is 345. Int16 spans "
            "the real extremes, -720 (Etc/GMT+12) to +840 (Pacific/Kiritimati)."
        ),
    )
    local_time_of_day_sin: float | None = _FEATURE_DTYPE
    local_time_of_day_cos: float | None = _FEATURE_DTYPE
    local_time_of_year_sin: float | None = _FEATURE_DTYPE
    local_time_of_year_cos: float | None = _FEATURE_DTYPE
    local_day_of_week_sin: float | None = _FEATURE_DTYPE
    local_day_of_week_cos: float | None = _FEATURE_DTYPE
    local_day_of_week: str | None = pt.Field(
        dtype=pl.Enum(
            ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
        )
    )
Attributes
valid_time = pt.Field(dtype=UTC_DATETIME_DTYPE) class-attribute instance-attribute
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
time_series_type = pt.Field(dtype=pl.Enum(LIST_OF_TIME_SERIES_TYPES), allow_missing=True, description="The substation's category. Only emitted when a feature set requests it, and never null: the feature pipeline drops any time series with no row in the metadata.") class-attribute instance-attribute
ensemble_member = pt.Field(dtype=pl.Int8, allow_missing=True) class-attribute instance-attribute
power_fcst_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description="When OCF's power forecast model was initialised. This might be called `t0` in other OCF projects.") class-attribute instance-attribute
nwp_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='When the NWP run was initialised.') class-attribute instance-attribute
power = pt.Field(dtype=pl.Float32, description='The observed power, or null where there is no observation. Live inference feeds a dense spine that runs past the last observation, and training drops the null rows.') class-attribute instance-attribute
nwp_lead_time_hours = pt.Field(dtype=pl.Float32, allow_missing=True) class-attribute instance-attribute
temperature_2m = _FEATURE_DTYPE class-attribute instance-attribute
dew_point_temperature_2m = _FEATURE_DTYPE class-attribute instance-attribute
wind_speed_10m = _FEATURE_DTYPE class-attribute instance-attribute
wind_direction_10m = _FEATURE_DTYPE class-attribute instance-attribute
wind_speed_100m = _FEATURE_DTYPE class-attribute instance-attribute
wind_direction_100m = _FEATURE_DTYPE class-attribute instance-attribute
pressure_surface = _FEATURE_DTYPE class-attribute instance-attribute
pressure_reduced_to_mean_sea_level = _FEATURE_DTYPE class-attribute instance-attribute
geopotential_height_500hpa = _FEATURE_DTYPE class-attribute instance-attribute
downward_long_wave_radiation_flux_surface = _FEATURE_DTYPE class-attribute instance-attribute
downward_short_wave_radiation_flux_surface = _FEATURE_DTYPE class-attribute instance-attribute
precipitation_surface = _FEATURE_DTYPE class-attribute instance-attribute
categorical_precipitation_type_surface = _FEATURE_DTYPE class-attribute instance-attribute
windchill = _FEATURE_DTYPE class-attribute instance-attribute
local_utc_offset_minutes = pt.Field(dtype=pl.Int16, allow_missing=True, description="Offset of the local time zone from UTC, in minutes (0 or 60 for GB). Minutes keep sub-hour zones distinct: India's +5:30 is 330 and Nepal's +5:45 is 345. Int16 spans the real extremes, -720 (Etc/GMT+12) to +840 (Pacific/Kiritimati).") class-attribute instance-attribute
local_time_of_day_sin = _FEATURE_DTYPE class-attribute instance-attribute
local_time_of_day_cos = _FEATURE_DTYPE class-attribute instance-attribute
local_time_of_year_sin = _FEATURE_DTYPE class-attribute instance-attribute
local_time_of_year_cos = _FEATURE_DTYPE class-attribute instance-attribute
local_day_of_week_sin = _FEATURE_DTYPE class-attribute instance-attribute
local_day_of_week_cos = _FEATURE_DTYPE class-attribute instance-attribute
local_day_of_week = pt.Field(dtype=pl.Enum(['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'])) class-attribute instance-attribute

Metrics

Bases: Model

Evaluation metrics for power forecasts, in tall format.

Tall format means one row per metric value, rather than one column per metric.

metric_param encodes the extra parameter dimension for metrics that have one, or "all" for scalar metrics with no extra dimension. Examples:

time_series_id fold_id horizon_slice metric_name metric_param metric_value
1 live all mae all 5.2
1 live day_ahead rmse all 7.1
1 live day_ahead pinball_loss p10 2.1
1 live day_ahead pinball_loss p50 3.4
1 live day_ahead mean_pinball_loss all 2.4
1 live day_ahead picp p10_p90 0.78

Primary key: (time_series_id, power_fcst_model_name, experiment_name, fold_id, evaluation_scope, horizon_slice, metric_name, metric_param, window_start, window_end). At most one metric value per series, model, experiment, fold, evaluation scope, horizon slice, metric, parameter, and window. Recomputing an existing key replaces that row rather than duplicating it.

Source code in packages/contracts/src/contracts/ml_schemas.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
class Metrics(pt.Model):
    """Evaluation metrics for power forecasts, in tall format.

    Tall format means one row per metric value, rather than one column per metric.

    `metric_param` encodes the extra parameter dimension for metrics that have one, or `"all"`
    for scalar metrics with no extra dimension. Examples:

    | time_series_id | fold_id | horizon_slice | metric_name       | metric_param | metric_value |
    |----------------|---------|---------------|-------------------|--------------|--------------|
    | 1              | live    | all           | mae               | all          | 5.2          |
    | 1              | live    | day_ahead     | rmse              | all          | 7.1          |
    | 1              | live    | day_ahead     | pinball_loss      | p10          | 2.1          |
    | 1              | live    | day_ahead     | pinball_loss      | p50          | 3.4          |
    | 1              | live    | day_ahead     | mean_pinball_loss | all          | 2.4          |
    | 1              | live    | day_ahead     | picp              | p10_p90      | 0.78         |

    Primary key: `(time_series_id, power_fcst_model_name, experiment_name, fold_id,
    evaluation_scope, horizon_slice, metric_name, metric_param, window_start, window_end)`. At
    most one metric value per series, model, experiment, fold, evaluation scope, horizon slice,
    metric, parameter, and window. Recomputing an existing key replaces that row rather than
    duplicating it.
    """

    time_series_id: int = _get_time_series_id_dtype()

    # String (not Categorical): fold_id/experiment_name are Delta partition columns, and delta-rs
    # stores dictionary-encoded columns as String anyway. String keeps those two columns cast-free
    # and lets predicate pushdown work. See the "Delta Lake dictionary-encoded columns" section of
    # the `polars-patito-gotchas` skill.
    power_fcst_model_name: str = pt.Field(
        dtype=pl.String,
        description="Identifier for the ML-based power forecasting model family.",
    )

    fold_id: str = pt.Field(
        dtype=pl.String,
        description=(
            "The CV fold's label from conf/cv/default.yaml (e.g. 'mid_2025_to_mid_2026'), or "
            "'live' for production forecasts. Matches PowerForecast.fold_id."
        ),
    )

    horizon_slice: str = pt.Field(
        dtype=pl.Enum(HORIZON_SLICES),
        description=(
            "'all' aggregates over all forecast horizons.  Other values select the HORIZON_SLICES"
            " bands."
        ),
    )

    metric_name: str = pt.Field(
        dtype=pl.Enum(METRIC_NAMES),
        description="The name of the metric (e.g. 'mae', 'rmse').",
    )

    metric_param: str = pt.Field(
        dtype=pl.Enum(METRIC_PARAMS),
        description=(
            "Extra parameter dimension for parametric metrics.  "
            "'all' for scalar metrics (MAE, NMAE, RMSE, MBE, CRPS, spread-skill ratio, mean"
            " pinball loss).  For pinball loss: a DELIVERY_QUANTILES label ('p1' … 'p99').  "
            "For PICP and interval width: a symmetric band label ('p1_p99' … 'p35_p65')."
        ),
    )

    metric_value: float = pt.Field(dtype=pl.Float32, description="The computed metric value.")

    evaluation_scope: str = pt.Field(
        dtype=pl.Enum(EVALUATION_SCOPES),
        allow_missing=True,
        description=(
            "Which evaluation produced this row, so leaderboard, live-monitoring, and "
            "one-off metrics coexist in one table and stay separable."
        ),
    )
    """`evaluation_scope` and the columns below it are populated by the ``metrics`` Dagster asset
    through ``enrich_metrics_rows()``. Those columns are ``allow_missing`` so that the pure
    ``compute_metrics()`` helper can emit the core metric rows on its own. The asset then enriches
    those rows with scope/window provenance, before the frame is written to the
    ``forecast_metrics`` Delta table. The one exception is `time_series_type`:
    ``compute_metrics()`` populates `time_series_type` itself, by joining it on from the
    metadata."""

    time_series_type: str = pt.Field(
        dtype=pl.Enum(TIME_SERIES_TYPE_SLICES),
        allow_missing=True,
        description=(
            "The time-series category for this row. Never null: `compute_metrics` raises if any "
            "scored series has no metadata row, because a null would drop that series out of "
            "every per-type aggregate while still counting towards the overall mean."
        ),
    )

    window_start: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description=(
            "Inclusive start of the valid_time window this row covers. For a CV fold this "
            "is the fold's val_start; for monitoring it is the trailing-window start."
        ),
    )

    window_end: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description="End of the valid_time window this row covers (fold val_end or window end).",
    )

    window_label: str = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        description="Human label for the window, e.g. '2025', '24h', '7d', 'full_fold'.",
    )

    computed_at: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description=(
            "When this metric row was last computed. computed_at is pure provenance: PRIMARY_KEY "
            "(below), not computed_at, decides whether a write is a new row or a recomputation of "
            "an existing one. A recomputation replaces the existing row rather than being told "
            "apart from it by computed_at."
        ),
    )

    mlflow_run_id: str | None = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        description=(
            "Convenience cross-link to the MLflow run this metric row belongs to. "
            "Null for 'ad_hoc' rows, which have no MLflow run."
        ),
    )

    experiment_name: str = pt.Field(
        dtype=pl.String,  # String (not Categorical) — Delta partition column, see fold_id above.
        allow_missing=True,
        description=(
            "Experiment that produced these forecasts. Delta partition key alongside fold_id. "
            "Populated by the metrics Dagster asset after compute_metrics() returns; "
            "allow_missing so compute_metrics() itself need not produce it."
        ),
    )

    PRIMARY_KEY: ClassVar[tuple[str, ...]] = (
        "time_series_id",
        "power_fcst_model_name",
        "experiment_name",
        "fold_id",
        "evaluation_scope",
        "horizon_slice",
        "metric_name",
        "metric_param",
        "window_start",
        "window_end",
    )
    """See the class docstring's "Primary key" paragraph."""

    @classmethod
    def validate(  # ty: ignore[invalid-method-override]
        cls,
        dataframe: pl.DataFrame,
        columns: Sequence[str] | None = None,
        allow_missing_columns: bool = False,
        allow_superfluous_columns: bool = False,
        drop_superfluous_columns: bool = False,
    ) -> pt.DataFrame[Self]:
        """Validate the given dataframe, ensuring the primary key is unique where it is present.

        A duplicated primary key means either a join fanned out on the way here or the same rows
        were written twice. Both corrupt the leaderboard or monitoring chart built from them. A
        duplicated primary key is our own bug rather than the outside world misbehaving, so
        `validate` raises rather than degrading — see
        <https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/>.

        Unlike `PowerForecast`, four `PRIMARY_KEY` columns (`experiment_name`,
        `evaluation_scope`, `window_start`, `window_end`) are `allow_missing`:
        `compute_metrics()` validates its output before the `metrics` Dagster asset's
        `enrich_metrics_rows()` adds them. Whenever any of those four columns is absent, the
        uniqueness check below is skipped entirely, rather than run against a partial key. The
        check that matters runs inside `enrich_metrics_rows()`, on the fully-enriched frame that
        is actually written to `forecast_metrics`.
        """
        validated_df = super().validate(
            dataframe=dataframe,
            columns=columns,
            allow_missing_columns=allow_missing_columns,
            allow_superfluous_columns=allow_superfluous_columns,
            drop_superfluous_columns=drop_superfluous_columns,
        )

        pk_cols = list(cls.PRIMARY_KEY)
        if not set(pk_cols).issubset(validated_df.columns):
            return validated_df

        # `n_unique`, not `is_duplicated().any()`: the two expressions are equivalent here (every
        # primary-key column is non-nullable once present) but `is_duplicated` materialises a
        # per-row mask, costing ~5x the peak memory on a predict-sized frame.
        if validated_df.select(pk_cols).n_unique() != validated_df.height:
            raise ValueError(
                f"Duplicate entries found for primary key columns: {pk_cols}. "
                "Either an upstream join fanned out or these rows were written twice."
            )

        return validated_df
Attributes
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
power_fcst_model_name = pt.Field(dtype=pl.String, description='Identifier for the ML-based power forecasting model family.') class-attribute instance-attribute
fold_id = pt.Field(dtype=pl.String, description="The CV fold's label from conf/cv/default.yaml (e.g. 'mid_2025_to_mid_2026'), or 'live' for production forecasts. Matches PowerForecast.fold_id.") class-attribute instance-attribute
horizon_slice = pt.Field(dtype=pl.Enum(HORIZON_SLICES), description="'all' aggregates over all forecast horizons. Other values select the HORIZON_SLICES bands.") class-attribute instance-attribute
metric_name = pt.Field(dtype=pl.Enum(METRIC_NAMES), description="The name of the metric (e.g. 'mae', 'rmse').") class-attribute instance-attribute
metric_param = pt.Field(dtype=pl.Enum(METRIC_PARAMS), description="Extra parameter dimension for parametric metrics. 'all' for scalar metrics (MAE, NMAE, RMSE, MBE, CRPS, spread-skill ratio, mean pinball loss). For pinball loss: a DELIVERY_QUANTILES label ('p1' … 'p99'). For PICP and interval width: a symmetric band label ('p1_p99' … 'p35_p65').") class-attribute instance-attribute
metric_value = pt.Field(dtype=pl.Float32, description='The computed metric value.') class-attribute instance-attribute
evaluation_scope = pt.Field(dtype=pl.Enum(EVALUATION_SCOPES), allow_missing=True, description='Which evaluation produced this row, so leaderboard, live-monitoring, and one-off metrics coexist in one table and stay separable.') class-attribute instance-attribute

evaluation_scope and the columns below it are populated by the metrics Dagster asset through enrich_metrics_rows(). Those columns are allow_missing so that the pure compute_metrics() helper can emit the core metric rows on its own. The asset then enriches those rows with scope/window provenance, before the frame is written to the forecast_metrics Delta table. The one exception is time_series_type: compute_metrics() populates time_series_type itself, by joining it on from the metadata.

time_series_type = pt.Field(dtype=pl.Enum(TIME_SERIES_TYPE_SLICES), allow_missing=True, description='The time-series category for this row. Never null: `compute_metrics` raises if any scored series has no metadata row, because a null would drop that series out of every per-type aggregate while still counting towards the overall mean.') class-attribute instance-attribute
window_start = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description="Inclusive start of the valid_time window this row covers. For a CV fold this is the fold's val_start; for monitoring it is the trailing-window start.") class-attribute instance-attribute
window_end = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='End of the valid_time window this row covers (fold val_end or window end).') class-attribute instance-attribute
window_label = pt.Field(dtype=pl.String, allow_missing=True, description="Human label for the window, e.g. '2025', '24h', '7d', 'full_fold'.") class-attribute instance-attribute
computed_at = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='When this metric row was last computed. computed_at is pure provenance: PRIMARY_KEY (below), not computed_at, decides whether a write is a new row or a recomputation of an existing one. A recomputation replaces the existing row rather than being told apart from it by computed_at.') class-attribute instance-attribute
mlflow_run_id = pt.Field(dtype=pl.String, allow_missing=True, description="Convenience cross-link to the MLflow run this metric row belongs to. Null for 'ad_hoc' rows, which have no MLflow run.") class-attribute instance-attribute
experiment_name = pt.Field(dtype=pl.String, allow_missing=True, description='Experiment that produced these forecasts. Delta partition key alongside fold_id. Populated by the metrics Dagster asset after compute_metrics() returns; allow_missing so compute_metrics() itself need not produce it.') class-attribute instance-attribute
PRIMARY_KEY = ('time_series_id', 'power_fcst_model_name', 'experiment_name', 'fold_id', 'evaluation_scope', 'horizon_slice', 'metric_name', 'metric_param', 'window_start', 'window_end') class-attribute

See the class docstring's "Primary key" paragraph.

Methods:
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the given dataframe, ensuring the primary key is unique where it is present.

A duplicated primary key means either a join fanned out on the way here or the same rows were written twice. Both corrupt the leaderboard or monitoring chart built from them. A duplicated primary key is our own bug rather than the outside world misbehaving, so validate raises rather than degrading — see https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/.

Unlike PowerForecast, four PRIMARY_KEY columns (experiment_name, evaluation_scope, window_start, window_end) are allow_missing: compute_metrics() validates its output before the metrics Dagster asset's enrich_metrics_rows() adds them. Whenever any of those four columns is absent, the uniqueness check below is skipped entirely, rather than run against a partial key. The check that matters runs inside enrich_metrics_rows(), on the fully-enriched frame that is actually written to forecast_metrics.

Source code in packages/contracts/src/contracts/ml_schemas.py 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
@classmethod
def validate(  # ty: ignore[invalid-method-override]
    cls,
    dataframe: pl.DataFrame,
    columns: Sequence[str] | None = None,
    allow_missing_columns: bool = False,
    allow_superfluous_columns: bool = False,
    drop_superfluous_columns: bool = False,
) -> pt.DataFrame[Self]:
    """Validate the given dataframe, ensuring the primary key is unique where it is present.

    A duplicated primary key means either a join fanned out on the way here or the same rows
    were written twice. Both corrupt the leaderboard or monitoring chart built from them. A
    duplicated primary key is our own bug rather than the outside world misbehaving, so
    `validate` raises rather than degrading — see
    <https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/>.

    Unlike `PowerForecast`, four `PRIMARY_KEY` columns (`experiment_name`,
    `evaluation_scope`, `window_start`, `window_end`) are `allow_missing`:
    `compute_metrics()` validates its output before the `metrics` Dagster asset's
    `enrich_metrics_rows()` adds them. Whenever any of those four columns is absent, the
    uniqueness check below is skipped entirely, rather than run against a partial key. The
    check that matters runs inside `enrich_metrics_rows()`, on the fully-enriched frame that
    is actually written to `forecast_metrics`.
    """
    validated_df = super().validate(
        dataframe=dataframe,
        columns=columns,
        allow_missing_columns=allow_missing_columns,
        allow_superfluous_columns=allow_superfluous_columns,
        drop_superfluous_columns=drop_superfluous_columns,
    )

    pk_cols = list(cls.PRIMARY_KEY)
    if not set(pk_cols).issubset(validated_df.columns):
        return validated_df

    # `n_unique`, not `is_duplicated().any()`: the two expressions are equivalent here (every
    # primary-key column is non-nullable once present) but `is_duplicated` materialises a
    # per-row mask, costing ~5x the peak memory on a predict-sized frame.
    if validated_df.select(pk_cols).n_unique() != validated_df.height:
        raise ValueError(
            f"Duplicate entries found for primary key columns: {pk_cols}. "
            "Either an upstream join fanned out or these rows were written twice."
        )

    return validated_df

EligibleTimeSeries

Bases: Model

The canonical per-fold population of eligible time_series_ids.

Written by the eligible_time_series Dagster asset (one Delta partition per fold_id) and read by trained_cv_model. cv_power_forecasts does not read this table; it inherits the same population through the trained model's trained_time_series_ids. Eligibility depends on data coverage and the fold dates only, never on the model or experiment config. Every experiment therefore trains and scores a fold on the identical population, which is what makes leaderboard comparisons apples-to-apples.

One row per eligible (fold_id, time_series_id).

Source code in packages/contracts/src/contracts/ml_schemas.py 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
class EligibleTimeSeries(pt.Model):
    """The canonical per-fold population of eligible ``time_series_id``s.

    Written by the ``eligible_time_series`` Dagster asset (one Delta partition per ``fold_id``)
    and read by ``trained_cv_model``. ``cv_power_forecasts`` does not read this table; it
    inherits the same population through the trained model's ``trained_time_series_ids``.
    Eligibility depends on data coverage and the fold dates **only**, never on the model or
    experiment config. Every experiment therefore trains and scores a fold on the identical
    population, which is what makes leaderboard comparisons apples-to-apples.

    One row per eligible ``(fold_id, time_series_id)``.
    """

    fold_id: str = pt.Field(
        dtype=pl.String,
        description=(
            "The CV fold this eligibility row belongs to (e.g. 'mid_2025_to_mid_2026'); the Delta "
            "partition key."
        ),
    )
    time_series_id: int = _get_time_series_id_dtype()
Attributes
fold_id = pt.Field(dtype=pl.String, description="The CV fold this eligibility row belongs to (e.g. 'mid_2025_to_mid_2026'); the Delta partition key.") class-attribute instance-attribute
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute

Functions:

contracts.config_schemas

Configuration schemas, and the class-path round-trip behind a _target_ string.

A _target_ string is the key a YAML config file uses to name the Python class to build.

Attributes

Classes

CvFoldConfig

Bases: BaseModel

Configuration for a single expanding-window cross-validation (CV) fold.

leaderboard distinguishes two kinds of fold. The leaderboard folds are the apples-to-apples evaluation protocol, and they are epoch-pinned: their dates are fixed for the life of a leaderboard epoch, and changing them founds a new epoch. Optional dev folds, such as smoke_test, are non-leaderboard folds. A leaderboard=False fold runs through the identical pipeline but never feeds the leaderboard.

min_training_months overrides CvConfig.min_training_months for this fold alone (None falls back to the config-level value). A short dev fold sets it to its train length so eligibility does not demand the leaderboard's longer history.

Source code in packages/contracts/src/contracts/config_schemas.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class CvFoldConfig(BaseModel):
    """Configuration for a single expanding-window cross-validation (CV) fold.

    ``leaderboard`` distinguishes two kinds of fold. The leaderboard folds are the
    apples-to-apples evaluation protocol, and they are epoch-pinned: their dates are fixed for
    the life of a leaderboard epoch, and changing them founds a new epoch. Optional dev folds,
    such as ``smoke_test``, are non-leaderboard folds. A ``leaderboard=False`` fold runs through
    the identical pipeline but never feeds the leaderboard.

    ``min_training_months`` overrides ``CvConfig.min_training_months`` for this fold alone
    (``None`` falls back to the config-level value). A short dev fold sets it to its train length
    so eligibility does not demand the leaderboard's longer history.
    """

    fold_id: FoldId
    train_start: date
    train_end: date
    val_start: date
    val_end: date
    leaderboard: bool = True
    min_training_months: int | None = Field(default=None, ge=1)
Attributes
fold_id instance-attribute
train_start instance-attribute
train_end instance-attribute
val_start instance-attribute
val_end instance-attribute
leaderboard = True class-attribute instance-attribute
min_training_months = Field(default=None, ge=1) class-attribute instance-attribute

CvConfig

Bases: BaseModel

Configuration for expanding-window cross-validation.

Each fold trains on the history from its train_start to its train_end, and is scored on the val_start-to-val_end span that follows. The window expands because each later fold keeps the earlier folds' training history and adds more to it.

The folds list defines the evaluation protocol shared by all experiments on the leaderboard. All models must be evaluated against the same folds to ensure apples-to-apples comparison.

min_training_months controls which time series are eligible for each fold. A time series is included only if it has at least min_training_months months of data before val_start, and data through val_end.

Source code in packages/contracts/src/contracts/config_schemas.py 
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
class CvConfig(BaseModel):
    """Configuration for expanding-window cross-validation.

    Each fold trains on the history from its train_start to its train_end, and is scored on the
    val_start-to-val_end span that follows. The window expands because each later fold keeps the
    earlier folds' training history and adds more to it.

    The folds list defines the evaluation protocol shared by all experiments on the leaderboard.
    All models must be evaluated against the same folds to ensure apples-to-apples comparison.

    min_training_months controls which time series are eligible for each fold. A time series is
    included only if it has at least min_training_months months of data before val_start, and
    data through val_end.
    """

    folds: list[CvFoldConfig]
    min_training_months: int = Field(default=6, ge=1)

    @property
    def fold_ids(self) -> list[str]:
        """The fold ids in declaration order (e.g. ``["mid_2025_to_mid_2026", "smoke_test"]``).

        Used to build the ``cv_experiment_folds`` partitions and to expand experiment
        registration into per-fold partition keys — always read from config, never hard-coded.
        """
        return [fold.fold_id for fold in self.folds]

    @property
    def leaderboard_fold_ids(self) -> list[str]:
        """The fold ids of the leaderboard folds only, in declaration order.

        Non-leaderboard dev folds (e.g. ``smoke_test``) are excluded. Used to expand the
        ``full_cv`` / ``register_only`` run modes and to scope leaderboard metrics.
        """
        return [fold.fold_id for fold in self.folds if fold.leaderboard]

    def get_fold(self, fold_id: str) -> CvFoldConfig:
        """Return the fold with the given ``fold_id``.

        Args:
            fold_id: The fold identifier to look up (e.g. ``"mid_2025_to_mid_2026"``).

        Returns:
            The ``CvFoldConfig`` whose ``fold_id`` matches, carrying that fold's train and
            validation date ranges, its ``leaderboard`` flag, and its
            ``min_training_months`` override.

        Raises:
            KeyError: If no fold with that id exists in the config.
        """
        for fold in self.folds:
            if fold.fold_id == fold_id:
                return fold
        raise KeyError(f"No fold with fold_id={fold_id!r}; available folds: {self.fold_ids}")
Attributes
folds instance-attribute
min_training_months = Field(default=6, ge=1) class-attribute instance-attribute
fold_ids property

The fold ids in declaration order (e.g. ["mid_2025_to_mid_2026", "smoke_test"]).

Used to build the cv_experiment_folds partitions and to expand experiment registration into per-fold partition keys — always read from config, never hard-coded.

leaderboard_fold_ids property

The fold ids of the leaderboard folds only, in declaration order.

Non-leaderboard dev folds (e.g. smoke_test) are excluded. Used to expand the full_cv / register_only run modes and to scope leaderboard metrics.

Methods:
get_fold(fold_id)

Return the fold with the given fold_id.

Parameters:

Name Type Description Default
fold_id str

The fold identifier to look up (e.g. "mid_2025_to_mid_2026").

 required

Returns:

Type Description
CvFoldConfig

The CvFoldConfig whose fold_id matches, carrying that fold's train and
CvFoldConfig

validation date ranges, its leaderboard flag, and its
CvFoldConfig

min_training_months override.

Raises:

Type Description
KeyError

If no fold with that id exists in the config.
Source code in packages/contracts/src/contracts/config_schemas.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def get_fold(self, fold_id: str) -> CvFoldConfig:
    """Return the fold with the given ``fold_id``.

    Args:
        fold_id: The fold identifier to look up (e.g. ``"mid_2025_to_mid_2026"``).

    Returns:
        The ``CvFoldConfig`` whose ``fold_id`` matches, carrying that fold's train and
        validation date ranges, its ``leaderboard`` flag, and its
        ``min_training_months`` override.

    Raises:
        KeyError: If no fold with that id exists in the config.
    """
    for fold in self.folds:
        if fold.fold_id == fold_id:
            return fold
    raise KeyError(f"No fold with fold_id={fold_id!r}; available folds: {self.fold_ids}")

Functions:

class_target(obj)

Return the fully-qualified import path of a class (or of an instance's class).

The inverse of import_class. class_target and import_class together are how a config file, an MLflow experiment tag, and a saved model's meta.json all name a Python class: by a module.ClassName string. That string survives being written to disk and read back in another process.

Parameters:

Name Type Description Default
obj type | object

The class to name, or an instance whose class should be named.

 required

Returns:

Type Description
str

The module.ClassName path, e.g. "xgboost_forecaster.forecaster.XGBoostForecaster".

Raises:

Type Description
ValueError

obj's class is not defined at module level — it is nested inside another class or inside a function. import_class resolves a single attribute lookup on a module, so such a class has no path it could resolve. Failing here, where the class is defined, beats emitting a target string that only breaks when a later caller tries to load the class.
Source code in packages/contracts/src/contracts/config_schemas.py 
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
def class_target(obj: type | object) -> str:
    """Return the fully-qualified import path of a class (or of an instance's class).

    The inverse of ``import_class``. ``class_target`` and ``import_class`` together are how a
    config file, an MLflow experiment tag, and a saved model's ``meta.json`` all name a Python
    class: by a ``module.ClassName`` string. That string survives being written to disk and read
    back in another process.

    Args:
        obj: The class to name, or an instance whose class should be named.

    Returns:
        The ``module.ClassName`` path, e.g. ``"xgboost_forecaster.forecaster.XGBoostForecaster"``.

    Raises:
        ValueError: ``obj``'s class is not defined at module level — it is nested inside another
            class or inside a function. ``import_class`` resolves a single attribute lookup on a
            module, so such a class has no path it could resolve. Failing here, where the class is
            defined, beats emitting a target string that only breaks when a later caller tries to
            load the class.
    """
    cls = obj if isinstance(obj, type) else type(obj)
    if "." in cls.__qualname__:
        raise ValueError(
            f"{cls.__module__}.{cls.__qualname__} is not defined at module level (it is nested "
            "inside a class or a function), so it has no importable path. Move it to module level "
            "to use it as a _target_."
        )
    return f"{cls.__module__}.{cls.__qualname__}"

import_class(target)

Resolve a fully-qualified class path to the class object.

The inverse of class_target: imports the module and looks the class up on it.

Parameters:

Name Type Description Default
target str

A module.ClassName path, as produced by class_target.

 required

Returns:

Type Description
type

The class named by target.

Raises:

Type Description
ValueError

target names no module path, is relative, its module cannot be imported, the module has no such attribute, or the attribute is not a class. A module that fails on its own imports is reported the same way — the original ImportError is chained as __cause__. Any other failure of the module's body (a SyntaxError, say) propagates unchanged, so a broken module reports its own error rather than hiding behind "cannot import".
Source code in packages/contracts/src/contracts/config_schemas.py 
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
def import_class(target: str) -> type:
    """Resolve a fully-qualified class path to the class object.

    The inverse of ``class_target``: imports the module and looks the class up on it.

    Args:
        target: A ``module.ClassName`` path, as produced by ``class_target``.

    Returns:
        The class named by ``target``.

    Raises:
        ValueError: ``target`` names no module path, is relative, its module cannot be imported,
            the module has no such attribute, or the attribute is not a class. A module that
            fails on its *own* imports is reported the same way — the original ``ImportError``
            is chained as ``__cause__``. Any other failure of the module's body (a ``SyntaxError``,
            say) propagates unchanged, so a broken module reports its own error rather than
            hiding behind "cannot import".
    """
    module_path, _, class_name = target.rpartition(".")
    # A leading dot would make import_module read the target as a relative import and demand a
    # package to resolve that target against. A relative target is malformed, so reject the target
    # here rather than letting it surface as the TypeError import_module would raise.
    if not module_path or module_path.startswith("."):
        raise ValueError(f"{target!r} is not a fully-qualified class path (expected 'module.Cls').")
    try:
        module = importlib.import_module(module_path)
    except ImportError as error:
        raise ValueError(f"Cannot import module {module_path!r} of target {target!r}.") from error
    try:
        resolved = getattr(module, class_name)
    except AttributeError as error:
        raise ValueError(f"Module {module_path!r} has no attribute {class_name!r}.") from error
    if not isinstance(resolved, type):
        # TRY004 wants a TypeError, but the isinstance check is on what `target` *resolved to*,
        # not on `target` itself, which is a perfectly well-typed str. What is wrong is the target's
        # value. Every failure here is therefore a ValueError, and a caller needs to catch only
        # ValueError.
        raise ValueError(  # noqa: TRY004
            f"Target {target!r} resolved to {resolved!r}, which is not a class."
        )
    return resolved

load_cv_config(path)

Load and validate the cross-validation config from a YAML file.

The CV folds are the leaderboard's evaluation protocol and must be read from conf/cv/default.yaml (never hard-coded) so every experiment and asset shares one canonical definition.

Parameters:

Name Type Description Default
path Path

Path to the CV config YAML (e.g. conf/cv/default.yaml).

 required

Returns:

Type Description
CvConfig

The validated CvConfig.
Source code in packages/contracts/src/contracts/config_schemas.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def load_cv_config(path: Path) -> CvConfig:
    """Load and validate the cross-validation config from a YAML file.

    The CV folds are the leaderboard's evaluation protocol and must be read from
    ``conf/cv/default.yaml`` (never hard-coded) so every experiment and asset shares one
    canonical definition.

    Args:
        path: Path to the CV config YAML (e.g. ``conf/cv/default.yaml``).

    Returns:
        The validated ``CvConfig``.
    """
    with path.open(encoding="utf-8") as file:
        return CvConfig.model_validate(yaml.safe_load(file))

contracts.settings

The pipeline's environment-resolved configuration, and the cached accessor that reaches it.

Settings is the configuration class itself, and get_settings() is the cached accessor that keeps one shared instance. Prefer the accessor wherever a Settings would otherwise be built while a module is still being imported. Building a Settings is what reads .env and the environment. An import-time build therefore couples a pure schema module to the environment, and freezes the values before a test can change them. A function is unaffected if it builds its own Settings when it runs, and every Settings this repo's Dagster code builds is built inside a function body. PROJECT_ROOT resolves the workspace root. The .env file and the repo-relative path defaults are anchored to that root. Anchoring the defaults to a resolved root is what lets one set of defaults serve an editable install, a non-editable install, and the Docker image alike.

Attributes

url_adapter = TypeAdapter(AnyHttpUrl) module-attribute

PROJECT_ROOT = _find_project_root(Path(__file__)) module-attribute

The repo root — anchor for the repo-relative defaults below (conf/, data/, .env).

Resolved by _find_project_root; see its docstring for the per-install-mode behaviour and the caveat for wheels installed outside a workspace checkout.

Classes

Settings

Bases: BaseSettings

Every setting the pipeline reads from its environment.

The fields cover the data paths, the object-store credentials, the MLflow tracking URI, and the four Sentry settings: sentry_dsn, sentry_environment, sentry_traces_sample_rate, and sentry_monitor_forecasts.

Each field takes its value from an environment variable of the same name, from the workspace .env, or from the default declared here, in that order of precedence. The managed data-table paths default to "", a sentinel meaning "derive me". Pydantic runs the after-mode model validator _derive_unset_paths once, immediately after the Settings is built. That validator fills a path left unset from data_path_internal, data_path_delivery, or local_artifacts_path. A caller therefore never observes the sentinel and always reads a concrete path, while a path set explicitly keeps the value it was given. Every value is read when the Settings is built. A test that changes the environment afterwards therefore has to build a fresh Settings. If that test reads through the cached accessor, it must call get_settings.cache_clear() first.

Source code in packages/contracts/src/contracts/settings.py 
 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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
class Settings(BaseSettings):
    """Every setting the pipeline reads from its environment.

    The fields cover the data paths, the object-store credentials, the MLflow tracking URI, and
    the four Sentry settings: ``sentry_dsn``, ``sentry_environment``,
    ``sentry_traces_sample_rate``, and ``sentry_monitor_forecasts``.

    Each field takes its value from an environment variable of the same name, from the workspace
    ``.env``, or from the default declared here, in that order of precedence. The managed
    data-table paths default to ``""``, a sentinel meaning "derive me". Pydantic runs the
    ``after``-mode model validator ``_derive_unset_paths`` once, immediately after the
    ``Settings`` is built. That validator fills a path left unset from ``data_path_internal``,
    ``data_path_delivery``, or ``local_artifacts_path``. A caller therefore never observes the
    sentinel and always reads a concrete path, while a path set explicitly keeps the value it was
    given. Every value is read when the ``Settings`` is built. A test that changes the
    environment afterwards therefore has to build a fresh ``Settings``. If that test reads
    through the cached accessor, it must call ``get_settings.cache_clear()`` first.
    """

    mlflow_tracking_uri: str = Field(
        default="sqlite:///mlflow.db",
        description=(
            "MLflow tracking URI. Centralised here for environment-specific configuration"
            " (e.g., local SQLite for development, remote server for production)."
            " The application entrypoint should read this setting and set the"
            " MLFLOW_TRACKING_URI environment variable accordingly, which MLflow will"
            " automatically pick up."
        ),
    )

    # Credentials for NGED's *source* bucket, which lives in NGED's AWS account. Empty by default,
    # not required: `get_nged_s3_store` is their only consumer, and its only caller is the
    # `power_time_series_and_metadata` ingest asset. Making them required would mean every Delta
    # read, training run and dashboard could not build a `Settings` without third-party credentials
    # it never uses. `.env` is gitignored, so a fresh clone could not run the test suite at all.
    # That is exactly what "the whole system must be exercisable on one laptop" forbids:
    # https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/design-principles/#6-the-whole-system-must-be-exercisable-on-one-laptop
    #
    # Absence is therefore caught at the point of *use*, in `get_nged_s3_store`. Catching it there
    # confines a mis-wired secret to the one schedule that needs it. Inference — the scheduled run
    # that produces a forecast — reads our own Delta tables and a model baked into the deployment
    # image, so failing inference over an ingest credential would stop the forecast.
    nged_s3_bucket_url: str = Field(
        default="",
        description=(
            "NGED source S3 bucket URL. Typically stored in the .env file. Empty unless you need"
            " to ingest from NGED's bucket; `get_nged_s3_store` raises if it is."
        ),
    )
    nged_s3_bucket_access_key: str = Field(
        default="",
        description="Access key for the NGED source S3 bucket. Typically stored in the .env file.",
    )
    nged_s3_bucket_secret: str = Field(
        default="",
        description="Secret key for the NGED source S3 bucket. Typically stored in the .env file.",
    )

    def require_nged_source_credentials(self) -> None:
        """Raise ``ValueError`` unless all three NGED source-bucket credentials are set.

        Call this immediately before doing something that reads NGED's bucket, so the failure names
        the missing configuration instead of surfacing as an opaque auth error from ``obstore``.

        Do *not* call this method from process start-up or module import in order to fail a
        deployment fast. Inference needs none of these credentials, so a start-up call would stop
        the forecast over a missing ingest secret. See the [AWS
        runbook](https://openclimatefix.github.io/nged-substation-forecast/live_service/aws/#step-8-store-secrets-in-parameter-store).

        Raises:
            ValueError: Naming exactly which of the three environment variables are unset.
        """
        missing = [
            name
            for name, value in (
                ("NGED_S3_BUCKET_URL", self.nged_s3_bucket_url),
                ("NGED_S3_BUCKET_ACCESS_KEY", self.nged_s3_bucket_access_key),
                ("NGED_S3_BUCKET_SECRET", self.nged_s3_bucket_secret),
            )
            if not value
        ]
        if missing:
            raise ValueError(
                f"NGED source-bucket credentials are not set: {', '.join(missing)}. Reading"
                " NGED's telemetry needs all three. Set them in `.env` (locally) or inject them"
                " from Parameter Store (on AWS). Everything that does not read NGED's own bucket"
                " — training, cross-validation, the dashboards, every Delta read — works without"
                " them."
            )

    def get_nged_s3_store(self) -> obstore.store.S3Store:
        """Returns an initialised obstore.store.S3Store instance for the NGED bucket.

        Returns:
            An ``S3Store`` bound to ``nged_s3_bucket_url`` and authenticated with the
            ``nged_s3_bucket_access_key`` / ``nged_s3_bucket_secret`` pair, ready to read
            NGED's source JSON files.

        Raises:
            ValueError: If any of the three source-bucket credentials is unset.
        """
        self.require_nged_source_credentials()
        return obstore.store.S3Store.from_url(
            url=self.nged_s3_bucket_url,
            config={
                "aws_access_key_id": self.nged_s3_bucket_access_key,
                "aws_secret_access_key": self.nged_s3_bucket_secret,
            },
        )

    cv_config_path: Path = Field(
        default=PROJECT_ROOT / "conf" / "cv" / "default.yaml",
        description=(
            "Path to the canonical cross-validation fold definitions. These folds are the"
            " shared leaderboard evaluation protocol; read them from here, never hard-coded."
        ),
    )
    # --- Storage roots -------------------------------------------------------------------
    #
    # data_path_internal and data_path_delivery hold the (S3-capable) data tables;
    # local_artifacts_path holds the always-local production model. Why the three roots are
    # split, and what belongs in each:
    # https://openclimatefix.github.io/nged-substation-forecast/live_service/setup/

    data_path_internal: str = Field(
        default=str(PROJECT_ROOT / "data"),
        description=(
            "Root of OCF's own working data tables (NWP, power, forecast_metrics, …). A local"
            " path by default; may be a remote URI (e.g. 's3://bucket/data') so the data tables"
            " live on S3. Every *_data_path below that is left unset derives from this root,"
            " except the NGED-facing delivery tables, which derive from data_path_delivery"
            " instead."
        ),
    )
    data_path_delivery: str = Field(
        default=str(PROJECT_ROOT / "data"),
        description=(
            "Root of the NGED-facing delivery tables (power_forecasts, effective_capacity, …)."
            " Defaults to the same local path as data_path_internal, so local dev sees one"
            " directory and the split is invisible; on AWS this points at the separate,"
            " NGED-facing bucket."
        ),
    )
    local_artifacts_path: str = Field(
        default=str(PROJECT_ROOT / "data"),
        description=(
            "Root of the always-local artifacts (the production model). Kept separate from"
            " the data-table roots because these back local-filesystem-only libraries and"
            " must stay local even when the data tables live on S3."
        ),
    )

    # --- Object-store credentials for the data tables (used only when a data-path root is remote)
    #
    # All four are empty by default. On AWS they stay unset, because object_store — the Rust
    # object-storage library underneath delta-rs, Polars, and obstore — auto-discovers the
    # credentials of the AWS Identity and Access Management (IAM) role. Set them only for a dev
    # endpoint or for MinIO, the S3-compatible store a developer runs locally in place of S3. The
    # AWS/dev split, and how these four data-store settings differ from the nged_s3_bucket_*
    # source-bucket credentials above:
    # https://openclimatefix.github.io/nged-substation-forecast/live_service/setup/

    data_store_endpoint_url: str = Field(
        default="",
        description=(
            "S3-compatible endpoint URL for the data tables (e.g. a MinIO/dev endpoint). Empty on"
            " AWS, where the endpoint is inferred. When set, the store is also allowed to use"
            " plain HTTP (dev endpoints rarely have TLS)."
        ),
    )
    data_store_access_key_id: str = Field(
        default="", description="Access key for the data-table object store; empty on AWS (IAM)."
    )
    data_store_secret_access_key: str = Field(
        default="", description="Secret key for the data-table object store; empty on AWS (IAM)."
    )
    data_store_region: str = Field(
        default="", description="Region for the data-table object store; empty to auto-discover."
    )

    @property
    def storage_options(self) -> ObjectStoreOptions:
        """delta-rs / polars / obstore ``storage_options`` for the managed data tables.

        Empty on AWS, where object_store auto-discovers the IAM-role credentials and region of
        the AWS Fargate task the pipeline runs in. Empty too for a local data-path root, because
        delta-rs ignores storage_options there. Populated from the ``data_store_*`` settings only
        for a dev/MinIO/S3-compatible endpoint. The ``aws_*`` keys are the shared object_store
        aliases understood by delta-rs, polars cloud IO, and obstore alike, so one value feeds
        every IO site. Returned as an ``ObjectStoreOptions`` ``TypedDict``, so that ``ty``, the
        type checker this repo runs, checks each key here, where the keys are authored. Widen the
        ``TypedDict`` to a plain ``dict`` at each IO boundary with ``typeddict_to_dict``.
        """
        options: ObjectStoreOptions = {}
        if self.data_store_endpoint_url:
            options["aws_endpoint_url"] = self.data_store_endpoint_url
            options["aws_allow_http"] = "true"
        if self.data_store_access_key_id:
            options["aws_access_key_id"] = self.data_store_access_key_id
        if self.data_store_secret_access_key:
            options["aws_secret_access_key"] = self.data_store_secret_access_key
        if self.data_store_region:
            options["aws_region"] = self.data_store_region
        return options

    # --- Managed data tables (derive from a root unless explicitly set) --------------------
    #
    # Each defaults to "" — a sentinel meaning "derive from data_path_internal or
    # data_path_delivery in _derive_unset_paths". The derive-from-root convention and per-table
    # overrides: https://openclimatefix.github.io/nged-substation-forecast/live_service/setup/

    nged_data_path: str = ""
    """Directory holding the NGED power_time_series Delta table and metadata parquet."""
    nwp_data_path: str = ""
    """Delta table of NWP weather data."""
    power_forecasts_data_path: str = ""
    """Delta table of power forecasts, partitioned by experiment_name and fold_id, so its files
    are grouped on disk by those two columns.

    An NGED-facing delivery table — derives from data_path_delivery, not data_path_internal.
    """
    forecast_metrics_data_path: str = ""
    """Delta table of forecast evaluation metrics."""
    power_time_series_data_path: str = ""
    """Delta table of half-hourly power observations (under nged_data_path)."""
    metadata_path: str = ""
    """Parquet file of per-series substation metadata (under nged_data_path)."""
    eligible_time_series_data_path: str = Field(
        default="",
        description=(
            "Delta table of the canonical per-fold eligible time_series_id population, written"
            " by the eligible_time_series asset (partitioned by fold_id) and read by"
            " trained_cv_model, so every experiment trains a fold on the identical,"
            " experiment-independent population. cv_power_forecasts inherits that same population"
            " from the trained model's trained_time_series_ids rather than reading this table."
        ),
    )
    effective_capacity_data_path: str = Field(
        default="",
        description=(
            "Delta table of per-series effective capacity (v0.1: full-history P99 of |power|),"
            " written by the effective_capacity asset and read by the metrics asset as the NMAE"
            " denominator. An NGED-facing delivery table — derives from data_path_delivery, not"
            " data_path_internal."
        ),
    )
    h3_grid_weights_path: str = ""
    """Parquet file of fractional H3 cell overlap with the Great Britain boundary."""

    # --- Always-local artifacts (derive from local_artifacts_path unless explicitly set) --

    production_model_path: str = Field(
        default="",
        description=(
            "Directory holding the current production model, written by the promoted_model"
            " asset (ml_core.production_helpers.fetch_model_artifacts) and read by"
            " live_forecasts via a plain BaseForecaster disk load — no MLflow at inference"
            " time. Later COPY'd into the container image at build time (issue #222)."
        ),
    )

    # --- Sentry observability (all optional; an empty data source name (DSN) disables Sentry) ---

    sentry_dsn: str = Field(
        default="",
        description=(
            "Sentry.io project DSN. Empty (the default) disables Sentry entirely — every Sentry"
            " code path becomes a no-op — so laptops and CI need no Sentry config. Set it in the"
            " .env file to enable error telemetry. Passed explicitly to sentry_sdk.init as the"
            " single source of truth (the SDK would otherwise also auto-read the SENTRY_DSN env"
            " var)."
        ),
    )
    sentry_environment: str = Field(
        default="local",
        description=(
            "Sentry environment tag, the dimension that separates deployments in Sentry's UI,"
            " alerts, and cron monitors. The always-on production box sets 'production'; each"
            " developer overrides the 'local' fallback with '<name>-laptop' (e.g. 'jacks-laptop')"
            " so laptop telemetry is cleanly filterable and the production-scoped missed-check-in"
            " alert never fires for a laptop."
        ),
    )
    sentry_traces_sample_rate: float = Field(
        default=0.0,
        description=(
            "Fraction of transactions sampled for Sentry performance tracing (off by default)."
        ),
    )
    sentry_monitor_forecasts: bool = Field(
        default=False,
        description=(
            "Emit the live_forecasts success heartbeat to Sentry's cron monitor (the"
            " missed-check-in alarm). True only on the always-on production deployment; left"
            " False on laptops so an intermittently-run laptop never registers a monitor"
            " environment that Sentry would then mark as missed."
        ),
    )

    @model_validator(mode="after")
    def _derive_unset_paths(self) -> Self:
        """Fill any unset ("") path from its root, so callers always see a concrete path.

        The default layout lives here and nowhere else. A field set explicitly (e.g. via its env
        var) keeps its value; only the "" sentinels are derived.
        """
        self.nged_data_path = self.nged_data_path or uri_join(self.data_path_internal, "NGED")
        self.nwp_data_path = self.nwp_data_path or uri_join(self.data_path_internal, "NWP")
        self.forecast_metrics_data_path = self.forecast_metrics_data_path or uri_join(
            self.data_path_internal, "forecast_metrics"
        )
        self.eligible_time_series_data_path = self.eligible_time_series_data_path or uri_join(
            self.data_path_internal, "eligible_time_series"
        )
        self.h3_grid_weights_path = self.h3_grid_weights_path or uri_join(
            self.data_path_internal, "h3_grid_weights.parquet"
        )
        # NGED-facing delivery tables derive from data_path_delivery instead, so a new delivery
        # table can't silently land in the internal bucket by inheriting the default derivation.
        # `power_forecast` and `effective_capacity` are two of the five tables in NGED's stable
        # delivery contract — the agreement with NGED about which tables we publish, not a Patito
        # schema of the kind this package otherwise calls a contract. The other three tables are
        # listed at
        # <https://openclimatefix.github.io/nged-substation-forecast/roadmap/delivery-tables/>,
        # with what each one holds. Why the delivery tables live in a bucket of their own is at
        # <https://openclimatefix.github.io/nged-substation-forecast/architecture/forecast-delivery/#securing-it>.
        self.power_forecasts_data_path = self.power_forecasts_data_path or uri_join(
            self.data_path_delivery, "power_forecasts"
        )
        self.effective_capacity_data_path = self.effective_capacity_data_path or uri_join(
            self.data_path_delivery, "effective_capacity"
        )
        # Derived from nged_data_path (itself derived above).
        self.power_time_series_data_path = self.power_time_series_data_path or uri_join(
            self.nged_data_path, "power_time_series.delta"
        )
        self.metadata_path = self.metadata_path or uri_join(self.nged_data_path, "metadata.parquet")
        # Always-local artifacts.
        self.production_model_path = self.production_model_path or uri_join(
            self.local_artifacts_path, "production_model"
        )
        return self

    # Tell Pydantic to override defaults with fields set in the .env file.
    model_config = SettingsConfigDict(
        env_file=PROJECT_ROOT / ".env",
        extra="forbid",
        env_file_encoding="utf-8",
        env_prefix="",
    )

    @field_validator("nged_s3_bucket_url")
    @classmethod
    def validate_url(cls, v: str) -> str:
        """Validate that the S3 bucket URL is a valid URL, if one was given at all.

        Empty means "no NGED ingest configured", which is the default and not an error;
        ``require_nged_source_credentials`` is what rejects it where it matters. A *non-empty*
        value still has to be a real URL — strict about malformed, liberal about missing.
        """
        if v:
            url_adapter.validate_python(v)
        return v
Attributes
mlflow_tracking_uri = Field(default='sqlite:///mlflow.db', description='MLflow tracking URI. Centralised here for environment-specific configuration (e.g., local SQLite for development, remote server for production). The application entrypoint should read this setting and set the MLFLOW_TRACKING_URI environment variable accordingly, which MLflow will automatically pick up.') class-attribute instance-attribute
nged_s3_bucket_url = Field(default='', description="NGED source S3 bucket URL. Typically stored in the .env file. Empty unless you need to ingest from NGED's bucket; `get_nged_s3_store` raises if it is.") class-attribute instance-attribute
nged_s3_bucket_access_key = Field(default='', description='Access key for the NGED source S3 bucket. Typically stored in the .env file.') class-attribute instance-attribute
nged_s3_bucket_secret = Field(default='', description='Secret key for the NGED source S3 bucket. Typically stored in the .env file.') class-attribute instance-attribute
cv_config_path = Field(default=PROJECT_ROOT / 'conf' / 'cv' / 'default.yaml', description='Path to the canonical cross-validation fold definitions. These folds are the shared leaderboard evaluation protocol; read them from here, never hard-coded.') class-attribute instance-attribute
data_path_internal = Field(default=str(PROJECT_ROOT / 'data'), description="Root of OCF's own working data tables (NWP, power, forecast_metrics, …). A local path by default; may be a remote URI (e.g. 's3://bucket/data') so the data tables live on S3. Every *_data_path below that is left unset derives from this root, except the NGED-facing delivery tables, which derive from data_path_delivery instead.") class-attribute instance-attribute
data_path_delivery = Field(default=str(PROJECT_ROOT / 'data'), description='Root of the NGED-facing delivery tables (power_forecasts, effective_capacity, …). Defaults to the same local path as data_path_internal, so local dev sees one directory and the split is invisible; on AWS this points at the separate, NGED-facing bucket.') class-attribute instance-attribute
local_artifacts_path = Field(default=str(PROJECT_ROOT / 'data'), description='Root of the always-local artifacts (the production model). Kept separate from the data-table roots because these back local-filesystem-only libraries and must stay local even when the data tables live on S3.') class-attribute instance-attribute
data_store_endpoint_url = Field(default='', description='S3-compatible endpoint URL for the data tables (e.g. a MinIO/dev endpoint). Empty on AWS, where the endpoint is inferred. When set, the store is also allowed to use plain HTTP (dev endpoints rarely have TLS).') class-attribute instance-attribute
data_store_access_key_id = Field(default='', description='Access key for the data-table object store; empty on AWS (IAM).') class-attribute instance-attribute
data_store_secret_access_key = Field(default='', description='Secret key for the data-table object store; empty on AWS (IAM).') class-attribute instance-attribute
data_store_region = Field(default='', description='Region for the data-table object store; empty to auto-discover.') class-attribute instance-attribute
storage_options property

delta-rs / polars / obstore storage_options for the managed data tables.

Empty on AWS, where object_store auto-discovers the IAM-role credentials and region of the AWS Fargate task the pipeline runs in. Empty too for a local data-path root, because delta-rs ignores storage_options there. Populated from the data_store_* settings only for a dev/MinIO/S3-compatible endpoint. The aws_* keys are the shared object_store aliases understood by delta-rs, polars cloud IO, and obstore alike, so one value feeds every IO site. Returned as an ObjectStoreOptions TypedDict, so that ty, the type checker this repo runs, checks each key here, where the keys are authored. Widen the TypedDict to a plain dict at each IO boundary with typeddict_to_dict.

nged_data_path = '' class-attribute instance-attribute

Directory holding the NGED power_time_series Delta table and metadata parquet.

nwp_data_path = '' class-attribute instance-attribute

Delta table of NWP weather data.

power_forecasts_data_path = '' class-attribute instance-attribute

Delta table of power forecasts, partitioned by experiment_name and fold_id, so its files are grouped on disk by those two columns.

An NGED-facing delivery table — derives from data_path_delivery, not data_path_internal.

forecast_metrics_data_path = '' class-attribute instance-attribute

Delta table of forecast evaluation metrics.

power_time_series_data_path = '' class-attribute instance-attribute

Delta table of half-hourly power observations (under nged_data_path).

metadata_path = '' class-attribute instance-attribute

Parquet file of per-series substation metadata (under nged_data_path).

eligible_time_series_data_path = Field(default='', description="Delta table of the canonical per-fold eligible time_series_id population, written by the eligible_time_series asset (partitioned by fold_id) and read by trained_cv_model, so every experiment trains a fold on the identical, experiment-independent population. cv_power_forecasts inherits that same population from the trained model's trained_time_series_ids rather than reading this table.") class-attribute instance-attribute
effective_capacity_data_path = Field(default='', description='Delta table of per-series effective capacity (v0.1: full-history P99 of |power|), written by the effective_capacity asset and read by the metrics asset as the NMAE denominator. An NGED-facing delivery table — derives from data_path_delivery, not data_path_internal.') class-attribute instance-attribute
h3_grid_weights_path = '' class-attribute instance-attribute

Parquet file of fractional H3 cell overlap with the Great Britain boundary.

production_model_path = Field(default='', description="Directory holding the current production model, written by the promoted_model asset (ml_core.production_helpers.fetch_model_artifacts) and read by live_forecasts via a plain BaseForecaster disk load — no MLflow at inference time. Later COPY'd into the container image at build time (issue #222).") class-attribute instance-attribute
sentry_dsn = Field(default='', description='Sentry.io project DSN. Empty (the default) disables Sentry entirely — every Sentry code path becomes a no-op — so laptops and CI need no Sentry config. Set it in the .env file to enable error telemetry. Passed explicitly to sentry_sdk.init as the single source of truth (the SDK would otherwise also auto-read the SENTRY_DSN env var).') class-attribute instance-attribute
sentry_environment = Field(default='local', description="Sentry environment tag, the dimension that separates deployments in Sentry's UI, alerts, and cron monitors. The always-on production box sets 'production'; each developer overrides the 'local' fallback with '<name>-laptop' (e.g. 'jacks-laptop') so laptop telemetry is cleanly filterable and the production-scoped missed-check-in alert never fires for a laptop.") class-attribute instance-attribute
sentry_traces_sample_rate = Field(default=0.0, description='Fraction of transactions sampled for Sentry performance tracing (off by default).') class-attribute instance-attribute
sentry_monitor_forecasts = Field(default=False, description="Emit the live_forecasts success heartbeat to Sentry's cron monitor (the missed-check-in alarm). True only on the always-on production deployment; left False on laptops so an intermittently-run laptop never registers a monitor environment that Sentry would then mark as missed.") class-attribute instance-attribute
model_config = SettingsConfigDict(env_file=PROJECT_ROOT / '.env', extra='forbid', env_file_encoding='utf-8', env_prefix='') class-attribute instance-attribute
Methods:
require_nged_source_credentials()

Raise ValueError unless all three NGED source-bucket credentials are set.

Call this immediately before doing something that reads NGED's bucket, so the failure names the missing configuration instead of surfacing as an opaque auth error from obstore.

Do not call this method from process start-up or module import in order to fail a deployment fast. Inference needs none of these credentials, so a start-up call would stop the forecast over a missing ingest secret. See the AWS runbook.

Raises:

Type Description
ValueError

Naming exactly which of the three environment variables are unset.
Source code in packages/contracts/src/contracts/settings.py 
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
def require_nged_source_credentials(self) -> None:
    """Raise ``ValueError`` unless all three NGED source-bucket credentials are set.

    Call this immediately before doing something that reads NGED's bucket, so the failure names
    the missing configuration instead of surfacing as an opaque auth error from ``obstore``.

    Do *not* call this method from process start-up or module import in order to fail a
    deployment fast. Inference needs none of these credentials, so a start-up call would stop
    the forecast over a missing ingest secret. See the [AWS
    runbook](https://openclimatefix.github.io/nged-substation-forecast/live_service/aws/#step-8-store-secrets-in-parameter-store).

    Raises:
        ValueError: Naming exactly which of the three environment variables are unset.
    """
    missing = [
        name
        for name, value in (
            ("NGED_S3_BUCKET_URL", self.nged_s3_bucket_url),
            ("NGED_S3_BUCKET_ACCESS_KEY", self.nged_s3_bucket_access_key),
            ("NGED_S3_BUCKET_SECRET", self.nged_s3_bucket_secret),
        )
        if not value
    ]
    if missing:
        raise ValueError(
            f"NGED source-bucket credentials are not set: {', '.join(missing)}. Reading"
            " NGED's telemetry needs all three. Set them in `.env` (locally) or inject them"
            " from Parameter Store (on AWS). Everything that does not read NGED's own bucket"
            " — training, cross-validation, the dashboards, every Delta read — works without"
            " them."
        )
get_nged_s3_store()

Returns an initialised obstore.store.S3Store instance for the NGED bucket.

Returns:

Type Description
S3Store

An S3Store bound to nged_s3_bucket_url and authenticated with the
S3Store

nged_s3_bucket_access_key / nged_s3_bucket_secret pair, ready to read
S3Store

NGED's source JSON files.

Raises:

Type Description
ValueError

If any of the three source-bucket credentials is unset.
Source code in packages/contracts/src/contracts/settings.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def get_nged_s3_store(self) -> obstore.store.S3Store:
    """Returns an initialised obstore.store.S3Store instance for the NGED bucket.

    Returns:
        An ``S3Store`` bound to ``nged_s3_bucket_url`` and authenticated with the
        ``nged_s3_bucket_access_key`` / ``nged_s3_bucket_secret`` pair, ready to read
        NGED's source JSON files.

    Raises:
        ValueError: If any of the three source-bucket credentials is unset.
    """
    self.require_nged_source_credentials()
    return obstore.store.S3Store.from_url(
        url=self.nged_s3_bucket_url,
        config={
            "aws_access_key_id": self.nged_s3_bucket_access_key,
            "aws_secret_access_key": self.nged_s3_bucket_secret,
        },
    )
validate_url(v) classmethod

Validate that the S3 bucket URL is a valid URL, if one was given at all.

Empty means "no NGED ingest configured", which is the default and not an error; require_nged_source_credentials is what rejects it where it matters. A non-empty value still has to be a real URL — strict about malformed, liberal about missing.

Source code in packages/contracts/src/contracts/settings.py 
416
417
418
419
420
421
422
423
424
425
426
427
@field_validator("nged_s3_bucket_url")
@classmethod
def validate_url(cls, v: str) -> str:
    """Validate that the S3 bucket URL is a valid URL, if one was given at all.

    Empty means "no NGED ingest configured", which is the default and not an error;
    ``require_nged_source_credentials`` is what rejects it where it matters. A *non-empty*
    value still has to be a real URL — strict about malformed, liberal about missing.
    """
    if v:
        url_adapter.validate_python(v)
    return v

Functions:

get_settings() cached

Return the shared, lazily-constructed Settings singleton.

Prefer get_settings() over constructing Settings() at module import time, because instantiation reads .env and the environment. Deferring that read to first use keeps library modules such as the contracts schemas importable whatever the environment holds. Deferring it also lets a test change the environment before the first read.

Cached with lru_cache so every caller shares one instance. Call get_settings.cache_clear() in a test that needs to re-read the environment after changing it.

Source code in packages/contracts/src/contracts/settings.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
@lru_cache
def get_settings() -> Settings:
    """Return the shared, lazily-constructed ``Settings`` singleton.

    Prefer ``get_settings()`` over constructing ``Settings()`` at module import time, because
    instantiation reads ``.env`` and the environment. Deferring that read to first *use* keeps
    library modules such as the ``contracts`` schemas importable whatever the environment holds.
    Deferring it also lets a test change the environment before the first read.

    Cached with ``lru_cache`` so every caller shares one instance. Call
    ``get_settings.cache_clear()`` in a test that needs to re-read the environment after changing
    it.
    """
    return Settings()

contracts.uri

Small URI helpers for paths that may be local filesystem paths or remote URIs.

Settings data-location fields are plain str so they can hold either a local path (/home/.../data/NWP) or a remote URI (s3://bucket/NWP). pathlib.Path mangles a remote URI (Path("s3://b/a") / "c" collapses the double slash after the scheme and yields "s3:/b/a/c"), so joins route through here.

The asset IO layer is the code that writes our Delta tables and parquet files. That layer takes two steps around every Delta/parquet write: it makes sure the parent directory exists, and it checks whether a table or object is already there. The existence/parent helpers below give that layer a single local-or-remote-aware call for those two steps. Making the parent directory is a no-op on object stores, which have no directories. Remote calls go through delta-rs / obstore with the caller's storage_options so the same code path serves both a local data-path root and an s3:// one.

Classes

ObjectStoreOptions

Bases: TypedDict

object_store options for the managed data tables.

These are the shared aws_* aliases understood by delta-rs, Polars, and obstore alike, so one value feeds every IO site.

Authored as a TypedDict (rather than a bare dict[str, str]) so ty checks every key where it is written — see Settings.storage_options. Widen it to the plain dict the IO libraries expect with typeddict_to_dict at each call boundary. The mapping is empty on AWS, where object_store auto-discovers the identity-and-access-management (IAM) role's credentials and region. The mapping is empty for a local data-path root too.

Source code in packages/contracts/src/contracts/uri.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class ObjectStoreOptions(TypedDict, total=False):
    """object_store options for the managed data tables.

    These are the shared ``aws_*`` aliases understood by delta-rs, Polars, and obstore alike, so
    one value feeds every IO site.

    Authored as a ``TypedDict`` (rather than a bare ``dict[str, str]``) so ``ty`` checks every
    key where it is written — see ``Settings.storage_options``. Widen it to the plain ``dict``
    the IO libraries expect with ``typeddict_to_dict`` at each call boundary. The mapping is
    empty on AWS, where object_store auto-discovers the identity-and-access-management (IAM)
    role's credentials and region. The mapping is empty for a local data-path root too.
    """

    aws_endpoint_url: str
    aws_allow_http: str
    aws_access_key_id: str
    aws_secret_access_key: str
    aws_region: str
Attributes
aws_endpoint_url instance-attribute
aws_allow_http instance-attribute
aws_access_key_id instance-attribute
aws_secret_access_key instance-attribute
aws_region instance-attribute

Functions:

is_remote_uri(uri)

Return whether uri carries a URI scheme (e.g. s3://) rather than a local path.

Source code in packages/contracts/src/contracts/uri.py 
50
51
52
def is_remote_uri(uri: str) -> bool:
    """Return whether ``uri`` carries a URI scheme (e.g. ``s3://``) rather than a local path."""
    return _SCHEME_SEP in uri

uri_join(base, *parts)

Join parts onto base with local- or remote-aware semantics.

A local base joins through pathlib, which yields an absolute path string. A remote, scheme-bearing base joins posix-style, so "s3://bucket/a" + "b" stays "s3://bucket/a/b" rather than being mangled by Path.__truediv__.

Source code in packages/contracts/src/contracts/uri.py 
55
56
57
58
59
60
61
62
63
64
def uri_join(base: str, *parts: str) -> str:
    """Join ``parts`` onto ``base`` with local- or remote-aware semantics.

    A local base joins through ``pathlib``, which yields an absolute path string. A remote,
    scheme-bearing base joins posix-style, so ``"s3://bucket/a"`` + ``"b"`` stays
    ``"s3://bucket/a/b"`` rather than being mangled by ``Path.__truediv__``.
    """
    if is_remote_uri(base):
        return posixpath.join(base.rstrip("/"), *parts)
    return str(Path(base).joinpath(*parts))

if_local_path_then_make_parent_dir(uri)

Create the parent directory of a local uri; a no-op for a remote URI.

Local filesystems need a table or file's parent directory to exist before a write. Object stores (s3://) have no directories, and a write creates the key's prefix implicitly. There is therefore nothing to do for a remote URI, and this function returns immediately.

Source code in packages/contracts/src/contracts/uri.py 
67
68
69
70
71
72
73
74
75
76
def if_local_path_then_make_parent_dir(uri: str) -> None:
    """Create the parent directory of a *local* ``uri``; a no-op for a remote URI.

    Local filesystems need a table or file's parent directory to exist before a write. Object
    stores (``s3://``) have no directories, and a write creates the key's prefix implicitly.
    There is therefore nothing to do for a remote URI, and this function returns immediately.
    """
    if is_remote_uri(uri):
        return
    Path(uri).parent.mkdir(parents=True, exist_ok=True)

delta_table_exists(uri, storage_options=None)

Return whether a Delta table already exists at uri (local path or remote URI).

Wraps DeltaTable.is_deltatable, which inspects the _delta_log through delta-rs' object_store and so works identically for a local path and an s3:// URI given the matching storage_options. Use this function instead of Path(uri).exists() at the write-guard sites. There, pathlib reads s3://bucket/key as a relative local path that happens not to exist. Path.exists() therefore returns False without raising, and the guard concludes the table is absent.

Source code in packages/contracts/src/contracts/uri.py 
79
80
81
82
83
84
85
86
87
88
89
def delta_table_exists(uri: str, storage_options: ObjectStoreOptions | None = None) -> bool:
    """Return whether a Delta table already exists at ``uri`` (local path or remote URI).

    Wraps ``DeltaTable.is_deltatable``, which inspects the ``_delta_log`` through delta-rs'
    object_store and so works identically for a local path and an ``s3://`` URI given the
    matching ``storage_options``. Use this function instead of ``Path(uri).exists()`` at the
    write-guard sites. There, ``pathlib`` reads ``s3://bucket/key`` as a relative local path that
    happens not to exist. ``Path.exists()`` therefore returns ``False`` without raising, and the
    guard concludes the table is absent.
    """
    return DeltaTable.is_deltatable(uri, storage_options=typeddict_to_dict(storage_options) or {})

object_exists(uri, storage_options=None)

Return whether a single object/file at uri exists (local file or remote object).

For a local uri this is Path.exists(); for a remote URI it issues an object-store head via obstore, so it works for a plain file (e.g. a .parquet) that is not a Delta table. Use delta_table_exists for Delta tables.

Source code in packages/contracts/src/contracts/uri.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def object_exists(uri: str, storage_options: ObjectStoreOptions | None = None) -> bool:
    """Return whether a single object/file at ``uri`` exists (local file or remote object).

    For a local ``uri`` this is ``Path.exists()``; for a remote URI it issues an object-store
    ``head`` via obstore, so it works for a plain file (e.g. a ``.parquet``) that is not a Delta
    table. Use ``delta_table_exists`` for Delta tables.
    """
    if not is_remote_uri(uri):
        return Path(uri).exists()
    parsed = urlparse(uri)
    store = obstore.store.S3Store(parsed.netloc, config=typeddict_to_dict(storage_options) or {})
    try:
        obstore.head(store, parsed.path.lstrip("/"))
    except FileNotFoundError:
        return False
    return True