timeseries_validation

timeseries_validation

Shared time-series validation for preflight and training preprocessing.

Classes:

Name	Description
TimeSeriesValidationReason	Domain-specific reason for a time-series validation failure.
TimeSeriesDataValidationError	Data-side time-series validation failure.
TimeSeriesParameterValidationError	Parameter-side time-series validation failure.
TimeSeriesGroupTimestampStats	Statistics collected from a single group's timestamps.
TimeSeriesValidationResult	Validated time-series data and inferred timestamp metadata.

Functions:

Name	Description
resolve_elapsed_time_column_name	Return an unused generated timestamp column name.
validate_start_stop_consistency	Validate all groups have the same start/stop timestamps.
validate_timeseries_data	Validate time-series data shape and infer timestamp metadata.

TimeSeriesValidationReason

Bases: Enum

Domain-specific reason for a time-series validation failure.

TimeSeriesDataValidationError(reason, message)

Bases: _TimeSeriesValidationError, DataError

Data-side time-series validation failure.

Source code in src/nemo_safe_synthesizer/data_processing/timeseries_validation.py

def __init__(self, reason: TimeSeriesValidationReason, message: str) -> None:
    self.reason = reason
    super().__init__(message)

TimeSeriesParameterValidationError(reason, message)

Bases: _TimeSeriesValidationError, ParameterError

Parameter-side time-series validation failure.

Source code in src/nemo_safe_synthesizer/data_processing/timeseries_validation.py

def __init__(self, reason: TimeSeriesValidationReason, message: str) -> None:
    self.reason = reason
    super().__init__(message)

TimeSeriesGroupTimestampStats(group_name, start_timestamp, stop_timestamp, interval_seconds, record_count=0) dataclass

Statistics collected from a single group's timestamps.

Attributes:

Name	Type	Description
group_name	Any	Identifier for the time-series group.
start_timestamp	Any	First timestamp in the sorted group.
stop_timestamp	Any	Last timestamp in the sorted group.
interval_seconds	int | None	Seconds between consecutive timestamps, or None when no consistent interval exists.
record_count	int	Number of records in the group.

group_name instance-attribute

Identifier for the time-series group.

start_timestamp instance-attribute

First timestamp in the sorted group.

stop_timestamp instance-attribute

Last timestamp in the sorted group.

interval_seconds instance-attribute

Seconds between consecutive timestamps, or None when no consistent interval exists.

record_count = 0 class-attribute instance-attribute

Number of records in the group.

TimeSeriesValidationResult(data, group_by_column, timestamp_column, timestamp_format, is_elapsed_time, timestamp_interval_seconds, start_timestamp, stop_timestamp, group_stats) dataclass

Validated time-series data and inferred timestamp metadata.

Attributes:

Name	Type	Description
data	DataFrame	Validated and timestamp-sorted DataFrame copy.
group_by_column	str	Column used to group time-series records.
timestamp_column	str	Column used as the timestamp after generated-column resolution.
timestamp_format	str	Resolved timestamp format, either "elapsed_seconds" or a strftime format.
is_elapsed_time	bool	Whether timestamps are numeric elapsed seconds.
timestamp_interval_seconds	int | None	Validated or inferred interval between timestamps, if available.
start_timestamp	str	Common start timestamp shared by all groups.
stop_timestamp	str	Common stop timestamp shared by all groups.
group_stats	tuple[TimeSeriesGroupTimestampStats, ...]	Per-group timestamp statistics used to derive the result.

data instance-attribute

Validated and timestamp-sorted DataFrame copy.

group_by_column instance-attribute

Column used to group time-series records.

timestamp_column instance-attribute

Column used as the timestamp after generated-column resolution.

timestamp_format instance-attribute

Resolved timestamp format, either "elapsed_seconds" or a strftime format.

is_elapsed_time instance-attribute

Whether timestamps are numeric elapsed seconds.

timestamp_interval_seconds instance-attribute

Validated or inferred interval between timestamps, if available.

start_timestamp instance-attribute

Common start timestamp shared by all groups.

stop_timestamp instance-attribute

Common stop timestamp shared by all groups.

group_stats instance-attribute

Per-group timestamp statistics used to derive the result.

resolve_elapsed_time_column_name(columns)

Return an unused generated timestamp column name.

Source code in src/nemo_safe_synthesizer/data_processing/timeseries_validation.py

def resolve_elapsed_time_column_name(columns: Iterable[str]) -> str:
    """Return an unused generated timestamp column name."""
    existing = set(columns)
    if "elapsed_seconds" not in existing:
        return "elapsed_seconds"
    if "_elapsed_seconds" not in existing:
        return "_elapsed_seconds"

    suffix = 1
    while True:
        candidate = f"_elapsed_seconds_{suffix}"
        if candidate not in existing:
            return candidate
        suffix += 1

validate_start_stop_consistency(group_stats)

Validate all groups have the same start/stop timestamps.

Source code in src/nemo_safe_synthesizer/data_processing/timeseries_validation.py

