data_processor

Core processing, transformation, and forecasting logic for CFS data.

This module is the heart of the src package. It turns raw CFS GRIB files into lake-averaged variables and runs them through trained models to produce NBS forecasts. The main components are:

• CFSProcessor — read CFS GRIB files, remap to the basin mask, and store lake/land-averaged precipitation, temperature, and evaporation.

• SeasonalCycleProcessor — fit a monthly climatology and convert between absolute values and anomalies (including lead-wide helpers).

• CFSTransformer — reshape and feature-engineer the processed data (filtering, lag/lead shifts, wide-format structuring, time features).

• CNBSForecaster — load trained models and scalers and generate forecasts in either absolute or anomaly mode.

• ForecastTransformer — reshape forecast output between wide and long/tidy formats and filter by forecast month.

• CEPCalculator — compute Climatology Exceedance Probabilities from forecasts and probability-of-exceedance curves.

• calculate_acc_variables() — accumulate forecast values over multi-month windows.

Heavy dependencies (cfgrib, netCDF4, xarray, scikit-learn, properscoring) are imported at module scope and mocked when building the documentation.

class src.data_processor.CFSProcessor(database, table)

Bases: object

Process downloaded CFS GRIB files into lake-averaged variables.

Reads CFS forecast GRIB2 files for a run, remaps the relevant fields (precipitation, 2-metre air temperature, evaporation from latent heat flux) onto a basin mask grid, computes area-weighted lake/land averages, and writes the results to the forecast database via CFSDatabase.

Parameters:

• database (str) – Path to the SQLite database file used for output.

• table (str) – Name of the table to write processed values into.

__init__(database, table)

Initialize the processor with database path and table name.

process_files(download_dir, mask_file, mask_variables)

Process CFS GRIB files and insert lake-averaged variables into the database.

This function loops through downloaded CFS GRIB files for a forecast run, extracts relevant variables, remaps them to the mask grid, calculates lake- or land-area weighted averages, and stores the results in the forecast database.

Processed variables include:

• precipitation from pgbf files

• 2-meter air temperature from flxf files

• evaporation from latent heat flux in flxf files

Parameters:

• download_dir (str) – Directory containing downloaded CFS GRIB files.

• mask_file (str) –

  Path to the NetCDF mask file. The mask file must include:

  • latitude

  • longitude

  • lake/land mask variables listed in mask_variables

• mask_variables (list of str) –

  List of mask variable names to process.

  Expected format:

  ”{lake_abbreviation}_{surface_type}”

  Examples:

  • ”sup_lake”

  • ”sup_land”

  • ”mih_lake”

  • ”eri_land”

  Valid lake abbreviations are:

  • ”sup” -> “superior”

  • ”mih” -> “michigan-huron”

  • ”eri” -> “erie”

  • ”ont” -> “ontario”

Returns:

Results are inserted directly into the database using self.db.add().

Return type:

None

class src.data_processor.SeasonalCycleProcessor

Bases: object

Computes and applies a monthly climatology to a pandas DataFrame with a DatetimeIndex at monthly frequency.

This class supports:

• fit(): compute monthly climatology from a baseline period

• transform(): convert raw values -> anomalies

• inverse_transform(): convert anomalies -> raw values

• save()/load(): persist climatology artifact to disk

Assumptions

• Input DataFrames must have a pandas.DatetimeIndex.

• Index must represent monthly timestamps.

• Monthly climatology is computed as the mean for each calendar month (1–12).

fit(df: DataFrame, var_list: Sequence[str] | None = None, baseline_time: slice | None = None, baseline_definition: Dict | None = None) → SeasonalCycleProcessor

Compute monthly climatology from a baseline period.

Parameters:

• df (pandas.DataFrame) – Input DataFrame with a DatetimeIndex. Rows represent monthly observations. Columns represent variables.

• var_list (sequence of str, optional) – Subset of columns to compute climatology for. If None, all numeric columns are used.

• baseline_time (slice, optional) – Time slice applied before computing climatology, e.g. slice("1981-01-01", "2008-12-01"). If None, the full DataFrame is used.

• baseline_definition (dict, optional) – Metadata describing baseline choice (for reproducibility), e.g. {"train_start": "1981-01-01", "train_end": "2008-12-01"}.

