make_exogenous_classification¶

def make_exogenous_classification(
    *,
    n_samples: int = 400,
    forecasting_horizon: int = 6,
    noise: float = 0.1,
    forecast_bias: float = 0.3,
    random_state: int = 42,
) -> Bunch:
    """Generate a synthetic classification dataset with exogenous features.

    Creates hourly air quality readings classified into three categories
    based on pollutant concentration thresholds.

    Three exogenous feature types are produced:

    - **X_actual** (observation features): realized pollutant readings
      with a 24 hour sinusoidal cycle.
    - **X_future** (known future): an ``is_holiday`` indicator drawn from a
      fixed civil holiday calendar, covering the full time range. Holidays
      reduce traffic and therefore pollutant levels. The dates are irregular
      with respect to weekday, so the indicator cannot be derived from the
      timestamp and genuinely requires the ``X_future`` channel.
    - **X_forecast** (external forecasts): pollutant concentration
      forecasts with one vintage per observation from index
      ``forecasting_horizon`` up to (but not including) the last
      observation, each covering the next ``forecasting_horizon`` steps
      that remain within the sample. The final observation produces no
      vintage because all of its forecast steps fall outside the sample
      range.

    Classification thresholds are applied to the *effective* pollutant signal
    (the stored ``X_actual`` ``pollutant`` minus a 5-unit weekend offset, plus
    jitter), not to the raw ``X_actual["pollutant"]`` feature. As a result a
    small fraction of rows near the 40/60 boundaries do not satisfy these
    thresholds when read off the stored feature:

    - ``"good"``: effective pollutant < 40
    - ``"moderate"``: 40 <= effective pollutant < 60
    - ``"poor"``: effective pollutant >= 60

    Parameters
    ----------
    n_samples : int, default=400
        Number of hourly observations.
    forecasting_horizon : int, default=6
        Number of forward steps per X_forecast vintage.
    noise : float, default=0.1
        Standard deviation of an additional jitter layered on top of the
        pollutant signal just before classification. The underlying pollutant
        signal already carries a fixed ``std=5.0`` noise term, so at the
        default ``noise=0.1`` this jitter only marginally perturbs the
        decision boundary; raise it to inject more label noise.
    forecast_bias : float, default=0.3
        Systematic bias added to pollutant forecasts relative to actuals.
    random_state : int, default=42
        Seed for reproducibility.

    Returns
    -------
    Bunch
        Dictionary-like object with the following attributes:

        y : pl.DataFrame
            Target with columns ``["time", "air_quality"]``. Values are
            one of ``"good"``, ``"moderate"``, ``"poor"``.
        X_actual : pl.DataFrame
            Observation features with columns ``["time", "pollutant"]``.
        X_future : pl.DataFrame
            Known future features with columns ``["time", "is_holiday"]``.
        X_forecast : pl.DataFrame
            External forecasts with columns
            ``["vintage_time", "time", "pollutant_forecast"]``.
        frame : pl.DataFrame
            ``y``, ``X_actual``, and ``X_future`` joined on ``"time"``.
        feature_names : list of str
            ``["pollutant", "is_holiday", "pollutant_forecast"]``.
        target_names : list of str
            ``["air_quality"]``.
        classes : list of str
            ``["good", "moderate", "poor"]``.
        frequency : str
            ``"1h"``.
        DESCR : str
            Human readable description.

    See Also
    --------
    - [`make_exogenous_regression`][yohou.datasets._generators.make_exogenous_regression] : Regression variant with continuous target.
    - [`fetch_air_quality_classification`][yohou.datasets._fetchers.fetch_air_quality_classification] : Real air quality classification dataset.

    Examples
    --------
    >>> from yohou.datasets import make_exogenous_classification
    >>> data = make_exogenous_classification(n_samples=200)
    >>> data.y.columns
    ['time', 'air_quality']
    >>> data.classes
    ['good', 'moderate', 'poor']

    """
    rng = np.random.default_rng(random_state)
    times = pl.datetime_range(
        datetime(2024, 1, 1),
        datetime(2024, 1, 1) + timedelta(hours=n_samples - 1),
        interval="1h",
        eager=True,
    ).alias("time")
    t = np.arange(n_samples, dtype=float)

    # Pollutant with 24h cycle centered at 50, amplitude 20
    pollutant = 50.0 + 20.0 * np.sin(2 * np.pi * t / 24) + rng.normal(0, 5.0, n_samples)

    is_holiday = _holiday_indicator(times)

    # Holiday effect: less traffic, lower pollutant readings
    effective_pollutant = pollutant - 5.0 * is_holiday + rng.normal(0, noise, n_samples)

    # Classify
    classes = ["good", "moderate", "poor"]
    labels = np.where(
        effective_pollutant < 40,
        "good",
        np.where(effective_pollutant < 60, "moderate", "poor"),
    )

    y = pl.DataFrame({"time": times, "air_quality": labels})
    X_actual = pl.DataFrame({"time": times, "pollutant": pollutant})
    X_future = pl.DataFrame({"time": times, "is_holiday": is_holiday})

    vintage_idx, target_idx = _forecast_index_grid(n_samples, forecasting_horizon)
    pollutant_noise = rng.normal(0, 2.0, vintage_idx.shape[0])
    X_forecast = pl.DataFrame(
        {
            "vintage_time": times.gather(vintage_idx),
            "time": times.gather(target_idx),
            "pollutant_forecast": pollutant[target_idx] + forecast_bias + pollutant_noise,
        },
        schema={"vintage_time": times.dtype, "time": times.dtype, "pollutant_forecast": pl.Float64},
    )

    frame = y.join(X_actual, on="time").join(X_future, on="time")

    return Bunch(
        y=y,
        X_actual=X_actual,
        X_future=X_future,
        X_forecast=X_forecast,
        frame=frame,
        feature_names=["pollutant", "is_holiday", "pollutant_forecast"],
        target_names=["air_quality"],
        classes=classes,
        frequency="1h",
        DESCR=(
            "Synthetic hourly air quality classification with exogenous features.\n"
            "Target: air_quality in {good, moderate, poor} based on pollutant thresholds.\n"
            "X_actual: realized pollutant readings (sinusoidal 24h cycle + noise).\n"
            "X_future: is_holiday indicator from a fixed civil holiday calendar\n"
            "  (irregular dates, not derivable from the timestamp).\n"
            "X_forecast: pollutant concentration forecasts with systematic bias."
        ),
    )