Skip to content

EvaluationWindow

EvaluationWindow dataclass

A single evaluation window on which the forecast accuracy is measured.

Corresponds to a single train/test split of the time series data at the provided cutoff.

You should never manually create EvaluationWindow objects. Instead, use Task.iter_windows() or Task.get_window() to obtain the evaluation windows corresponding to the task.

Source code in src/fev/task.py 
 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
 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
@dataclasses.dataclass
class EvaluationWindow:
    """
    A single evaluation window on which the forecast accuracy is measured.

    Corresponds to a single train/test split of the time series data at the provided `cutoff`.

    You should never manually create `EvaluationWindow` objects. Instead, use [`Task.iter_windows()`][fev.Task.iter_windows]
    or [`Task.get_window()`][fev.Task.get_window] to obtain the evaluation windows corresponding to the task.
    """

    full_dataset: datasets.Dataset = dataclasses.field(repr=False)
    cutoff: int | str
    horizon: int
    min_context_length: int
    max_context_length: int | None
    # Dataset info
    id_column: str
    timestamp_column: str
    target_columns: list[str]
    known_dynamic_columns: list[str]
    past_dynamic_columns: list[str]
    static_columns: list[str]

    def _get_past_future_test_data(self) -> tuple[datasets.Dataset, datasets.Dataset, datasets.Dataset]:
        dataset = self.full_dataset.select_columns(
            [self.id_column, self.timestamp_column]
            + self.target_columns
            + self.known_dynamic_columns
            + self.past_dynamic_columns
            + self.static_columns
        )

        past_data, future_data = utils.past_future_split(
            dataset,
            timestamp_column=self.timestamp_column,
            cutoff=self.cutoff,
            horizon=self.horizon,
            min_context_length=self.min_context_length,
            max_context_length=self.max_context_length,
        )
        if len(past_data) == 0:
            raise ValueError(
                "All time series in the dataset are too short for the chosen cutoff, horizon and min_context_length"
            )

        future_known = future_data.remove_columns(self.target_columns + self.past_dynamic_columns)
        test_data = future_data.select_columns([self.id_column, self.timestamp_column] + self.target_columns)
        return past_data, future_known, test_data

    def get_input_data(self) -> tuple[datasets.Dataset, datasets.Dataset]:
        """Get data available to the model at prediction time for this evaluation window.

        To convert the input data to a different format, use [`fev.convert_input_data`][fev.convert_input_data].

        Returns
        -------
        past_data : datasets.Dataset
            Historical observations up to the cutoff point.
            Contains: id, timestamps, target values, static covariates, and all dynamic covariates.

            Columns corresponding to `id_column`, `timestamp_column`, `target_columns`, `static_columns`,
            `past_dynamic_columns`, `known_dynamic_columns`.
        future_data : datasets.Dataset
            Known future information for the forecast horizon.

            Columns corresponding to `id_column`, `timestamp_column`, `static_columns`, `known_dynamic_columns`.
        """
        past_data, future_known, _ = self._get_past_future_test_data()
        num_items_before = len(self.full_dataset)
        num_items_after = len(past_data)

        if num_items_after < num_items_before:
            logger.info(
                f"Dropped {num_items_before - num_items_after} out of {num_items_before} time series "
                f"because they had fewer than min_context_length ({self.min_context_length}) "
                f"observations before cutoff ({self.cutoff}) "
                f"or fewer than horizon ({self.horizon}) "
                f"observations after cutoff."
            )

        return past_data, future_known

    def get_ground_truth(self) -> datasets.Dataset:
        """Get ground truth future test data.

        **This data should never be provided to the model!**

        This is a convenience method that exists for debugging and additional evaluation.
        """
        _, _, test_data = self._get_past_future_test_data()
        return test_data

    def compute_metrics(
        self,
        predictions: datasets.DatasetDict,
        metrics: list[Metric],
        seasonality: int,
        quantile_levels: list[float],
    ) -> dict[str, float]:
        """Compute accuracy metrics on the predictions made for this window.

        To compute metrics on your predictions, use [`Task.evaluation_summary`][fev.Task.evaluation_summary] instead.

        This is a convenience method that exists for debugging and additional evaluation.
        """
        past_data, _, test_data = self._get_past_future_test_data()
        past_data.set_format("numpy")
        test_data.set_format("numpy")

        for target_column, predictions_for_column in predictions.items():
            if len(predictions_for_column) != len(test_data):
                raise ValueError(
                    f"Length of predictions for column {target_column} ({len(predictions)}) must "
                    f"match the length of test data ({len(test_data)})"
                )

        test_scores: dict[str, float] = {}
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=RuntimeWarning)
            for metric in metrics:
                scores = []
                for col in self.target_columns:
                    scores.append(
                        metric.compute(
                            test_data=test_data,
                            predictions=predictions[col],
                            past_data=past_data,
                            seasonality=seasonality,
                            quantile_levels=quantile_levels,
                            target_column=col,
                        )
                    )
                test_scores[metric.name] = float(np.mean(scores))
        return test_scores

