Skip to content

Contracts API

Contracts

Defines the "data contracts": the schemas defining the precise shape of each data source and its semantics.

Dependency Isolation

This package is designed to be extremely lightweight. It defines the shape of the data using Patito and Polars, but it does not contain any ML-specific logic or heavy dependencies like MLflow. This ensures that any component in the system (e.g., a data ingestion script or a dashboard) can import these schemas without bringing in the entire ML stack.

Key Data Contracts

  • PowerTimeSeries: Half-hourly power observations (MW or MVA) per time_series_id, as received from NGED.
  • TimeSeriesMetadata: Substation and customer meter metadata, including lat/lon, H3 index, and asset type (primary substation, GSP, BSP, solar PV, wind, BESS, etc.).
  • Nwp: ECMWF ENS NWP weather data in physical units (Float32), on disk and in memory alike. The on-disk copy is rounded to a 13-bit significand and laid out for compression and row-group pruning by delta_store.nwp.
  • AllFeatures: The final joined dataset passed to ML models. Primary key is (time_series_id, power_fcst_init_time, valid_time[, ensemble_member]). Includes NWP weather variables, power lag/rolling features, static metadata columns, and datetime features.
  • PowerForecast: ML model output schema. Power values are in the range [−1, +1] (normalised). Includes power_fcst_model_name, power_fcst_model_version, power_fcst_init_time, nwp_init_time, valid_time, time_series_id, and ensemble_member.

Design Principles

  • Column naming: Prefer snake_case, except for acronyms or SI units. Capitalise "DER" (distributed energy resource) and use uppercase for "MW" (megawatts).
  • 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.
  • 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.

contracts.common

Attributes

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

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 thirteen quantile levels agreed with NGED for the delivery tables.

Deliberately tail-heavy: 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 land in v0.5.

Functions:

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 
39
40
41
42
43
44
45
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)

Validates that the schema of a Polars DataFrame or LazyFrame matches the schema defined in a Patito model, raising DataFrameValidationError on failure. On LazyFrames, this function doesn't materialize any data, it just calls collect_schema().

Source code in packages/contracts/src/contracts/common.py 
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
def validate_schema(model: Type[pt.Model], df: pl.DataFrame | pl.LazyFrame) -> None:
    """Validates that the schema of a Polars DataFrame or LazyFrame matches the schema defined in a
    Patito model, raising DataFrameValidationError on failure. On LazyFrames, this function doesn't
    materialize any data, it just calls `collect_schema()`.
    """
    # Get actual schema
    if isinstance(df, pl.LazyFrame):
        actual_schema = dict(df.collect_schema())
    else:
        actual_schema = 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

Data schemas for the NGED substation forecast project.

Attributes

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.
  • Raw Flow: Used for BSP and 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

PowerTimeSeries

Bases: Model

Source code in packages/contracts/src/contracts/power_schemas.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class PowerTimeSeries(pt.Model):
    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).",
    )

    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.",
    )

    @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 at :00 or :30 and uniqueness."""
        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 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

    # Define it as a ClassVar so Patito/Pydantic knows it's 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='End time of the 30-minute observation period (all NGED data is already half-hourly).') 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.') 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 at :00 or :30 and uniqueness.

Source code in packages/contracts/src/contracts/power_schemas.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@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 at :00 or :30 and uniqueness."""
    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 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

TimeSeriesMetadata

Bases: Model

