diff --git a/DashAI/alembic/versions/a5f2c71e9d40_add_split_to_prediction.py b/DashAI/alembic/versions/a5f2c71e9d40_add_split_to_prediction.py new file mode 100644 index 000000000..8d0f948c0 --- /dev/null +++ b/DashAI/alembic/versions/a5f2c71e9d40_add_split_to_prediction.py @@ -0,0 +1,42 @@ +"""add the split column to prediction + +Revision ID: a5f2c71e9d40 +Revises: b7c1d4e9f206 +Create Date: 2026-09-01 10:00:00.000000 + +Predictions made before this column existed covered the whole dataset, which +is what a null value means, so no backfill is needed. +""" + +from typing import Sequence, Union + +import sqlalchemy as sa + +from alembic import op + +revision: str = "a5f2c71e9d40" +down_revision: Union[str, None] = "b7c1d4e9f206" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + if "prediction" not in inspector.get_table_names(): + return + columns = {column["name"] for column in inspector.get_columns("prediction")} + if "split" in columns: + return + op.add_column("prediction", sa.Column("split", sa.String(), nullable=True)) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + if "prediction" not in inspector.get_table_names(): + return + columns = {column["name"] for column in inspector.get_columns("prediction")} + if "split" not in columns: + return + op.drop_column("prediction", "split") diff --git a/DashAI/back/api/api_v1/endpoints/explainers.py b/DashAI/back/api/api_v1/endpoints/explainers.py index d71d7ebde..06f25620f 100644 --- a/DashAI/back/api/api_v1/endpoints/explainers.py +++ b/DashAI/back/api/api_v1/endpoints/explainers.py @@ -28,7 +28,7 @@ ModelSession, Run, ) -from DashAI.back.splitters.splits_payload import splitter_class_for +from DashAI.back.splitters.splits_payload import run_splits if TYPE_CHECKING: from sqlalchemy.orm import sessionmaker @@ -1107,8 +1107,6 @@ async def get_explainable_splits( HTTPException If the run does not exist in the database. """ - import json - with session_factory() as db: try: run: Run = db.get(Run, run_id) @@ -1123,10 +1121,8 @@ async def get_explainable_splits( status_code=status.HTTP_404_NOT_FOUND, detail="Model session not found", ) - split_indexes = json.loads(run.split_indexes) if run.split_indexes else {} + split_indexes = run.split_indexes session_splits = model_session.splits - if isinstance(session_splits, str): - session_splits = json.loads(session_splits) except exc.SQLAlchemyError as e: log.exception(e) raise HTTPException( @@ -1135,14 +1131,12 @@ async def get_explainable_splits( ) from e try: - splitter_class = splitter_class_for(session_splits, component_registry) + return {"splits": run_splits(session_splits, split_indexes, component_registry)} except ValueError as e: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e) ) from e - return {"splits": splitter_class.explainable_splits(split_indexes)} - @router.post("/local/valid-datasets") @inject @@ -1167,9 +1161,6 @@ async def valid_datasets( dict ``{"valid_dataset_ids": [...]}`` with the ids of the valid datasets. """ - # get_columns_spec reads only the Arrow schema metadata (column names + - # types), never the rows, so validating every dataset stays cheap even with - # many/large (e.g. image) datasets on the platform. from DashAI.back.dataloaders.classes.dashai_dataset import get_columns_spec with session_factory() as db: diff --git a/DashAI/back/api/api_v1/endpoints/predict.py b/DashAI/back/api/api_v1/endpoints/predict.py index 23da1a336..da3778af3 100644 --- a/DashAI/back/api/api_v1/endpoints/predict.py +++ b/DashAI/back/api/api_v1/endpoints/predict.py @@ -15,6 +15,7 @@ Run, ) from DashAI.back.job.predict_job import run_manual_prediction +from DashAI.back.splitters.splits_payload import predictable_splits if TYPE_CHECKING: from sqlalchemy.orm import Session, sessionmaker @@ -44,6 +45,10 @@ async def create_prediction( The ID of the trained model/run. dataset_id : int | None The ID of the dataset to use for prediction (optional). + split : str | None + The partition of that dataset to predict on (optional). Only applies + when the dataset is the one the model was trained on; ``None`` and + ``"all"`` both mean every row. session_factory : Callable[..., ContextManager[Session]] A factory that creates a context manager that handles a SQLAlchemy session. The generated session can be used to access and query the database. @@ -69,6 +74,7 @@ async def create_prediction( prediction = Prediction( run_id=run.id, dataset_id=params.dataset_id, + split=params.split, ) db.add(prediction) db.commit() @@ -127,6 +133,78 @@ async def get_all_predictions( return predictions +@router.get("/splits/{run_id}") +@inject +async def get_prediction_splits( + run_id: int, + session_factory: "sessionmaker" = Depends(lambda: di["session_factory"]), + component_registry: "ComponentRegistry" = Depends(lambda: di["component_registry"]), +): + """Return the dataset partitions a prediction of this run may target. + + The partitions describe the dataset the model was trained on, so they only + apply to a prediction run on that same dataset; any other dataset is + predicted whole. Which partitions exist depends on how the run was + evaluated, so the splitter that produced it decides the list and its names, + and a task whose models only predict forward keeps just the partitions + outside the window its saved model was fitted through. + + Parameters + ---------- + run_id : int + The ID of the trained model/run. + session_factory : Callable[..., ContextManager[Session]] + A factory that creates a context manager that handles a SQLAlchemy session. + component_registry : ComponentRegistry + Registry used to resolve the splitter that produced the run. + + Returns + ------- + dict + A ``splits`` list of ``{"name", "rows"}`` entries, empty when the run + has no partition worth offering. + + Raises + ------ + HTTPException + If the run or its model session does not exist, or if the run was + produced by a splitter that is no longer registered. + """ + db: Session + with session_factory() as db: + run: Run = db.get(Run, run_id) + if not run: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Run not found" + ) + model_session: ModelSession = db.get(ModelSession, run.model_session_id) + if not model_session: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Model session not found", + ) + session_splits = model_session.splits + split_indexes = run.split_indexes + training_dataset_id = model_session.dataset_id + task_name = model_session.task_name + evaluation_strategy = model_session.evaluation_strategy + + try: + splits = predictable_splits( + session_splits, + split_indexes, + component_registry, + task_name=task_name, + evaluation_strategy=evaluation_strategy, + ) + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e) + ) from e + + return {"splits": splits, "training_dataset_id": training_dataset_id} + + @router.get("/filter_datasets") async def filter_datasets_endpoint( run_id: int = Query(..., description="The ID of the trained model/run"), diff --git a/DashAI/back/api/api_v1/schemas/prediction_params.py b/DashAI/back/api/api_v1/schemas/prediction_params.py index 0b963bee7..5f1dae7c0 100644 --- a/DashAI/back/api/api_v1/schemas/prediction_params.py +++ b/DashAI/back/api/api_v1/schemas/prediction_params.py @@ -6,3 +6,4 @@ class PredictionCreationParams(BaseModel): run_id: int dataset_id: Optional[int] = None + split: Optional[str] = None diff --git a/DashAI/back/dependencies/database/models.py b/DashAI/back/dependencies/database/models.py index c6ab9ff95..fa7b2fe6e 100644 --- a/DashAI/back/dependencies/database/models.py +++ b/DashAI/back/dependencies/database/models.py @@ -233,6 +233,7 @@ class Prediction(Base): id: Mapped[int] = mapped_column(primary_key=True) run_id: Mapped[int] = mapped_column(ForeignKey("run.id", ondelete="CASCADE")) dataset_id: Mapped[int] = mapped_column(ForeignKey("dataset.id"), nullable=True) + split: Mapped[str] = mapped_column(String, nullable=True) huey_id: Mapped[str] = mapped_column(String, nullable=True) created: Mapped[DateTime] = mapped_column(DateTime, default=datetime.now) last_modified: Mapped[DateTime] = mapped_column( diff --git a/DashAI/back/evaluation/base_evaluation_strategy.py b/DashAI/back/evaluation/base_evaluation_strategy.py index 5f911cdea..0e23ff3ef 100644 --- a/DashAI/back/evaluation/base_evaluation_strategy.py +++ b/DashAI/back/evaluation/base_evaluation_strategy.py @@ -23,16 +23,11 @@ class BaseEvaluationStrategy(metaclass=ABCMeta): TYPE: Final[str] = "EvaluationStrategy" - # How this strategy divides the dataset. The frontend renders holdout - # controls or fold controls from this rather than comparing class names, - # which is what previously made a new strategy unreachable from the UI. KIND: str = "holdout" - - # Which partitions this strategy records metrics for. Scoring the training - # partition means predicting on rows the model was fitted on, which is a - # fit statistic; a forecaster has no such thing to report. SCORED_SPLITS: tuple = (SplitEnum.TRAIN, SplitEnum.VALIDATION, SplitEnum.TEST) + FINAL_FIT_PARTITIONS: tuple = ("train",) + @classmethod def get_metadata(cls) -> dict: """Describe the strategy for the frontend. diff --git a/DashAI/back/evaluation/forecasting_holdout.py b/DashAI/back/evaluation/forecasting_holdout.py index fcb2fbe31..dba6abf06 100644 --- a/DashAI/back/evaluation/forecasting_holdout.py +++ b/DashAI/back/evaluation/forecasting_holdout.py @@ -5,11 +5,10 @@ class ForecastingHoldoutEvaluationStrategy(SinglePartitionEvaluationStrategy): - """Holdout evaluation that treats validation as history rather than a sample. + """Holdout evaluation that records no in-sample metrics. - Two things the ordinary holdout strategy assumes are wrong for a - forecaster, and both of them are decisions about evaluation rather than - about any model. + One thing the ordinary holdout strategy assumes is wrong for a forecaster, + and it is a decision about evaluation rather than about any model. **The training partition is not scored.** Scoring it would mean asking the model about dates it was fitted on. That is an in-sample fit statistic, @@ -17,95 +16,33 @@ class ForecastingHoldoutEvaluationStrategy(SinglePartitionEvaluationStrategy): several steps out; showing the two side by side in one results table invites exactly that comparison. Only validation and test are recorded. - **The kept model is fitted through validation.** For most tasks the - validation partition is a held out sample that has to stay out of the fit. - For a forecaster it is simply the most recent stretch of the series, and - the stretch nearest to whatever comes next. Leaving it out makes the model - reach across the whole validation window before arriving at the first test - row, so the test metrics describe a longer horizon than the one being - asked about. + **The kept model is fitted on the training partition alone**, like every + other holdout run, and nothing is fed to it afterwards. Two approaches that + would have changed that were tried and dropped, both because they hand the + model data from a partition it was meant to be held out from: - The validation metrics are still measured on a model fitted on training - data alone, which is what makes them honest: they are recorded before the - refit. So the two columns in the results table answer different questions, - and both answer them fairly. + refitting through validation before scoring test, which overwrote the + fit the validation metrics came from, so the saved model could not + reproduce its own results table; - validation metrics <- model fitted on train - test metrics <- model fitted on train + validation + advancing the model through the observed validation rows at predict + time, which re-estimates nothing but still lets a held out partition + reach the model, which no other task in DashAI does. - Hyperparameter search is untouched. Its trials are scored on validation, - so they must not be fitted on it. - """ - - COMPATIBLE_COMPONENTS = ["ForecastingTask"] - SCORED_SPLITS: tuple = (SplitEnum.VALIDATION, SplitEnum.TEST) - - def execute(self, x, y, run, db): - """Score validation on a trial fit, then refit and score test. - - Parameters - ---------- - x : DatasetDict - Input partitions, keyed by split name. - y : DatasetDict - Target partitions, keyed by split name. - run : Run - Database model representing the current run. - db : Session - SQLAlchemy session used to persist metrics. - - Returns - ------- - tuple - The trained model and the paths of any HPO plots. - """ - plot_paths = [] - model = self.model - - model.x_data = x - model.y_data = y - - if self.optimizer and self.run_optimizable_parameters: - self._report_progress(0.2, "Hyperparameter optimization") - model = self._do_hpo(model, x, y, run, db) - plot_paths = self._generate_hpo_plots(run) + So the two columns describe different horizons, and deliberately: - # Fitted on training data only, so the validation score below measures - # a model that has not seen the rows it is being scored on. - self._report_progress(0.5, "Training") - model.train(x["train"], y["train"]) + validation metrics <- forecasting 1..len(val) past the fit + test metrics <- forecasting len(val)+1..len(val)+len(test), + its own forecasts standing in for validation - self._report_progress(0.8, "Computing validation metrics") - self._calculate_metrics_if_missing(model, run, db, SplitEnum.VALIDATION) + The test column is therefore the harder question, not the same one further + along. Comparing like with like over a chosen horizon is what + ``RollingOriginSplitter`` is for, since its ``horizon`` says outright how + many steps ahead each refit is scored on. - # Now the model that gets kept: the same configuration, refitted with - # the validation rows included, since for a series they are history. - self._report_progress(0.9, "Refitting on train and validation") - self._fit_final_model(model, x, y) - - self._report_progress(0.95, "Computing test metrics") - self._calculate_metrics_if_missing(model, run, db, SplitEnum.TEST) - - return model, plot_paths - - def _fit_final_model(self, model, x, y): - """Fit the kept model on the training and validation rows together. - - Parameters - ---------- - model : BaseModel - The model to fit. - x : DatasetDict - Input partitions. - y : DatasetDict - Target partitions. - """ - validation_x = x.get("validation") - validation_y = y.get("validation") - - if validation_x is None or validation_y is None or len(validation_x) == 0: - model.train(x["train"], y["train"]) - return + Hyperparameter search is untouched. Its trials are scored on validation, so + they must not be fitted on it. + """ - extend = type(model)._extend - model.train(extend(x["train"], validation_x), extend(y["train"], validation_y)) + COMPATIBLE_COMPONENTS = ["ForecastingTask"] + SCORED_SPLITS: tuple = (SplitEnum.VALIDATION, SplitEnum.TEST) diff --git a/DashAI/back/evaluation/holdout.py b/DashAI/back/evaluation/holdout.py index 3e5783899..c23fff871 100644 --- a/DashAI/back/evaluation/holdout.py +++ b/DashAI/back/evaluation/holdout.py @@ -156,10 +156,7 @@ def evaluate(self, model, input_dataset, output_dataset, metric): output_dataset["validation"], is_fit=False ) - # Calculate metric for train and validation data each trial. The - # training partition is skipped for a strategy that does not score it, - # which for a forecaster is not a preference: predicting on dates it - # was fitted on is refused, so asking would fail every trial. + # Calculate metric for train and validation data each trial. if SplitEnum.TRAIN in self.SCORED_SPLITS: model.calculate_metrics(split=SplitEnum.TRAIN, level=LevelEnum.TRIAL) model.calculate_metrics(split=SplitEnum.VALIDATION, level=LevelEnum.TRIAL) diff --git a/DashAI/back/job/predict_job.py b/DashAI/back/job/predict_job.py index 6ed3d1d0d..f509c101d 100644 --- a/DashAI/back/job/predict_job.py +++ b/DashAI/back/job/predict_job.py @@ -11,6 +11,7 @@ from DashAI.back.dependencies.database.models import Dataset, ModelSession, Prediction from DashAI.back.job.base_job import BaseJob, JobError from DashAI.back.models.base_model import BaseModel +from DashAI.back.splitters.splits_payload import run_split_indexes from DashAI.back.tasks.base_task import BaseTask from DashAI.back.tasks.regression_task import RegressionTask @@ -30,7 +31,26 @@ def _run_prediction_pipeline( loaded_dataset: "DashAIDataset", model_session: ModelSession, ) -> Tuple["DashAIDataset", Any]: - """Run shared prediction steps from prepared input data to final predictions.""" + """Run shared prediction steps from prepared input data to final predictions. + + Parameters + ---------- + task : BaseTask + The task the run belongs to. + trained_model : BaseModel + The model loaded from the run. + train_dataset : DashAIDataset + The dataset the model was trained on, for type and label information. + loaded_dataset : DashAIDataset + The rows to predict. + model_session : ModelSession + The session declaring the input and output columns. + + Returns + ------- + tuple + The prepared inputs and the predictions for them. + """ import numpy as np prepared_dataset = loaded_dataset.select_columns(model_session.input_columns) @@ -411,12 +431,31 @@ def run( f"{dataset_trained.file_path}/dataset/" ) from e + row_indexes = None + if prediction.split and dataset_id == model_session.dataset_id: + try: + row_indexes = run_split_indexes( + model_session.splits, + prediction.run.split_indexes, + component_registry, + prediction.split, + ) + except ValueError as e: + prediction.set_status_as_error() + db.commit() + log.exception(e) + raise JobError( + f"Cannot predict on the {prediction.split} split: {e}" + ) from e + try: # Load or create prediction dataset if dataset_id: loaded_dataset: "DashAIDataset" = load_dataset( str(Path(f"{dataset.file_path}/dataset/")) ) + if row_indexes is not None: + loaded_dataset = loaded_dataset.select(row_indexes) else: dataset_trained_path = str( Path(f"{dataset_trained.file_path}/dataset/") @@ -438,16 +477,12 @@ def run( prediction.set_status_as_error() db.commit() log.error(f"Validation Error: {ve}") - raise HTTPException( - status_code=400, - detail=f"Invalid input data: {str(ve)}", - ) from ve + raise JobError(f"Invalid input data: {ve}") from ve except TypeError as te: + prediction.set_status_as_error() + db.commit() log.error(f"Type Error: {te}") - raise HTTPException( - status_code=400, - detail=f"Type validation failed: {str(te)}", - ) from te + raise JobError(f"Type validation failed: {te}") from te except Exception as e: prediction.set_status_as_error() db.commit() diff --git a/DashAI/back/models/forecasting/arima.py b/DashAI/back/models/forecasting/arima.py index fbd22eaf7..1ae4773ee 100644 --- a/DashAI/back/models/forecasting/arima.py +++ b/DashAI/back/models/forecasting/arima.py @@ -250,9 +250,6 @@ def train( ) with warnings.catch_warnings(): - # statsmodels warns that it is assuming evenly spaced observations - # because no date index was supplied. That is the assumption this - # model makes on purpose, so the warning says nothing new. warnings.simplefilter("ignore") self._result = _ARIMA(series, order=(self.p, self.d, self.q)).fit() @@ -261,18 +258,17 @@ def train( self._fitted = True return self - def predict(self, x: "DashAIDataset") -> "np.ndarray": - """Forecast forward from the end of the training series. + def _forecast(self, steps: int) -> "np.ndarray": + """Forecast the next ``steps`` periods after the end of the history. Parameters ---------- - x : DashAIDataset - The rows to forecast, whose dates say how far ahead each one is. + steps : int + How many periods to forecast. Returns ------- np.ndarray - One forecast value per requested row. + One value per period, in order. """ - self._require_fitted() - return self._forecast_at(x, lambda steps: self._result.forecast(steps=steps)) + return self._result.forecast(steps=steps) diff --git a/DashAI/back/models/forecasting/base_forecasting_model.py b/DashAI/back/models/forecasting/base_forecasting_model.py index d4e9019e0..dc2739ee3 100644 --- a/DashAI/back/models/forecasting/base_forecasting_model.py +++ b/DashAI/back/models/forecasting/base_forecasting_model.py @@ -175,6 +175,54 @@ def _steps_from_grid(self, dates: "pd.Series") -> "np.ndarray | None": return positions + def _dates_of(self, x: "DashAIDataset") -> "pd.Series | None": + """Read and parse the date column of a set of rows. + + Parameters + ---------- + x : DashAIDataset + Rows holding the date column. + + Returns + ------- + pd.Series or None + The parsed dates, or None when the rows carry no date column. + """ + from DashAI.back.types.date_utils import parse_date_column + from DashAI.back.types.value_types import Date + + date_columns = [ + name for name in x.column_names if isinstance(x.types.get(name), Date) + ] + if not date_columns: + return None + return parse_date_column(x.to_pandas()[date_columns[0]], self._date_format) + + def _steps_of(self, dates: "pd.Series") -> "np.ndarray": + """Count how many periods past the end of training each date falls. + + Unlike :meth:`_steps_ahead` this reports what it finds, including the + zero and negative counts of dates inside the fitted window, because + the caller may be looking for exactly those. + + Parameters + ---------- + dates : pd.Series + Dates already parsed. + + Returns + ------- + np.ndarray + One step number per date, in the order given. + """ + import numpy as np + + steps = self._steps_from_grid(dates) + if steps is None: + offsets = (dates - self._last_train_date) / self._step_delta + steps = np.rint(offsets.to_numpy(dtype=float)).astype(int) + return steps + def _steps_ahead(self, x: "DashAIDataset") -> "np.ndarray": """Work out how many periods past training each requested date falls. @@ -197,29 +245,15 @@ def _steps_ahead(self, x: "DashAIDataset") -> "np.ndarray": """ import numpy as np - from DashAI.back.types.date_utils import parse_date_column - from DashAI.back.types.value_types import Date - - date_columns = [ - name for name in x.column_names if isinstance(x.types.get(name), Date) - ] - if ( - not date_columns - or self._last_train_date is None - or self._step_delta is None - ): + dates = None + if self._last_train_date is not None and self._step_delta is not None: + dates = self._dates_of(x) + if dates is None: # No dates to align against, so the rows can only mean "the next # len(x) periods", which is what they meant before. return np.arange(1, len(x) + 1) - dates = parse_date_column(x.to_pandas()[date_columns[0]], self._date_format) - - steps = self._steps_from_grid(dates) - if steps is None: - # No regular grid to count on, so the best available reading is - # how many typical gaps each date lies past the end of training. - offsets = (dates - self._last_train_date) / self._step_delta - steps = np.rint(offsets.to_numpy(dtype=float)).astype(int) + steps = self._steps_of(dates) if (steps < 1).any(): raise ValueError( @@ -231,6 +265,51 @@ def _steps_ahead(self, x: "DashAIDataset") -> "np.ndarray": return steps + def predict(self, x: "DashAIDataset") -> "np.ndarray": + """Forecast the requested dates from the end of the training data. + + A partition that does not directly follow the training data is + reached by forecasting across the gap, so the model's own forecasts + stand in for the periods in between and the error compounds over a + horizon nobody asked about. That is what a model fitted on the + training partition alone can say, and every other task predicts + from that same fit. + + Parameters + ---------- + x : DashAIDataset + The rows to forecast, whose dates say how far ahead each one is. + + Returns + ------- + np.ndarray + One forecast value per requested row. + """ + self._require_fitted() + return self._forecast_at(x, self._forecast) + + def _forecast(self, steps: int) -> "np.ndarray": + """Forecast the next ``steps`` periods after the end of the history. + + Parameters + ---------- + steps : int + How many periods to forecast. + + Returns + ------- + np.ndarray + One value per period, in order. + + Raises + ------ + NotImplementedError + If the subclass does not provide an implementation. + """ + raise NotImplementedError( + "A forecasting model must say how it forecasts a number of steps." + ) + def _forecast_at(self, x: "DashAIDataset", forecast) -> "np.ndarray": """Pick the forecast values for the requested dates. @@ -252,31 +331,6 @@ def _forecast_at(self, x: "DashAIDataset", forecast) -> "np.ndarray": values = np.asarray(forecast(int(steps.max())), dtype=float) return values[steps - 1] - @staticmethod - def _extend(earlier: "DashAIDataset", later: "DashAIDataset") -> "DashAIDataset": - """Join two consecutive partitions into one continuous history. - - Parameters - ---------- - earlier : DashAIDataset - The partition that comes first in time. - later : DashAIDataset - The partition that follows it. - - Returns - ------- - DashAIDataset - The two concatenated, keeping the column types of the first. - """ - import pandas as pd - - from DashAI.back.dataloaders.classes.dashai_dataset import to_dashai_dataset - - return to_dashai_dataset( - pd.concat([earlier.to_pandas(), later.to_pandas()], ignore_index=True), - types=dict(earlier.types), - ) - def save(self, filename: str) -> None: """Serialise the model to disk using joblib. diff --git a/DashAI/back/models/forecasting/exponential_smoothing.py b/DashAI/back/models/forecasting/exponential_smoothing.py index 8bd54b0c5..b82db1d30 100644 --- a/DashAI/back/models/forecasting/exponential_smoothing.py +++ b/DashAI/back/models/forecasting/exponential_smoothing.py @@ -299,8 +299,6 @@ def train( ) with warnings.catch_warnings(): - # As with ARIMA, statsmodels notes the absence of a date index and - # assumes even spacing, which is this model's assumption anyway. warnings.simplefilter("ignore") self._result = _ExponentialSmoothing( series, @@ -314,18 +312,17 @@ def train( self._fitted = True return self - def predict(self, x: "DashAIDataset") -> "np.ndarray": - """Forecast forward from the end of the training series. + def _forecast(self, steps: int) -> "np.ndarray": + """Forecast the next ``steps`` periods after the end of the history. Parameters ---------- - x : DashAIDataset - The rows to forecast, whose dates say how far ahead each one is. + steps : int + How many periods to forecast. Returns ------- np.ndarray - One forecast value per requested row. + One value per period, in order. """ - self._require_fitted() - return self._forecast_at(x, lambda steps: self._result.forecast(steps)) + return self._result.forecast(steps) diff --git a/DashAI/back/models/forecasting/naive.py b/DashAI/back/models/forecasting/naive.py index 64ff309b6..36b3be93f 100644 --- a/DashAI/back/models/forecasting/naive.py +++ b/DashAI/back/models/forecasting/naive.py @@ -107,13 +107,13 @@ def train( self._fitted = True return self - def predict(self, x: "DashAIDataset") -> "np.ndarray": + def _forecast(self, steps: int) -> "np.ndarray": """Repeat the last observed value for every requested step. Parameters ---------- - x : DashAIDataset - The rows to forecast, whose dates say how far ahead each one is. + steps : int + How many periods to forecast. Returns ------- @@ -122,7 +122,4 @@ def predict(self, x: "DashAIDataset") -> "np.ndarray": """ import numpy as np - self._require_fitted() - return self._forecast_at( - x, lambda steps: np.full(steps, self._last_value, dtype=float) - ) + return np.full(steps, self._last_value, dtype=float) diff --git a/DashAI/back/models/forecasting/seasonal_naive.py b/DashAI/back/models/forecasting/seasonal_naive.py index e63f29229..f5c5d8ce3 100644 --- a/DashAI/back/models/forecasting/seasonal_naive.py +++ b/DashAI/back/models/forecasting/seasonal_naive.py @@ -190,13 +190,13 @@ def train( self._fitted = True return self - def predict(self, x: "DashAIDataset") -> "np.ndarray": - """Repeat the last full season out to the dates requested. + def _forecast(self, steps: int) -> "np.ndarray": + """Repeat the last full season out to the requested length. Parameters ---------- - x : DashAIDataset - The rows to forecast, whose dates say how far ahead each one is. + steps : int + How many periods to forecast. Returns ------- @@ -205,11 +205,7 @@ def predict(self, x: "DashAIDataset") -> "np.ndarray": """ import numpy as np - self._require_fitted() - return self._forecast_at( - x, - lambda steps: np.array( - [self._last_season[i % self.season_length] for i in range(steps)], - dtype=float, - ), + return np.array( + [self._last_season[i % self.season_length] for i in range(steps)], + dtype=float, ) diff --git a/DashAI/back/optimizers/optuna_optimizer.py b/DashAI/back/optimizers/optuna_optimizer.py index e285e3cda..e56a0ba3b 100644 --- a/DashAI/back/optimizers/optuna_optimizer.py +++ b/DashAI/back/optimizers/optuna_optimizer.py @@ -1,3 +1,4 @@ +import logging from typing import TYPE_CHECKING from DashAI.back.core.schema_fields import ( @@ -12,6 +13,10 @@ if TYPE_CHECKING: import optuna +logger = logging.getLogger(__name__) + +UNFITTABLE_TRIAL_ERRORS = (ValueError, ArithmeticError) + class OptunaSchema(BaseSchema): n_trials: schema_field( @@ -265,6 +270,8 @@ def optimize( self.metric = metric["class"] + failures = [] + def objective(trial): # Set value for each hyperparameter and for each model # (either self or submodels nested inside) @@ -294,6 +301,9 @@ def objective(trial): score = strategy( self.model, self.input_dataset, self.output_dataset, self.metric ) + except UNFITTABLE_TRIAL_ERRORS as e: + failures.append(e) + raise finally: # Cleared even when the trial is pruned: the model instance is # reused across trials and by the final refit afterwards. @@ -301,7 +311,31 @@ def objective(trial): return score - study.optimize(objective, n_trials=self.n_trials) + study.optimize(objective, n_trials=self.n_trials, catch=UNFITTABLE_TRIAL_ERRORS) + + completed = study.get_trials( + deepcopy=False, states=(optuna.trial.TrialState.COMPLETE,) + ) + if not completed: + raise ValueError( + f"Every one of the {len(study.trials)} trials failed: the model " + f"could not be fitted with any combination the search drew from " + f"the ranges given. Narrow them and try again. The last trial " + f"failed with: {failures[-1]}" + if failures + else ( + f"Every one of the {len(study.trials)} trials failed and none " + f"produced a score." + ) + ) + if failures: + logger.warning( + "%d of %d hyperparameter trials could not be fitted and were " + "skipped. The last one failed with: %s", + len(failures), + len(study.trials), + failures[-1], + ) # Write the best values back onto the objects that actually declare them. # `self.parameters` holds (owner, key, bounds, dtype) tuples built by diff --git a/DashAI/back/splitters/splits_payload.py b/DashAI/back/splitters/splits_payload.py index ef7d68d06..39322f9c1 100644 --- a/DashAI/back/splitters/splits_payload.py +++ b/DashAI/back/splitters/splits_payload.py @@ -7,7 +7,8 @@ different set of keys, so every reader normalizes the payload first. """ -from typing import Any, Dict, List, Tuple, Type +import json +from typing import Any, Dict, List, Optional, Tuple, Type # Before fold splitters existed the payload named no splitter at all: every # split was a holdout split. @@ -177,3 +178,215 @@ def explainable_indexes( evaluation = unseen.get("test") or next(iter(unseen.values())) return partitions.get(training, []), evaluation, partitions.get("val", []) + + +def _parse_payload(payload: Any) -> Dict[str, Any]: + """Return a splits payload as a dictionary, decoding it when stored as text. + + Parameters + ---------- + payload : Any + A ``ModelSession.splits`` or ``Run.split_indexes`` value, which the + database may hold either as a dict or as its JSON encoding. + + Returns + ------- + dict + The decoded payload, or an empty dictionary when there is none. + """ + if isinstance(payload, str): + return json.loads(payload) if payload else {} + return payload or {} + + +def run_splits( + session_splits: Any, split_indexes: Any, component_registry +) -> List[Dict[str, Any]]: + """Describe the dataset partitions of a run that a later job may target. + + Which partitions exist depends on how the run was evaluated, so the + splitter that produced it decides the list and its names. Both explaining + and predicting on the training dataset offer the same set: the partitions + the saved model was measured on, plus the whole dataset. + + Parameters + ---------- + session_splits : Any + The ``ModelSession.splits`` payload, as a dict or its JSON encoding. + split_indexes : Any + The ``Run.split_indexes`` payload, as a dict or its JSON encoding. + component_registry : ComponentRegistry + Registry used to resolve the splitter by name. + + Returns + ------- + list[dict] + One ``{"name", "rows"}`` entry per partition, empty when the run has no + partition worth offering. + + Raises + ------ + ValueError + If the payload names a splitter that is not registered. + """ + splitter_class = splitter_class_for( + _parse_payload(session_splits), component_registry + ) + return splitter_class.explainable_splits(_parse_payload(split_indexes)) + + +def run_split_indexes( + session_splits: Any, split_indexes: Any, component_registry, split: str +) -> Optional[List[int]]: + """Resolve the rows one named partition of a run holds. + + Parameters + ---------- + session_splits : Any + The ``ModelSession.splits`` payload, as a dict or its JSON encoding. + split_indexes : Any + The ``Run.split_indexes`` payload, as a dict or its JSON encoding. + component_registry : ComponentRegistry + Registry used to resolve the splitter by name. + split : str + Name of the partition, as reported by :func:`run_splits`. ``"all"`` + stands for the whole dataset. + + Returns + ------- + list[int] or None + The row indexes of the partition, or None when ``split`` covers the + whole dataset and no selection is needed. + + Raises + ------ + ValueError + If the splitter is not registered, or if it declares no partition by + that name. + """ + if not split or split == "all": + return None + + splitter_class = splitter_class_for( + _parse_payload(session_splits), component_registry + ) + try: + partitions = splitter_class.explainable_partitions( + _parse_payload(split_indexes) + ) + except (KeyError, NotImplementedError, TypeError) as e: + raise ValueError( + "The run's split indexes do not match the splitter that produced it." + ) from e + + if split not in partitions: + raise ValueError(f"{split} is not a partition of this run.") + return list(partitions[split]) + + +def predictable_splits( + session_splits: Any, + split_indexes: Any, + component_registry, + *, + task_name: str, + evaluation_strategy: str, +) -> List[Dict[str, Any]]: + """Describe the partitions of a run its saved model can be asked to predict. + + For most tasks that is every partition the run has, plus the whole dataset: + a fitted classifier will label the rows it was trained on. A model of a + task that predicts forward only cannot. It reads a date and answers how far + past the end of training it lies, so every partition inside the window the + kept model was fitted through is dropped, and so is the whole dataset entry + that contains them. + + Which rows those are comes from the evaluation strategy, since strategies + differ on what the kept model is allowed to learn from: a holdout run fits + on its training partition, while the folds of a rolling origin run walk + through everything outside the reserved tail. The comparison is made on row + indexes, which run in time order for every splitter a forward-only task can + use. + + Parameters + ---------- + session_splits : Any + The ``ModelSession.splits`` payload, as a dict or its JSON encoding. + split_indexes : Any + The ``Run.split_indexes`` payload, as a dict or its JSON encoding. + component_registry : ComponentRegistry + Registry used to resolve the splitter, task and strategy by name. + task_name : str + The ``ModelSession.task_name`` of the run. + evaluation_strategy : str + The ``ModelSession.evaluation_strategy`` of the run. + + Returns + ------- + list[dict] + One ``{"name", "rows"}`` entry per partition that can be predicted, + empty when there is none. + + Raises + ------ + ValueError + If the payload names a splitter that is not registered. + """ + splits = run_splits(session_splits, split_indexes, component_registry) + + task_class = _registered_class(component_registry, task_name) + if not getattr(task_class, "PREDICTS_FORWARD_ONLY", False): + return splits + + strategy_class = _registered_class(component_registry, evaluation_strategy) + fitted_partitions = getattr(strategy_class, "FINAL_FIT_PARTITIONS", ("train",)) + + splitter_class = splitter_class_for( + _parse_payload(session_splits), component_registry + ) + try: + partitions = splitter_class.explainable_partitions( + _parse_payload(split_indexes) + ) + except (KeyError, NotImplementedError, TypeError): + return [] + + fitted = [ + index for name in fitted_partitions for index in partitions.get(name) or [] + ] + if not fitted: + return [] + + last_fitted = max(fitted) + return [ + {"name": name, "rows": len(indexes)} + for name, indexes in partitions.items() + if indexes and min(indexes) > last_fitted + ] + + +def _registered_class(component_registry, name: str) -> Optional[Type]: + """Return a registered component class, or None when it is not installed. + + A run whose task or strategy came from a plugin that has since been removed + still has to describe itself as well as it can, so a missing name is not an + error here. + + Parameters + ---------- + component_registry : ComponentRegistry + Registry used to resolve the component by name. + name : str + Name of the component to resolve. + + Returns + ------- + type or None + The registered class, or None when the registry does not have it. + """ + try: + if name not in component_registry: + return None + return component_registry[name]["class"] + except (KeyError, TypeError): + return None diff --git a/DashAI/back/tasks/base_task.py b/DashAI/back/tasks/base_task.py index 45f9b511b..357ba00b3 100644 --- a/DashAI/back/tasks/base_task.py +++ b/DashAI/back/tasks/base_task.py @@ -16,6 +16,8 @@ class BaseTask: TYPE: Final[str] = "Task" + PREDICTS_FORWARD_ONLY: bool = False + @property @abstractmethod def schema(self) -> Dict[str, Any]: diff --git a/DashAI/back/tasks/forecasting_task.py b/DashAI/back/tasks/forecasting_task.py index 74e847d66..245d49cf5 100644 --- a/DashAI/back/tasks/forecasting_task.py +++ b/DashAI/back/tasks/forecasting_task.py @@ -29,6 +29,8 @@ class ForecastingTask(BaseTask): directly and cannot be expressed that way. """ + PREDICTS_FORWARD_ONLY: bool = True + DESCRIPTION: str = MultilingualString( en=( "Predict the future values of a time series from its own history. " diff --git a/DashAI/front/src/api/predict.ts b/DashAI/front/src/api/predict.ts index c38555a3e..002a2229f 100644 --- a/DashAI/front/src/api/predict.ts +++ b/DashAI/front/src/api/predict.ts @@ -27,14 +27,42 @@ export const downloadPredict = async (prediction_id: string) => { export const createPrediction = async ( run_id: number, dataset_id?: number, + split?: string | null, ): Promise => { const response = await api.post(`${predictEndpoint}/`, { run_id, dataset_id, + split, }); return response.data; }; +export interface IPredictionSplit { + name: string; + rows: number; +} + +/** + * The partitions the run carved its training dataset into, which a prediction + * may target when it runs on that same dataset. Which ones exist depends on how + * the run was evaluated, so the backend decides the list and its names. + */ +export const getPredictionSplits = async ( + runId: number, +): Promise<{ + splits: IPredictionSplit[]; + trainingDatasetId: number | null; +}> => { + const response = await api.get<{ + splits: IPredictionSplit[]; + training_dataset_id: number | null; + }>(`${predictEndpoint}/splits/${runId}`); + return { + splits: response.data.splits, + trainingDatasetId: response.data.training_dataset_id, + }; +}; + export const getPredictions = async ( run_id?: number, prediction_id?: string, diff --git a/DashAI/front/src/components/models/DatasetPredictionPanel.jsx b/DashAI/front/src/components/models/DatasetPredictionPanel.jsx index e3346cb98..bc6713adc 100644 --- a/DashAI/front/src/components/models/DatasetPredictionPanel.jsx +++ b/DashAI/front/src/components/models/DatasetPredictionPanel.jsx @@ -29,6 +29,7 @@ export default function DatasetPredictionPanel({ }) { const [datasets, setDatasets] = useState([]); const [selectedDataset, setSelectedDataset] = useState(null); + const [selectedSplit, setSelectedSplit] = useState("all"); const [modelSession, setModelSession] = useState(null); const [loading, setLoading] = useState(true); const [isSubmitting, setIsSubmitting] = useState(false); @@ -91,7 +92,11 @@ export default function DatasetPredictionPanel({ setIsSubmitting(true); try { - const prediction = await createPrediction(run.id, selectedDataset.id); + const prediction = await createPrediction( + run.id, + selectedDataset.id, + selectedSplit !== "all" ? selectedSplit : null, + ); const jobResponse = await enqueuePredictionJob(prediction.id); if (!jobResponse || !jobResponse.id) { @@ -190,6 +195,8 @@ export default function DatasetPredictionPanel({ datasets={datasets} selectedDataset={selectedDataset} setSelectedDataset={setSelectedDataset} + runId={run.id} + onSplitChange={setSelectedSplit} actionSlot={