def validate_start_stop_consistency(
    group_stats: tuple[TimeSeriesGroupTimestampStats, ...] | list[TimeSeriesGroupTimestampStats],
) -> tuple[str, str]:
    """Validate all groups have the same start/stop timestamps."""
    if not group_stats:
        raise TimeSeriesDataValidationError(
            TimeSeriesValidationReason.TIMESERIES_EMPTY,
            "Time-series data must contain at least one record.",
        )

    unique_starts = set(s.start_timestamp for s in group_stats)
    unique_stops = set(s.stop_timestamp for s in group_stats)

    if len(unique_starts) > 1:
        raise TimeSeriesDataValidationError(
            TimeSeriesValidationReason.TIMESERIES_START_MISMATCH,
            f"Start timestamps differ across groups. Found {len(unique_starts)} different start timestamps: "
            f"{sorted([str(t) for t in list(unique_starts)[:5]])}{'...' if len(unique_starts) > 5 else ''}. "
            f"All groups must have the same start timestamp.",
        )

    if len(unique_stops) > 1:
        raise TimeSeriesDataValidationError(
            TimeSeriesValidationReason.TIMESERIES_STOP_MISMATCH,
            f"Stop timestamps differ across groups. Found {len(unique_stops)} different stop timestamps: "
            f"{sorted([str(t) for t in list(unique_stops)[:5]])}{'...' if len(unique_stops) > 5 else ''}. "
            f"All groups must have the same stop timestamp.",
        )

    return str(group_stats[0].start_timestamp), str(group_stats[0].stop_timestamp)

validate_timeseries_data(data, config)

Validate time-series data shape and infer timestamp metadata.

The validator performs the same timestamp normalization checks needed by training preprocessing, but operates on copies so preflight can run it without mutating the caller's DataFrame or config.

Parameters:

Name	Type	Description	Default
data	DataFrame	Training DataFrame to validate.	required
config	SafeSynthesizerParameters	Safe Synthesizer parameters containing time-series settings.	required

Returns:

Type	Description
TimeSeriesValidationResult	A validation result with a sorted DataFrame copy and resolved timestamp
TimeSeriesValidationResult	metadata suitable for updating the runtime config.

Raises:

Type	Description
TimeSeriesParameterValidationError	If configuration references missing columns or incompatible timestamp formats.
TimeSeriesDataValidationError	If the data violates time-series shape invariants.

Source code in src/nemo_safe_synthesizer/data_processing/timeseries_validation.py

def validate_timeseries_data(data: pd.DataFrame, config: SafeSynthesizerParameters) -> TimeSeriesValidationResult:
    """Validate time-series data shape and infer timestamp metadata.

    The validator performs the same timestamp normalization checks needed by
    training preprocessing, but operates on copies so preflight can run it
    without mutating the caller's DataFrame or config.

    Args:
        data: Training DataFrame to validate.
        config: Safe Synthesizer parameters containing time-series settings.

    Returns:
        A validation result with a sorted DataFrame copy and resolved timestamp
        metadata suitable for updating the runtime config.

    Raises:
        TimeSeriesParameterValidationError: If configuration references missing
            columns or incompatible timestamp formats.
        TimeSeriesDataValidationError: If the data violates time-series shape
            invariants.
    """
    if data.empty:
        raise TimeSeriesDataValidationError(
            TimeSeriesValidationReason.TIMESERIES_EMPTY,
            "Time-series data must contain at least one record.",
        )

    ts_config = config.time_series
    working_df, group_by_col = _resolve_group_column(data, config)

    timestamp_col = ts_config.timestamp_column
    if timestamp_col is None:
        working_df, timestamp_col = _add_elapsed_time_column(working_df, ts_config, group_by_col)
        timestamp_format = "elapsed_seconds"
        is_elapsed_time = True
    else:
        try:
            check_timestamp_column(working_df, timestamp_col)
        except ParameterError as exc:
            raise TimeSeriesParameterValidationError(TimeSeriesValidationReason.TIMESTAMP_NOT_FOUND, str(exc)) from exc
        except DataError as exc:
            raise TimeSeriesDataValidationError(TimeSeriesValidationReason.TIMESTAMP_NULLS, str(exc)) from exc

        is_elapsed_time = _detect_elapsed_seconds_format(working_df, ts_config, timestamp_col)

    if not is_elapsed_time:
        ts_config_copy = ts_config.model_copy(update={"timestamp_column": timestamp_col})
        working_df = _infer_and_convert_timestamp_format(working_df, ts_config_copy)
        timestamp_format = cast(str, ts_config_copy.timestamp_format)
    else:
        timestamp_format = "elapsed_seconds"

    working_df = _sort_by_group_and_timestamp(working_df, group_by_col, timestamp_col)
    group_stats = _collect_group_timestamp_stats(working_df, timestamp_col, group_by_col, is_elapsed_time)
    _validate_equal_group_lengths(group_stats)
    start_ts, stop_ts = validate_start_stop_consistency(group_stats)
    interval_seconds = _validate_interval_consistency(
        ts_config.timestamp_interval_seconds,
        group_stats,
    )

    return TimeSeriesValidationResult(
        data=working_df,
        group_by_column=group_by_col,
        timestamp_column=timestamp_col,
        timestamp_format=timestamp_format,
        is_elapsed_time=is_elapsed_time,
        timestamp_interval_seconds=interval_seconds,
        start_timestamp=start_ts,
        stop_timestamp=stop_ts,
        group_stats=group_stats,
    )