Source code in packages/contracts/src/contracts/power_schemas.py 
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
class TimeSeriesMetadata(pt.Model):
    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 trail area, this is always ‘EMids’).",
    )

    substation_number: int = pt.Field(
        dtype=pl.Int32,
        gt=0,
        lt=1_000_000,
        description="Perhaps surprisingly, 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. 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 the JSON data.
        description=(
            "WKT polygon for the asset’s area. In the trial, only Primary substations have this."
            " NGED don’t have polygons for customer sites (though they hope to add that in future)."
            " For customer sites, where present, 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 trail area, this is always ‘EMids’).') class-attribute instance-attribute
substation_number = pt.Field(dtype=(pl.Int32), gt=0, lt=1000000, description='Perhaps surprisingly, 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. 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='WKT polygon for the asset’s area. In the trial, only Primary substations have this. NGED don’t have polygons for customer sites (though they hope to add that in future). For customer sites, where present, 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 deterministic ensemble forecasts.

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

Source code in packages/contracts/src/contracts/power_schemas.py 
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
class PowerForecast(pt.Model):
    """Forecast data schema for deterministic ensemble forecasts.

    Internal vs delivered schema (Milestone 1 report Table 1, p.28): the columns
    ``experiment_name``, ``fold_id``, and ``ml_flow_experiment_id`` are INTERNAL-ONLY —
    they exist on this schema and the internal ``power_forecasts`` Delta table to support
    cross-validation and the leaderboard, but 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. 0 is the control 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 them cast-free and lets
    # predicate pushdown work. See the "declare Delta filter/partition columns as String" gotcha:
    # ../../../../CLAUDE.md#delta-lake-dictionary-encoded-columns-declare-delta-filterpartition-columns-as-string
    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`."
            """ Positive values mean "power sent to NGED's grid","""
            """ and negative values mean "power drawn from NGED's grid"."""
            " 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), per the
            # delivery-contract design agreed with NGED in the Milestone 1 report.
            # For this very early version we forecast raw MW/MVA because we are not yet
            # estimating capacity; we will switch to the scaled value once capacity estimation
            # lands (roadmap v0.7).
        ),
    )

    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."
        ),
    )
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. 0 is the control 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`. Positive values mean "power sent to NGED\'s grid", and negative values mean "power drawn from NGED\'s grid". 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

EffectiveCapacity

Bases: Model

Effective capacity of each time series at each half-hourly timestep.

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. This is a static scalar per series.

Planned upgrade (v0.7): replace the P99 scalar with a time-varying capacity estimate (see docs/techniques/convex-optimisation.md and docs/techniques/differentiable-physics.md for the candidate estimation methods), giving one row per (time_series_id, time) half-hourly timestep. This schema is unchanged; the effective_capacity asset body changes and the metrics pipeline swaps its time_series_id-only NMAE-denominator join for a temporal as-of join. Do not pre-densify the v0.1 scalar into one row per half-hour — densifying a constant buys nothing, and the as-of join handles sparse capacity rows naturally.

Source code in packages/contracts/src/contracts/power_schemas.py 
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
class EffectiveCapacity(pt.Model):
    """Effective capacity of each time series at each half-hourly timestep.

    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. This is a static scalar per series.

    **Planned upgrade (v0.7):** replace the P99 scalar with a time-varying capacity estimate
    (see ``docs/techniques/convex-optimisation.md`` and
    ``docs/techniques/differentiable-physics.md`` for the candidate estimation methods),
    giving one row per ``(time_series_id, time)`` half-hourly timestep. This schema is unchanged;
    the ``effective_capacity`` asset body changes and the ``metrics`` pipeline swaps its
    ``time_series_id``-only NMAE-denominator join for a temporal as-of join. Do **not** pre-densify
    the v0.1 scalar into one row per half-hour — densifying a constant buys nothing, 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 (MW) of this asset at this timestep. "
            "For generators: absorbs PV panel degradation, partial inverter trips, etc., "
            "but ignores 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 load over a rolling time window, "
            "under normal running arrangement only. '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 (MW) of this asset at this timestep. For generators: absorbs PV panel degradation, partial inverter trips, etc., but ignores 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 load over a rolling time window, under normal running arrangement only. 'Switched' power (Table 5) should be added or subtracted when a switching event is in effect.") class-attribute instance-attribute

contracts.weather_schemas

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

NWP_MODEL_ID_DTYPE = pl.Enum([(model.name) for model in NwpModelId]) module-attribute

Classes

NwpModelId

Bases: StrEnum

Source code in packages/contracts/src/contracts/weather_schemas.py 
34
35
class NwpModelId(StrEnum):
    ECMWF_ENS_0_25_degree = auto()
Attributes
ECMWF_ENS_0_25_degree = auto() class-attribute instance-attribute

NwpMetaData

Bases: Model

Metadata about numerical weather prediction models.

Source code in packages/contracts/src/contracts/weather_schemas.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class NwpMetaData(pt.Model):
    """Metadata about numerical weather prediction models."""

    nwp_model_id: str = pt.Field(
        dtype=NWP_MODEL_ID_DTYPE,
        description="Primary key for joining with NWP data (e.g. 'ECMWF_ENS_0_25_degree').",
        unique=True,
    )

    provider: str = pt.Field(
        dtype=pl.Enum(["ECMWF"]),
        description="NWP data provider (currently always 'ECMWF').",
    )

    h3_resolution: int = pt.Field(
        dtype=pl.Int8,
        description="H3 spatial resolution used for this NWP model's grid.",
    )

    is_ensemble: bool = pt.Field(description="Whether this NWP model produces ensemble forecasts.")

    @classmethod
    def load(cls, csv_path: str | Path | None = None) -> pt.DataFrame[Self]:
        """Load NWP metadata from a static CSV file.

        Args:
            csv_path: Path to the metadata CSV; defaults to ``get_settings().nwp_metadata_csv_path``
                (resolved lazily so importing this module needs no ``.env``).
        """
        if csv_path is None:
            csv_path = get_settings().nwp_metadata_csv_path
        df = pl.read_csv(csv_path)
        # Patito's .cast() will handle the conversion to the Enum type defined in the model
        return pt.DataFrame(df).set_model(cls).cast().validate()
Attributes
nwp_model_id = pt.Field(dtype=NWP_MODEL_ID_DTYPE, description="Primary key for joining with NWP data (e.g. 'ECMWF_ENS_0_25_degree').", unique=True) class-attribute instance-attribute
provider = pt.Field(dtype=(pl.Enum(['ECMWF'])), description="NWP data provider (currently always 'ECMWF').") class-attribute instance-attribute
h3_resolution = pt.Field(dtype=(pl.Int8), description="H3 spatial resolution used for this NWP model's grid.") class-attribute instance-attribute
is_ensemble = pt.Field(description='Whether this NWP model produces ensemble forecasts.') class-attribute instance-attribute
Methods:
load(csv_path=None) classmethod

Load NWP metadata from a static CSV file.

Parameters:

Name Type Description Default
csv_path str | Path | None

Path to the metadata CSV; defaults to get_settings().nwp_metadata_csv_path (resolved lazily so importing this module needs no .env).

 None
Source code in packages/contracts/src/contracts/weather_schemas.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
@classmethod
def load(cls, csv_path: str | Path | None = None) -> pt.DataFrame[Self]:
    """Load NWP metadata from a static CSV file.

    Args:
        csv_path: Path to the metadata CSV; defaults to ``get_settings().nwp_metadata_csv_path``
            (resolved lazily so importing this module needs no ``.env``).
    """
    if csv_path is None:
        csv_path = get_settings().nwp_metadata_csv_path
    df = pl.read_csv(csv_path)
    # Patito's .cast() will handle the conversion to the Enum type defined in the model
    return pt.DataFrame(df).set_model(cls).cast().validate()

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 — see docs/architecture/overview.md for the physical format and measured numbers.

validate is the fatal ingest gate; assess_nwp_quality reports the tolerated scattered nulls (the known upstream ECMWF ENS corruption) as a non-fatal check. Which null 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 
 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
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` — see `docs/architecture/overview.md` for the physical format
    and measured numbers.

    `validate` is the fatal ingest gate; `assess_nwp_quality` reports the tolerated scattered nulls
    (the known upstream ECMWF ENS corruption) as a non-fatal check. Which null patterns are fatal
    versus tolerated, and why, is documented at
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """

    nwp_model_id: str = pt.Field(
        dtype=NWP_MODEL_ID_DTYPE,
        description="The primary key for joining with NwpMetaData (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."
        ),
    )

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

    h3_index: int = pt.Field(
        dtype=pl.UInt64,
        description="H3 cell index. The H3 resolution for the nwp_model_id is stored in NwpMetaData.",
    )

    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: meters per second.",
        ge=0,
        le=200,  # Gemini says the highest non-tornadic surface wind speed recorded was 113 m/s
    )

    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.",
        ge=0,
        le=360,
    )

    wind_speed_100m: float = pt.Field(
        dtype=pl.Float32,
        description="Wind speed at 100 m. Unit: meters per second.",
        ge=0,
        le=200,  # Gemini says the highest non-tornadic surface wind speed recorded was 113 m/s
    )

    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.",
        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
    )

    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
    )

    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
    )

    # Precipitation and radiation variables are null for the first forecast step (lead time 0) in
    # ECMWF ENS. Also note that, whilst 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 these 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
    )

    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
    )

    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
    )

    categorical_precipitation_type_surface: int | None = pt.Field(
        dtype=pl.UInt8,
        description=(
            "This field is always NaN 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)."
            " 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.

    They share a de-accumulation step whose known upstream corruption leaves *scattered* per-pixel
    nulls beyond lead-0 (and all three are legitimately null at lead-0). Those scattered nulls are
    tolerated at ingest and reported by :func:`assess_nwp_quality`; only a whole-slice gap is fatal.
    See <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """

    # 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: rejecting whole-slice nulls in de-accumulated variables (scattered
        nulls are tolerated), enforcing uniqueness, and the ptype-introduction invariant."""
        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,
        )

        cls._check_no_whole_null_deaccumulated_slices(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_whole_null_deaccumulated_slices(cls, dataframe: pt.DataFrame[Self]) -> None:
        """Reject a *structural* gap: any de-accumulated variable whose (ensemble_member,
        valid_time) slice beyond lead-0 is *entirely* null across the grid.

        Scattered per-pixel nulls in the de-accumulated variables are *tolerated* — they are the
        known upstream ECMWF ENS corruption (empirically reaching a few percent of a slice), and
        every model already handles null precipitation/radiation because both are legitimately null
        at lead-0. A *whole-slice* null is different: it means the field is wholesale missing (a
        structural outage), which should fail ingest. :func:`assess_nwp_quality` reports the
        tolerated scatter as a non-fatal check. See
        <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.

        This relies on a slice having many grid cells (the production GB grid has ~1671), so that
        scatter (a few null cells) is cleanly distinct from a whole-slice gap (all cells null). On a
        degenerate one-cell slice the two are indistinguishable, but that does not arise in practice.
        """
        whole_null = _deaccumulated_null_breakdown(dataframe).filter(
            pl.col("n_null") == pl.col("n_total")
        )
        if whole_null.height:
            offenders = whole_null.select("variable", "ensemble_member", "valid_time").rows()
            raise ValueError(
                "Whole-slice null in a de-accumulated NWP variable beyond lead-0 — a wholesale "
                f"missing field, not tolerable scattered corruption. {whole_null.height} offending "
                f"(variable, ensemble_member, valid_time) slice(s), first 10: {offenders[:10]}"
            )

    @classmethod
    def _check_unique(cls, dataframe: pt.DataFrame[Self]) -> None:
        if (
            dataframe.select(["init_time", "valid_time", "ensemble_member", "h3_index"])
            .is_duplicated()
            .any()
        ):
            raise ValueError(
                "Duplicate entries found for (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 that `categorical_precipitation_type_surface` is all-null when
        init_time <= 2024-11-12, and is never null afterwards.

        Confirmed by direct inspection of the source data: the 2024-11-13 00Z run is the first
        run with `ptype` populated (0% null across all lead times), while 2024-11-12 and earlier
        are 100% null.
        """
        threshold_date = datetime(2024, 11, 12, tzinfo=timezone.utc)

        # Partition the dataframe based on the threshold date
        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``).
        """
        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=NWP_MODEL_ID_DTYPE, description="The primary key for joining with NwpMetaData (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.") class-attribute instance-attribute
ensemble_member = pt.Field(dtype=(pl.UInt8), description='Ensemble member index (0-based).') class-attribute instance-attribute
h3_index = pt.Field(dtype=(pl.UInt64), description='H3 cell index. The H3 resolution for the nwp_model_id is stored in NwpMetaData.') 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: meters per second.', 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.', ge=0, le=360) class-attribute instance-attribute
wind_speed_100m = pt.Field(dtype=(pl.Float32), description='Wind speed at 100 m. Unit: meters per second.', 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.', 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.UInt8), description="This field is always NaN 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). 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.

They share a de-accumulation step whose known upstream corruption leaves scattered per-pixel nulls beyond lead-0 (and all three are legitimately null at lead-0). Those scattered nulls are tolerated at ingest and reported by :func:assess_nwp_quality; only a whole-slice gap is fatal. See https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

Methods:
all_weather_var_names() classmethod

All meteorological variable field names (continuous + categorical).

Source code in packages/contracts/src/contracts/weather_schemas.py 
251
252
253
254
@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 
256
257
258
259
@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: rejecting whole-slice nulls in de-accumulated variables (scattered nulls are tolerated), enforcing uniqueness, and the ptype-introduction invariant.

Source code in packages/contracts/src/contracts/weather_schemas.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
@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: rejecting whole-slice nulls in de-accumulated variables (scattered
    nulls are tolerated), enforcing uniqueness, and the ptype-introduction invariant."""
    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,
    )

    cls._check_no_whole_null_deaccumulated_slices(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
Source code in packages/contracts/src/contracts/weather_schemas.py 
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
@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``).
    """
    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 scattered nulls in the de-accumulated variables beyond lead-0 — the known upstream ECMWF ENS corruption. A structural whole-slice gap is rejected by :meth:Nwp.validate before this runs, so a validated frame only ever has scatter here.

Source code in packages/contracts/src/contracts/weather_schemas.py 
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
@dataclass(frozen=True)
class NwpQualityReport:
    """Non-fatal data-quality summary for one NWP run.

    Carries the *tolerated* scattered nulls in the de-accumulated variables beyond lead-0 — the
    known upstream ECMWF ENS corruption. A structural whole-slice gap is rejected by
    :meth:`Nwp.validate` before this runs, so a validated frame only ever has scatter here.
    """

    scattered: 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 scattered null cells across all affected slices."""
        return int(self.scattered["n_null"].sum()) if self.scattered.height else 0

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

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

    @property
    def affected_variables(self) -> tuple[str, ...]:
        """The de-accumulated variables carrying scattered nulls, sorted."""
        if not self.scattered.height:
            return ()
        return tuple(sorted(self.scattered["variable"].unique().to_list()))
Attributes
scattered 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 scattered null cells across all affected slices.

n_affected_slices property

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

is_healthy property

True when the run has no unexpected nulls.

affected_variables property

The de-accumulated variables carrying scattered nulls, sorted.

Methods:
__init__(scattered)

Functions:

assess_nwp_quality(dataframe)

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

Reports the scattered per-pixel nulls in the de-accumulated variables (precipitation/radiation) beyond lead-0 — the known upstream ECMWF ENS corruption that :meth:Nwp.validate deliberately tolerates. 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 
454
455
456
457
458
459
460
461
462
463
464
465
466
def assess_nwp_quality(dataframe: pt.DataFrame[Nwp]) -> NwpQualityReport:
    """Summarise the tolerated-but-noteworthy nulls in a *validated* NWP run.

    Reports the scattered per-pixel nulls in the de-accumulated variables (precipitation/radiation)
    beyond lead-0 — the known upstream ECMWF ENS corruption that :meth:`Nwp.validate` deliberately
    tolerates. 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/>.
    """
    scattered = _deaccumulated_null_breakdown(dataframe).filter(
        pl.col("n_null") < pl.col("n_total")
    )
    return NwpQualityReport(scattered=scattered)

contracts.geo_schemas

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. It is used to ensure type safety when passing spatial mapping data 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 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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.
    It is used to ensure type safety when passing spatial mapping data 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

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'] 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 matching the four forecast ranges from the project report.

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 only metric here that is comparable across models with different ensemble sizes.
  • "spread_skill_ratio": Fortin-corrected 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 thirteen 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 they 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 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).

Note: Dynamic features are not explicitly typed as Patito fields below. This is intentional to allow infinite parameterization (e.g., any lag hour) without the overhead of metaprogramming or defining hundreds of static fields. The pipeline dynamically asserts their presence during feature engineering.

Source code in packages/contracts/src/contracts/ml_schemas.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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 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`).

    Note: Dynamic features are not explicitly typed as Patito fields below.
    This is intentional to allow infinite parameterization (e.g., any lag hour)
    without the overhead of metaprogramming or defining hundreds of static fields.
    The pipeline dynamically asserts their presence during feature engineering.
    """

    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))
    ensemble_member: int | None = pt.Field(dtype=pl.UInt8, 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 = pt.Field(dtype=pl.Float32)
    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: int | None = pt.Field(dtype=pl.Int8, allow_missing=True)
    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"]
        )
    )

    @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 uniqueness of the primary key.

        This custom validation step is crucial for preventing weather forecast leakage
        and ensuring data integrity. We enforce uniqueness on the combination of:

        - `time_series_id`: Identifies the specific substation/time series.
        - `power_fcst_init_time`: The initialization time of the power forecast.
        - `valid_time`: The target time for which the forecast is made.
        - `ensemble_member`: The weather forecast ensemble member (if present).

        If `ensemble_member` is present in the dataframe columns, we include it in the
        primary key check to support ensemble forecasts. If it is not present (e.g.,
        for deterministic forecasts or when ensemble members have been aggregated),
        we only check uniqueness across the remaining three columns.

        Args:
            dataframe: The Polars DataFrame to validate.
            columns: Optional list of columns to validate against.
            allow_missing_columns: Whether to allow missing columns.
            allow_superfluous_columns: Whether to allow extra columns.
            drop_superfluous_columns: Whether to drop extra columns.

        Returns:
            A validated Patito DataFrame.

        Raises:
            ValueError: If duplicate entries are found for the primary key.
        """
        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,
        )

        # Define the base primary key columns that must always be unique.
        # This ensures that for any given substation (time_series_id) and target time (valid_time),
        # there is at most one forecast initialized at a specific power_fcst_init_time.
        pk_cols = ["time_series_id", "power_fcst_init_time", "valid_time"]

        if "ensemble_member" in validated_df.columns:
            pk_cols.append("ensemble_member")

        # Check for duplicates. We fail loudly here to prevent downstream training or
        # evaluation on corrupted/duplicated datasets, which could lead to silent bugs.
        if validated_df.select(pk_cols).is_duplicated().any():
            raise ValueError(
                f"Duplicate entries found for primary key columns: {pk_cols}. "
                "This indicates a data integrity issue or potential weather forecast leakage."
            )

        return validated_df
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))) class-attribute instance-attribute
ensemble_member = pt.Field(dtype=(pl.UInt8), 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)) 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 = pt.Field(dtype=(pl.Int8), allow_missing=True) 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
Methods:
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the given dataframe, ensuring uniqueness of the primary key.