Returns:

self – Fitted processor with climatology stored internally.

Return type:

SeasonalCycleProcessor

transform(df: DataFrame) → DataFrame

Convert raw values to anomalies.

Parameters:

df (pandas.DataFrame) – DataFrame with DatetimeIndex. Must contain columns used during fit().

Returns:

anomalies – DataFrame with same shape as input, where climatological monthly means have been subtracted.

Return type:

pandas.DataFrame

inverse_transform(df: DataFrame) → DataFrame

Convert anomalies back to raw values.

Parameters:

df (pandas.DataFrame) – Anomaly DataFrame with DatetimeIndex.

Returns:

raw – DataFrame with climatology added back.

Return type:

pandas.DataFrame

save(base_dir: str = 'seasonal_cycles') → Dict[str, str]

Save climatology artifact to disk.

Parameters:

base_dir (str) – Directory where artifact files will be written.

Returns:

Dictionary containing the keys "climatology_path" and "metadata_path", each mapping to the path of the written file.

Return type:

dict

classmethod load(climatology_path: str, metadata_path: str) → SeasonalCycleProcessor

Load a previously saved climatology artifact.

Parameters:

• climatology_path (str) – Path to saved climatology CSV.

• metadata_path (str) – Path to saved metadata JSON.

Returns:

Processor with climatology loaded.

Return type:

SeasonalCycleProcessor

static build_climatology_variable_name_long(df: DataFrame, *, lake_col: str = 'lake', surface_col: str = 'surface', component_col: str = 'component') → Series

Build climatology variable names for long-format operational CFS forecast data.

Expected output names match training climatology columns, e.g.:

superior_lake_precipitation michigan-huron_land_air_temperature

static subtract_climatology_long(df: DataFrame, scp: SeasonalCycleProcessor, *, value_col: str = 'value', month_col: str = 'month', lake_col: str = 'lake', surface_col: str = 'surface_type', component_col: str = 'component', output_col: str = 'value_anom', variable_col: str = 'climatology_variable', strict: bool = True) → DataFrame

Convert absolute long-format operational CFS forecast values to anomalies.

This is intended for data from cfs_forecast_data.db, where each row contains:

cfs_run, year, month, lake, surface_type, component, value

Unlike lead-wide training data, the operational table already contains the verifying forecast month in month, so no init-date + lead calculation is needed.

Operation:

value_anom = value - climatology[month, variable]

where:

variable = f”{lake}_{surface_type}_{component}”

static subtract_climatology_leadwide(df_abs_leadwide: DataFrame, scp_X, strict: bool = False) → DataFrame

Convert lead-wide absolute forecast values into climatological anomalies by subtracting the monthly climatology corresponding to each forecast verifying month.

This function is designed for datasets containing multiple forecast lead months stored as separate columns using the naming convention:

<variable>_mo{k}

where:

• <variable> is the base predictor variable name

• k is the forecast lead month

Examples

superior_precipitation_mo0 superior_precipitation_mo1 erie_runoff_mo3

For each column:

1. The forecast lead month is extracted using the _mo{k} suffix.

2. The base variable name is identified by removing the suffix.

3. The verifying month is computed by shifting the initialization date forward by the forecast lead.

4. The corresponding monthly climatology is subtracted from the forecast value.

The verifying month is calculated as verifying_month = initialization_date + forecast_lead_months.

This ensures that climatology subtraction is aligned with the actual target month being forecasted rather than the initialization month.

If a column does not contain a _mo{k} suffix, it is treated as lead 0.

Parameters:

• df_abs_leadwide (pd.DataFrame) – DataFrame containing absolute forecast values indexed by forecast initialization date. The index must be a DatetimeIndex.

• scp_X (SeasonalCycleProcessor) – A fitted SeasonalCycleProcessor object containing a climatology attribute. The climatology DataFrame must be indexed by month (1-12) and contain columns matching the base variable names.

• strict (bool, optional) – If True, raise a KeyError when a climatology variable is not found. If False, missing variables are skipped with a warning message. Default is False.

Returns:

DataFrame with the same shape, index, and columns as the input, but with values transformed from absolute space into anomaly space by subtracting the monthly climatology.