Attributes

cutoff: int | str instance-attribute

horizon: int instance-attribute

min_context_length: int instance-attribute

max_context_length: int | None instance-attribute

id_column: str instance-attribute

timestamp_column: str instance-attribute

target_columns: list[str] instance-attribute

known_dynamic_columns: list[str] instance-attribute

past_dynamic_columns: list[str] instance-attribute

static_columns: list[str] instance-attribute

Functions

get_input_data() -> tuple[datasets.Dataset, datasets.Dataset]

Get data available to the model at prediction time for this evaluation window.

To convert the input data to a different format, use fev.convert_input_data.

Returns:

Name Type Description
past_data Dataset

Historical observations up to the cutoff point. Contains: id, timestamps, target values, static covariates, and all dynamic covariates.

Columns corresponding to id_column, timestamp_column, target_columns, static_columns, past_dynamic_columns, known_dynamic_columns.
future_data Dataset

Known future information for the forecast horizon.

Columns corresponding to id_column, timestamp_column, static_columns, known_dynamic_columns.
Source code in src/fev/task.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def get_input_data(self) -> tuple[datasets.Dataset, datasets.Dataset]:
    """Get data available to the model at prediction time for this evaluation window.

    To convert the input data to a different format, use [`fev.convert_input_data`][fev.convert_input_data].

    Returns
    -------
    past_data : datasets.Dataset
        Historical observations up to the cutoff point.
        Contains: id, timestamps, target values, static covariates, and all dynamic covariates.

        Columns corresponding to `id_column`, `timestamp_column`, `target_columns`, `static_columns`,
        `past_dynamic_columns`, `known_dynamic_columns`.
    future_data : datasets.Dataset
        Known future information for the forecast horizon.

        Columns corresponding to `id_column`, `timestamp_column`, `static_columns`, `known_dynamic_columns`.
    """
    past_data, future_known, _ = self._get_past_future_test_data()
    num_items_before = len(self.full_dataset)
    num_items_after = len(past_data)

    if num_items_after < num_items_before:
        logger.info(
            f"Dropped {num_items_before - num_items_after} out of {num_items_before} time series "
            f"because they had fewer than min_context_length ({self.min_context_length}) "
            f"observations before cutoff ({self.cutoff}) "
            f"or fewer than horizon ({self.horizon}) "
            f"observations after cutoff."
        )

    return past_data, future_known

get_ground_truth() -> datasets.Dataset

Get ground truth future test data.

This data should never be provided to the model!

This is a convenience method that exists for debugging and additional evaluation.

Source code in src/fev/task.py 
115
116
117
118
119
120
121
122
123
def get_ground_truth(self) -> datasets.Dataset:
    """Get ground truth future test data.

    **This data should never be provided to the model!**

    This is a convenience method that exists for debugging and additional evaluation.
    """
    _, _, test_data = self._get_past_future_test_data()
    return test_data

compute_metrics(predictions: datasets.DatasetDict, metrics: list[Metric], seasonality: int, quantile_levels: list[float]) -> dict[str, float]

Compute accuracy metrics on the predictions made for this window.

To compute metrics on your predictions, use Task.evaluation_summary instead.

This is a convenience method that exists for debugging and additional evaluation.

Source code in src/fev/task.py 
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
def compute_metrics(
    self,
    predictions: datasets.DatasetDict,
    metrics: list[Metric],
    seasonality: int,
    quantile_levels: list[float],
) -> dict[str, float]:
    """Compute accuracy metrics on the predictions made for this window.

    To compute metrics on your predictions, use [`Task.evaluation_summary`][fev.Task.evaluation_summary] instead.

    This is a convenience method that exists for debugging and additional evaluation.
    """
    past_data, _, test_data = self._get_past_future_test_data()
    past_data.set_format("numpy")
    test_data.set_format("numpy")

    for target_column, predictions_for_column in predictions.items():
        if len(predictions_for_column) != len(test_data):
            raise ValueError(
                f"Length of predictions for column {target_column} ({len(predictions)}) must "
                f"match the length of test data ({len(test_data)})"
            )

    test_scores: dict[str, float] = {}
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=RuntimeWarning)
        for metric in metrics:
            scores = []
            for col in self.target_columns:
                scores.append(
                    metric.compute(
                        test_data=test_data,
                        predictions=predictions[col],
                        past_data=past_data,
                        seasonality=seasonality,
                        quantile_levels=quantile_levels,
                        target_column=col,
                    )
                )
            test_scores[metric.name] = float(np.mean(scores))
    return test_scores