data.demo_loader

data.demo_loader

Demo data loader for safety-critical forecasting tasks.

This module provides flexible data loading functions for ground truth validation in forecasting demonstrations and production workflows.

Functions

Name Description
load_actual_combined Load ground truth and compute combined actual series with validation.

load_actual_combined

data.demo_loader.load_actual_combined(
    config,
    columns,
    forecast_horizon=None,
    weights=None,
    data_path=None,
)

Load ground truth and compute combined actual series with validation.

This function loads a CSV file containing ground truth data, validates the presence of required columns, extracts a subset based on forecast horizon, and aggregates multiple columns using weighted averaging.

Parameters

Name Type Description Default
config ConfigDemo Configuration object containing default paths and parameters. required
columns List[str] List of column names to extract from the ground truth data. required
forecast_horizon Optional[int] Number of time steps to extract. If None, uses config.forecast_horizon. Must be positive. None
weights Optional[List[float]] Weights for aggregating columns. If None, uses config.weights. Length must match number of columns. None
data_path Optional[Path] Path to the ground truth CSV file. If None, uses config.data_path. File must exist. None

Returns

Name Type Description
pd.Series Aggregated time series combining all columns with specified weights.

Raises

Name Type Description
FileNotFoundError If the ground truth file does not exist.
ValueError If required columns are missing from the data or if weights length doesn’t match columns.

Examples

Basic usage with explicit forecast horizon and weights:

import tempfile
from pathlib import Path

import pandas as pd

from spotforecast2_safe.configurator import ConfigDemo
from spotforecast2_safe.data.demo_loader import load_actual_combined

idx = pd.date_range("2020-01-01", periods=3, freq="h", name="timestamp")
sample_df = pd.DataFrame({"col1": [1.0, 3.0, 5.0], "col2": [2.0, 4.0, 6.0]}, index=idx)

with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
    sample_df.to_csv(f.name)
    temp_path = Path(f.name)

config = ConfigDemo(data_path=temp_path)
result = load_actual_combined(
    config, columns=["col1", "col2"], forecast_horizon=2, weights=[1.0, 1.0]
)
print(f"Result length: {len(result)}")
print(f"First value: {result.iloc[0]:.1f}")
temp_path.unlink()
Result length: 2
First value: 3.0

Override forecast horizon and weights:

idx = pd.date_range("2020-01-01", periods=10, freq="h", name="timestamp")
sample_df = pd.DataFrame(
    {"col1": [float(i) for i in range(10)], "col2": [float(i * 2) for i in range(10)]},
    index=idx,
)

with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
    sample_df.to_csv(f.name)
    temp_path = Path(f.name)

config = ConfigDemo(data_path=temp_path, forecast_horizon=24)
result = load_actual_combined(
    config, columns=["col1", "col2"], forecast_horizon=5, weights=[1.0, 0.5]
)
print(f"Custom horizon length: {len(result)}")
temp_path.unlink()
Custom horizon length: 5

Override data_path while keeping the rest of the config:

idx = pd.date_range("2020-01-01", periods=1, freq="h", name="timestamp")
sample_df = pd.DataFrame({"X": [100.0], "Y": [200.0]}, index=idx)

with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
    sample_df.to_csv(f.name)
    custom_path = Path(f.name)

config = ConfigDemo()  # default data_path
result = load_actual_combined(
    config,
    columns=["X", "Y"],
    data_path=custom_path,
    forecast_horizon=1,
    weights=[0.5, 0.5],
)
print(f"Custom path result: {result.iloc[0]:.1f}")
custom_path.unlink()
Custom path result: 150.0

Error handling — missing file:

config = ConfigDemo(data_path=Path("/nonexistent/file.csv"))
try:
    load_actual_combined(
        config, columns=["A"], forecast_horizon=1, weights=[1.0]
    )
except FileNotFoundError:
    print("File not found as expected")
File not found as expected

Error handling — missing columns:

idx = pd.date_range("2020-01-01", periods=1, freq="h", name="timestamp")
sample_df = pd.DataFrame({"col1": [1.0]}, index=idx)

with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
    sample_df.to_csv(f.name)
    temp_path = Path(f.name)

config = ConfigDemo(data_path=temp_path)
try:
    load_actual_combined(
        config,
        columns=["col1", "col2"],
        forecast_horizon=1,
        weights=[1.0, 1.0],
    )
except ValueError:
    print("Missing columns detected")
temp_path.unlink()
Missing columns detected

Production usage with horizon and weights drawn from the config:

idx = pd.date_range("2020-01-01", periods=30, freq="h", name="timestamp")
sample_df = pd.DataFrame(
    {
        "load": [100 + i for i in range(30)],
        "solar": [50 + i for i in range(30)],
        "wind": [25 + i for i in range(30)],
    },
    index=idx,
)

with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
    sample_df.to_csv(f.name)
    temp_path = Path(f.name)

config = ConfigDemo(
    data_path=temp_path, forecast_horizon=24, weights=[1.0, -0.5, -0.5]
)
result = load_actual_combined(config, columns=["load", "solar", "wind"])
print(f"Production forecast length: {len(result)}")
print(f"Result is pandas Series: {isinstance(result, pd.Series)}")
temp_path.unlink()
Production forecast length: 24
Result is pandas Series: True