This custom validation step is crucial for preventing weather forecast leakage and ensuring data integrity. We enforce uniqueness on the combination of:

  • time_series_id: Identifies the specific substation/time series.
  • power_fcst_init_time: The initialization time of the power forecast.
  • valid_time: The target time for which the forecast is made.
  • ensemble_member: The weather forecast ensemble member (if present).

If ensemble_member is present in the dataframe columns, we include it in the primary key check to support ensemble forecasts. If it is not present (e.g., for deterministic forecasts or when ensemble members have been aggregated), we only check uniqueness across the remaining three columns.

Parameters:

Name Type Description Default
dataframe DataFrame

The Polars DataFrame to validate.

 required
columns Sequence[str] | None

Optional list of columns to validate against.

 None
allow_missing_columns bool

Whether to allow missing columns.

 False
allow_superfluous_columns bool

Whether to allow extra columns.

 False
drop_superfluous_columns bool

Whether to drop extra columns.

 False

Returns:

Type Description
DataFrame[Self]

A validated Patito DataFrame.

Raises:

Type Description
ValueError

If duplicate entries are found for the primary key.
Source code in packages/contracts/src/contracts/ml_schemas.py 
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
@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 uniqueness of the primary key.

    This custom validation step is crucial for preventing weather forecast leakage
    and ensuring data integrity. We enforce uniqueness on the combination of:

    - `time_series_id`: Identifies the specific substation/time series.
    - `power_fcst_init_time`: The initialization time of the power forecast.
    - `valid_time`: The target time for which the forecast is made.
    - `ensemble_member`: The weather forecast ensemble member (if present).

    If `ensemble_member` is present in the dataframe columns, we include it in the
    primary key check to support ensemble forecasts. If it is not present (e.g.,
    for deterministic forecasts or when ensemble members have been aggregated),
    we only check uniqueness across the remaining three columns.

    Args:
        dataframe: The Polars DataFrame to validate.
        columns: Optional list of columns to validate against.
        allow_missing_columns: Whether to allow missing columns.
        allow_superfluous_columns: Whether to allow extra columns.
        drop_superfluous_columns: Whether to drop extra columns.

    Returns:
        A validated Patito DataFrame.

    Raises:
        ValueError: If duplicate entries are found for the primary key.
    """
    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,
    )

    # Define the base primary key columns that must always be unique.
    # This ensures that for any given substation (time_series_id) and target time (valid_time),
    # there is at most one forecast initialized at a specific power_fcst_init_time.
    pk_cols = ["time_series_id", "power_fcst_init_time", "valid_time"]

    if "ensemble_member" in validated_df.columns:
        pk_cols.append("ensemble_member")

    # Check for duplicates. We fail loudly here to prevent downstream training or
    # evaluation on corrupted/duplicated datasets, which could lead to silent bugs.
    if validated_df.select(pk_cols).is_duplicated().any():
        raise ValueError(
            f"Duplicate entries found for primary key columns: {pk_cols}. "
            "This indicates a data integrity issue or potential weather forecast leakage."
        )

    return validated_df

Metrics

Bases: Model

Evaluation metrics for power forecasts — tall format.

One row per (time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name, metric_param). 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 1 all mae all 5.2
1 1 day_ahead rmse all 7.1
1 1 day_ahead pinball_loss p10 2.1
1 1 day_ahead pinball_loss p50 3.4
1 1 day_ahead mean_pinball_loss all 2.4
1 1 day_ahead picp p10_p90 0.78

Primary key: (time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name, metric_param).

Source code in packages/contracts/src/contracts/ml_schemas.py 
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
class Metrics(pt.Model):
    """Evaluation metrics for power forecasts — tall format.

    One row per `(time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name,
    metric_param)`. `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              | 1       | all           | mae               | all          | 5.2          |
    | 1              | 1       | day_ahead     | rmse              | all          | 7.1          |
    | 1              | 1       | day_ahead     | pinball_loss      | p10          | 2.1          |
    | 1              | 1       | day_ahead     | pinball_loss      | p50          | 3.4          |
    | 1              | 1       | day_ahead     | mean_pinball_loss | all          | 2.4          |
    | 1              | 1       | day_ahead     | picp              | p10_p90      | 0.78         |

    Primary key: `(time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name,
    metric_param)`.
    """

    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 them cast-free and lets
    # predicate pushdown work. See the "declare Delta filter/partition columns as String" gotcha:
    # ../../../../CLAUDE.md#delta-lake-dictionary-encoded-columns-declare-delta-filterpartition-columns-as-string
    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="CV fold year (e.g. '2022'), 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."
        ),
    )
    """The columns from `evaluation_scope` and below are populated by the ``metrics`` Dagster asset.
    They are ``allow_missing`` so that the pure ``compute_metrics()`` helper can emit the core
    metric rows and have the asset enrich them with scope/window provenance before the frame is
    written to the ``forecast_metrics`` Delta table."""

    time_series_type: str | None = pt.Field(
        dtype=pl.Enum(TIME_SERIES_TYPE_SLICES),
        allow_missing=True,
        description=(
            "The time-series category for this row. Null when metadata is unavailable for a series."
        ),
    )

    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 written (provenance; orders the append-only "
            "monitoring series and distinguishes recomputations)."
        ),
    )

    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."
        ),
    )
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="CV fold year (e.g. '2022'), 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

The columns from evaluation_scope and below are populated by the metrics Dagster asset. They are allow_missing so that the pure compute_metrics() helper can emit the core metric rows and have the asset enrich them with scope/window provenance before the frame is written to the forecast_metrics Delta table.

time_series_type = pt.Field(dtype=(pl.Enum(TIME_SERIES_TYPE_SLICES)), allow_missing=True, description='The time-series category for this row. Null when metadata is unavailable for a series.') 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 written (provenance; orders the append-only monitoring series and distinguishes recomputations).') 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

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 and cv_power_forecasts. Eligibility is a function of data coverage and the fold dates only — never the model or experiment config — so every experiment 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 
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
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`` and ``cv_power_forecasts``. Eligibility is a function of
    data coverage and the fold dates **only** — never the model or experiment config — so every
    experiment 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.hydra_schemas

Hydra configuration schemas for the NGED substation forecast project.

Attributes

Classes

CvFoldConfig

Bases: BaseModel

Configuration for a single expanding-window CV fold.

leaderboard distinguishes the epoch-pinned leaderboard folds (the apples-to-apples evaluation protocol) from optional non-leaderboard dev folds such as smoke_test: 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/hydra_schemas.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class CvFoldConfig(BaseModel):
    """Configuration for a single expanding-window CV fold.

    ``leaderboard`` distinguishes the epoch-pinned leaderboard folds (the apples-to-apples
    evaluation protocol) from optional non-leaderboard dev folds such as ``smoke_test``: 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.

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 only included if it has at least this many months of data before val_start (and data through val_end).

Source code in packages/contracts/src/contracts/hydra_schemas.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class CvConfig(BaseModel):
    """Configuration for expanding-window cross-validation.

    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 only included if it has at least this many 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. ``["2022", "2023", ...]``).

        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. ``"2022"``).

        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. ["2022", "2023", ...]).

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. "2022").

 required

Raises:

Type Description
KeyError

If no fold with that id exists in the config.
Source code in packages/contracts/src/contracts/hydra_schemas.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
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. ``"2022"``).

    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:

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/hydra_schemas.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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``.
    """
    raw = OmegaConf.to_container(OmegaConf.load(path), resolve=True)
    return CvConfig.model_validate(raw)

contracts.settings

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/, metadata/, data/, .env).

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

Classes

DataQualitySettings

Bases: BaseSettings

Settings for data quality thresholds in substation flow processing.

These thresholds are used to identify problematic telemetry data:

  • stuck_std_threshold: When the rolling standard deviation falls below this value (across stuck_window_periods), the sensor is likely stuck. We replace such values with null to preserve the temporal grid. A value of 0.01 MW was chosen because substations with normal operation typically have much higher variability.

  • max_mw_threshold: Active power above this value is considered physically unrealistic for primary substations in the NGED portfolio. A threshold of 150 MW was chosen because typical primary substations operate in the tens of MW range, and values exceeding 100 MW are extremely rare anomalies.

  • min_mw_threshold: Active power below this value is potentially erroneous (negative values can occur at times of high renewable generation). A threshold of -50.0 MW was chosen to allow for reverse power flow during high renewable generation periods while still catching implausible extreme negative values.

Centralizing these in Settings allows them to be configurable per environment (dev/staging/prod) while preventing logic drift between asset checks and data cleaning steps. All code that references these thresholds should import them from here, not define them locally.

Source code in packages/contracts/src/contracts/settings.py 
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
class DataQualitySettings(BaseSettings):
    """Settings for data quality thresholds in substation flow processing.

    These thresholds are used to identify problematic telemetry data:

    - `stuck_std_threshold`: When the rolling standard deviation falls below this value
      (across `stuck_window_periods`), the sensor is likely stuck. We replace such
      values with null to preserve the temporal grid. A value of 0.01 MW was chosen
      because substations with normal operation typically have much higher variability.

    - `max_mw_threshold`: Active power above this value is considered physically
      unrealistic for primary substations in the NGED portfolio. A threshold of 150 MW
      was chosen because typical primary substations operate in the tens of MW range,
      and values exceeding 100 MW are extremely rare anomalies.

    - `min_mw_threshold`: Active power below this value is potentially erroneous
      (negative values can occur at times of high renewable generation). A threshold of
      -50.0 MW was chosen to allow for reverse power flow during high renewable
      generation periods while still catching implausible extreme negative values.

    Centralizing these in Settings allows them to be configurable per environment
    (dev/staging/prod) while preventing logic drift between asset checks and data
    cleaning steps. All code that references these thresholds should import them from
    here, not define them locally.
    """

    stuck_std_threshold: float = 0.01
    stuck_window_periods: int = 48  # Each period is 30 minutes.
    max_mw_threshold: float = 150.0
    min_mw_threshold: float = -50.0
Attributes
stuck_std_threshold = 0.01 class-attribute instance-attribute
stuck_window_periods = 48 class-attribute instance-attribute
max_mw_threshold = 150.0 class-attribute instance-attribute
min_mw_threshold = -50.0 class-attribute instance-attribute

Settings

Bases: BaseSettings

Configuration settings for the NGED substation forecast project.

Source code in packages/contracts/src/contracts/settings.py 
 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
class Settings(BaseSettings):
    """Configuration settings for the NGED substation forecast project."""

    mlflow_tracking_uri: str = Field(
        default="sqlite:///mlflow.db",
        description=(
            "MLflow tracking URI. Centralized 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."
        ),
    )

    data_quality: DataQualitySettings = Field(
        default_factory=DataQualitySettings,
        description="Configurable thresholds for data quality checks.",
    )

    nged_s3_bucket_url: str = Field(
        ..., description="NGED S3 bucket URL. Typically stored in the .env file."
    )
    nged_s3_bucket_access_key: str = Field(
        ..., description="Access key for the NGED S3 bucket. Typically stored in the .env file."
    )
    nged_s3_bucket_secret: str = Field(
        ..., description="Secret key for the NGED S3 bucket. Typically stored in the .env file."
    )

    def get_nged_s3_store(self) -> obstore.store.S3Store:
        """Returns an initialized obstore.store.S3Store instance for the NGED bucket."""
        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."
        ),
    )
    nwp_metadata_csv_path: Path = Field(
        default=PROJECT_ROOT / "metadata" / "nwp_metadata.csv",
        description=(
            "Static CSV of per-NWP-model metadata (H3 resolution, provider, ensemble flag),"
            " checked into the repo and read by NwpMetaData.load. A code-relative resource"
            " (like cv_config_path), so it stays a local Path even when the data tables are a"
            " remote URI."
        ),
    )

    # --- Storage roots -------------------------------------------------------------------
    #
    # data_path_internal and data_path_delivery hold the (S3-capable) data tables; local_
    # artifacts_path holds the always-local model cache and production model. Why they
    # 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 (model cache, 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 empty by default; unset on AWS (object_store auto-discovers the IAM-role credentials),
    # set only for a dev/MinIO endpoint. The AWS/dev split, and how these differ from the
    # nged_s3_bucket_* source-bucket creds 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 — object_store auto-discovers the Fargate task's IAM-role credentials and
        region — and empty for a local data-path root (delta-rs ignores it 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 ``ty`` checks each key here, where they are authored; widen it 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, fold_id).

    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 and cv_power_forecasts so every experiment scores a fold on the"
            " identical, experiment-independent population."
        ),
    )
    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 GB boundary."""

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

    model_cache_base_path: str = Field(
        default="",
        description=(
            "Root of the local-disk model cache, keyed by MLflow run ID, used by"
            " BaseForecaster.load_from_mlflow. Currently only the CV pipeline"
            " (cv_power_forecasts) reads a model back through this cache; production"
            " inference does not use it for v0.1 (the model is baked directly into the"
            " container image instead)."
        ),
    )
    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 DSN disables Sentry entirely) ---

    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 (see docs/roadmap/delivery-tables.md) derive from
        # data_path_delivery instead, so a new delivery table can't silently land in the
        # internal bucket by inheriting the default derivation.
        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.model_cache_base_path = self.model_cache_base_path or uri_join(
            self.local_artifacts_path, "model_cache"
        )
        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."""
        url_adapter.validate_python(v)
        return v
Attributes
mlflow_tracking_uri = Field(default='sqlite:///mlflow.db', description='MLflow tracking URI. Centralized 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
data_quality = Field(default_factory=DataQualitySettings, description='Configurable thresholds for data quality checks.') class-attribute instance-attribute
nged_s3_bucket_url = Field(..., description='NGED S3 bucket URL. Typically stored in the .env file.') class-attribute instance-attribute
nged_s3_bucket_access_key = Field(..., description='Access key for the NGED S3 bucket. Typically stored in the .env file.') class-attribute instance-attribute
nged_s3_bucket_secret = Field(..., description='Secret key for the NGED 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
nwp_metadata_csv_path = Field(default=(PROJECT_ROOT / 'metadata' / 'nwp_metadata.csv'), description='Static CSV of per-NWP-model metadata (H3 resolution, provider, ensemble flag), checked into the repo and read by NwpMetaData.load. A code-relative resource (like cv_config_path), so it stays a local Path even when the data tables are a remote URI.') 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 (model cache, 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 — object_store auto-discovers the Fargate task's IAM-role credentials and region — and empty for a local data-path root (delta-rs ignores it 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 ty checks each key here, where they are authored; widen it 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, fold_id).

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 and cv_power_forecasts so every experiment scores a fold on the identical, experiment-independent population.') 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 GB boundary.

model_cache_base_path = Field(default='', description='Root of the local-disk model cache, keyed by MLflow run ID, used by BaseForecaster.load_from_mlflow. Currently only the CV pipeline (cv_power_forecasts) reads a model back through this cache; production inference does not use it for v0.1 (the model is baked directly into the container image instead).') class-attribute instance-attribute
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:
get_nged_s3_store()

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

Source code in packages/contracts/src/contracts/settings.py 
112
113
114
115
116
117
118
119
120
def get_nged_s3_store(self) -> obstore.store.S3Store:
    """Returns an initialized obstore.store.S3Store instance for the NGED bucket."""
    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.

Source code in packages/contracts/src/contracts/settings.py 
372
373
374
375
376
377
@field_validator("nged_s3_bucket_url")
@classmethod
def validate_url(cls, v: str) -> str:
    """Validate that the S3 bucket URL is a valid URL."""
    url_adapter.validate_python(v)
    return v

Functions:

get_settings() cached

Return the shared, lazily-constructed Settings singleton.

Prefer this over constructing Settings() at module import time. Settings has required fields (the nged_s3_bucket_* credentials) with no defaults, so instantiation reads .env and raises ValidationError when those are absent. Deferring construction to first use keeps library modules (e.g. the contracts schemas) importable without live credentials — so type-checkers, doc builders, and tests that only touch in-memory frames don't need a populated .env — while still failing fast the moment settings are actually needed.

Cached with lru_cache so every caller shares one instance (matching the previous module-level singletons). 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 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
@lru_cache
def get_settings() -> Settings:
    """Return the shared, lazily-constructed ``Settings`` singleton.

    Prefer this over constructing ``Settings()`` at module import time. ``Settings`` has required
    fields (the ``nged_s3_bucket_*`` credentials) with no defaults, so instantiation reads ``.env``
    and raises ``ValidationError`` when those are absent. Deferring construction to first *use*
    keeps library modules (e.g. the ``contracts`` schemas) importable without live credentials —
    so type-checkers, doc builders, and tests that only touch in-memory frames don't need a
    populated ``.env`` — while still failing fast the moment settings are actually needed.

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