Return type:

pd.DataFrame

Raises:

• ValueError – If df_abs_leadwide does not have a DatetimeIndex.

• AttributeError – If scp_X does not contain a climatology attribute.

• KeyError – If strict=True and a base variable is not found in the climatology columns.

static add_climatology_back_leadwide(df_anom_leadwide: DataFrame, scp_y: SeasonalCycleProcessor, strict: bool = False) → DataFrame

Convert lead-wide anomaly targets to lead-wide absolute targets by adding a monthly climatology.

Parameters:

• df_anom_leadwide (pd.DataFrame) – Anomaly DataFrame indexed by init date (must be a DatetimeIndex).

• scp_y (SeasonalCycleProcessor) – A fitted seasonal-cycle processor.

• strict (bool) – If True, raise error if column does not match _mo{k}.

Returns:

Same shape/columns/index as df_anom_leadwide, but in absolute units.

Return type:

pd.DataFrame

static load_clim(directory)

Load SeasonalCycleProcessor objects for inputs and targets from a directory.

Parameters:

directory (str) – Base directory containing inputs and targets subfolders with climatology CSVs and metadata JSONs.

Returns:

• scp_X (SeasonalCycleProcessor) – Processor for input features.

• scp_y (SeasonalCycleProcessor) – Processor for target outputs.

class src.data_processor.CFSTransformer(df)

Bases: object

Reshape and feature-engineer processed CFS data for modelling.

Wraps a pandas.DataFrame of processed CFS values and provides transformations used to build the model input matrix: filtering to recent runs (filter()), creating lagged/lead columns (shift_variables()), pivoting long data into the wide per-lead feature layout (structure_input()), and adding cyclical month and time-trend features (add_time_features()).

Parameters:

df (pandas.DataFrame) – The processed CFS data to transform. Copied on construction.

__init__(df)

Initialize the transformer with a pandas DataFrame.

filter(date=None, months_back=9)

Filter rows to keep only CFS runs going back a specified number of months from the current operational forecast month.

Rules

• If date is before the 26th, use the current month.

• If date’s day is on or after the 26th, use the following month.

• Then go back months_back months.

• Keep rows where cfs_run is greater than or equal to that start date.

param date:

Reference date. Defaults to current date if None.

type date:

datetime, optional

param months_back:

Number of months to go back from the first forecast month.

type months_back:

int, default=9

returns:

Filtered dataframe.

rtype:

pd.DataFrame

shift_variables(lag=0, lead=0)

Create the variables columns to include lags (last month values) and lead variables

Parameters: - df (pd.DataFrame): The DataFrame containing the time series data. - lag (int): The number of months you want to include lagged variables. Default = 0 - lead (int): The number of months for the advance variables. Default = 0

Returns: - pd.DataFrame: The DataFrame with added variable columns for lags and leading.

structure_input()

Transforms a long-format CFS forecast DataFrame into a wide-format one.

add_time_features(add_month_cycle=True, add_time_trend=True, normalize_time=True, overwrite=True, copy=True)

Add cyclical month features and a continuous time trend.

Parameters:

• df (pandas.DataFrame) – Input dataframe indexed by a pandas DatetimeIndex.

• add_month_cycle (bool, default=True) –

  If True, add cyclical month features:

  • month_sin

  • month_cos

  These preserve the cyclical nature of calendar months (e.g., December is close to January).

• add_time_trend (bool, default=True) – If True, add a continuous decimal-year time trend feature.

• normalize_time (bool, default=True) –

  If True, standardize the time feature to:

  mean = 0 standard deviation = 1

  This improves numerical stability for many machine learning models.

• overwrite (bool, default=True) –

  If True, remove any pre-existing:

  • month_sin

  • month_cos

  • time

  before adding new versions.

• copy (bool, default=True) – If True, operate on a copy of the dataframe.

Returns:

Dataframe with added temporal predictor columns.

Return type:

pandas.DataFrame

Notes

The cyclical month encoding is defined as:

month_sin = sin(2π * month / 12) month_cos = cos(2π * month / 12)

which avoids discontinuities between December and January.

The continuous time trend is represented as:

year + (month - 1) / 12

and optionally normalized.

class src.data_processor.CNBSForecaster(model_dir, scaler_dir, mode='actual')

Bases: object

Load trained models and scalers and generate NBS forecasts.

On construction, loads the input/output scalers and all trained models found in the given directories, selecting the _anom artifacts when mode='anomaly'. predict() runs a named model over a feature matrix and returns a forecast DataFrame, converting anomalies back to absolute values when operating in anomaly mode.

Parameters:

• model_dir (str) – Directory containing the trained model .joblib files.

• scaler_dir (str) – Directory containing the input/output scaler .joblib files.

• mode (str, default "actual") – Either "actual" (absolute values) or "anomaly" (anomalies relative to climatology).

__init__(model_dir, scaler_dir, mode='actual')

Initialize the ForecastModel class.

Parameters:

• model_dir (str)

• scaler_dir (str)

• mode (str) – “actual” or “anomaly”

predict(X, model_name, scp_y=None)

Predict CNBS values.

Parameters:

• X (pd.DataFrame)

• model_name (str)

• scp_y (object, optional) – Required if mode=’anomaly’ to convert anomalies back to absolute

Return type:

pd.DataFrame

class src.data_processor.ForecastTransformer(df)

Bases: object

Transforms CFS forecast data from wide to long format, and filters based on forecast month.

melt()

Transforms the forecast DataFrame into long format and then pivots it by lake, component, and forecast month for easier analysis or storage.

pivot()

Reshape forecast DataFrame so each lake & variable type gets its own column with forecast_month calculated.

filter(first_forecast_month)

Filters rows to keep the first forecast month >= first_forecast_month and all subsequent months.

Works whether dataframe has ‘forecast_month’ (YYYY-MM) or ‘year’ + ‘month’.

class src.data_processor.CEPCalculator(prob)

Bases: object

Calculate Climatology Exceedance Probabilities (CEP) from forecast data and probability exceedance curves.

__init__(prob)

Parameters:

prob (pd.DataFrame) –

Probability dataframe containing:

• month

• lake

• prob_exceedance

• value

If already aligned, it may also contain:

• date

align_prob_with_start_date(start_date)

Align probability dataframe months to a forecast start date.

Parameters:

start_date (str or datetime-like) –

Forecast start date. Accepts:

• ”06-2026”

• ”2026-06”

• ”2026-06-01”

Returns:

Aligned probability dataframe with a date column.

Return type:

pandas.DataFrame

calculate_cep(df, date_col='forecast_month', component='nbs', output_file=None, sep=',')

Create a reduced dataframe containing forecast values and climatology exceedance probability (CEP).

Parameters:

• df (pandas.DataFrame) –

  Input forecast dataframe containing:

  • date_col

  • model

  • lake

  • component

• date_col (str, default='forecast_month') – Column used to determine forecast month.

• component (str, default='nbs') – Forecast component to evaluate.

• output_file (str or None, default=None) – Optional path to save output file.

• sep (str, default=',') – Delimiter used if output_file is provided.

Returns:

Columns:

date, forecast_month, model, lake, component, cep

Return type:

pandas.DataFrame

src.data_processor.calculate_acc_variables(df, accumulation_periods=(3, 6), exclude_columns=None)

Calculate accumulated forecast values for each CFS run and model.

For each unique cfs_run and model, this function sorts the forecast data by forecast_month and sums the first N forecast months for each requested accumulation period.

For example, accumulation_periods=(1, 3, 6) will create:

• variable_acc1 = month 1

• variable_acc3 = months 1-3 summed

• variable_acc6 = months 1-6 summed

Parameters:

• df (pandas.DataFrame) –

  Forecast dataframe containing one row per forecast month. Must include:

  • cfs_run

  • model

  • forecast_month

  All non-excluded columns are treated as forecast value columns.

• accumulation_periods (tuple of int, default=(3, 6)) – Number of forecast months to accumulate.

• exclude_columns (list, set, tuple, or None, default=None) – Additional columns to exclude from accumulation calculations.

Returns:

Dataframe with one row per cfs_run and model, containing only:

• cfs_run

• model

• accumulated forecast variables

Return type:

pandas.DataFrame