diff --git a/bluemath_tk/__init__.py b/bluemath_tk/__init__.py index 597ea01..919b62d 100644 --- a/bluemath_tk/__init__.py +++ b/bluemath_tk/__init__.py @@ -22,6 +22,7 @@ teslakit, tide, topo_bathy, + validation, waves, wrappers, ) @@ -42,6 +43,7 @@ "teslakit", "tide", "topo_bathy", + "validation", "waves", "wrappers", ] diff --git a/bluemath_tk/validation/__init__.py b/bluemath_tk/validation/__init__.py new file mode 100644 index 0000000..33dd0a6 --- /dev/null +++ b/bluemath_tk/validation/__init__.py @@ -0,0 +1,17 @@ +"""Reproducible validation utilities for climate and environmental data.""" + +from .chronological import ( + ChronologicalSplit, + RealScalar, + ValidationSplitManifest, + apply_split_manifest, + split_chronologically, +) + +__all__ = [ + "ChronologicalSplit", + "RealScalar", + "ValidationSplitManifest", + "apply_split_manifest", + "split_chronologically", +] diff --git a/bluemath_tk/validation/chronological.py b/bluemath_tk/validation/chronological.py new file mode 100644 index 0000000..1d062a1 --- /dev/null +++ b/bluemath_tk/validation/chronological.py @@ -0,0 +1,1439 @@ +"""Reproducible chronological train, validation, and test splitting.""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import date, datetime +from fractions import Fraction +from numbers import Integral, Real +from pathlib import Path +from types import MappingProxyType +from typing import Any, TypeAlias + +import numpy as np +import pandas as pd + +_SCHEMA_VERSION = 2 +_FINGERPRINT_SCHEMA = "bluemath-time-axis-v3" +_DEFAULT_FRACTIONS = (0.7, 0.15, 0.15) +_BOUNDARY_POLICY = "complete_interval_half_open" +_PARTITION_CLOSURE = "train:end=b1,end=b2" +_ROUNDING_POLICY = "cumulative_floor" +_SUPPORTED_TIME_KINDS = { + "datetime64[ns]-naive", + "datetime64[ns]-aware-utc", + "float64", + "integer-signed", + "integer-unsigned", +} + +JsonScalar = str | int | float | bool | None +JsonValue = JsonScalar | list["JsonValue"] | dict[str, "JsonValue"] +RealScalar: TypeAlias = Real | np.floating[Any] | np.integer[Any] +CanonicalFraction = tuple[Fraction, str] + +_FRACTION_SUM_TOLERANCE = Fraction(1, 10_000_000) +_FIXED_DATETIME_UNIT_TO_NS: dict[str, Fraction] = { + "W": Fraction(604_800_000_000_000, 1), + "D": Fraction(86_400_000_000_000, 1), + "h": Fraction(3_600_000_000_000, 1), + "m": Fraction(60_000_000_000, 1), + "s": Fraction(1_000_000_000, 1), + "ms": Fraction(1_000_000, 1), + "us": Fraction(1_000, 1), + "ns": Fraction(1, 1), + "ps": Fraction(1, 1_000), + "fs": Fraction(1, 1_000_000), + "as": Fraction(1, 1_000_000_000), +} + + +def _is_exact_integer(value: Any) -> bool: + return isinstance(value, Integral) and not isinstance(value, (bool, np.bool_)) + + +def _validate_index_values( + values: Sequence[int] | np.ndarray, + *, + name: str, + n_samples: int, +) -> tuple[int, ...]: + if isinstance(values, (str, bytes)): + raise TypeError(f"{name} must be a one-dimensional sequence of integers.") + array = np.asarray(values, dtype=object) + if array.ndim != 1: + raise ValueError(f"{name} must be one-dimensional.") + normalized: list[int] = [] + for value in array.tolist(): + if not _is_exact_integer(value): + raise TypeError(f"{name} must contain exact non-Boolean integer values.") + normalized.append(int(value)) + result = tuple(normalized) + if result != tuple(sorted(result)): + raise ValueError(f"{name} must be sorted in ascending order.") + if len(set(result)) != len(result): + raise ValueError(f"{name} must not contain repeated indices.") + if any(value < 0 or value >= n_samples for value in result): + raise ValueError(f"{name} contains an index outside [0, n_samples).") + return result + + +def _validate_manifest_index_values( + values: Any, + *, + name: str, + n_samples: int, +) -> tuple[int, ...]: + if type(values) not in {list, tuple}: + raise TypeError(f"{name} must be a JSON-style list or tuple of integers.") + if any(type(value) is not int for value in values): + raise TypeError(f"{name} must contain exact built-in integer values.") + result = tuple(values) + if result != tuple(sorted(result)): + raise ValueError(f"{name} must be sorted in ascending order.") + if len(set(result)) != len(result): + raise ValueError(f"{name} must not contain repeated indices.") + if any(value < 0 or value >= n_samples for value in result): + raise ValueError(f"{name} contains an index outside [0, n_samples).") + return result + + +def _readonly_int_array( + values: Sequence[int] | np.ndarray, + *, + name: str, + n_samples: int, +) -> np.ndarray: + normalized = _validate_index_values(values, name=name, n_samples=n_samples) + array = np.array(normalized, dtype=np.int64, copy=True) + array.setflags(write=False) + return array + + +def _validate_json_value(value: Any, *, path: str) -> JsonValue: + if value is None or type(value) in {str, bool, int}: + return value + if type(value) is float: + if not math.isfinite(value): + raise ValueError(f"{path} must not contain NaN or infinity.") + return value + if type(value) is list: + return [ + _validate_json_value(item, path=f"{path}[{index}]") + for index, item in enumerate(value) + ] + if type(value) is dict: + validated: dict[str, JsonValue] = {} + for key, item in value.items(): + if type(key) is not str: + raise TypeError(f"{path} must use string object keys.") + validated[key] = _validate_json_value(item, path=f"{path}.{key}") + return validated + raise TypeError(f"{path} contains a non-JSON value of type {type(value).__name__}.") + + +def _freeze_json(value: JsonValue) -> Any: + if isinstance(value, dict): + return MappingProxyType( + {key: _freeze_json(item) for key, item in value.items()} + ) + if isinstance(value, list): + return tuple(_freeze_json(item) for item in value) + return value + + +def _thaw_json(value: Any) -> JsonValue: + if isinstance(value, Mapping): + return {str(key): _thaw_json(item) for key, item in value.items()} + if isinstance(value, tuple): + return [_thaw_json(item) for item in value] + return value + + +def _canonical_json(payload: Mapping[str, Any], *, indent: int | None = None) -> str: + return json.dumps( + payload, + sort_keys=True, + separators=(",", ":") if indent is None else None, + indent=indent, + ensure_ascii=True, + allow_nan=False, + ) + + +def _json_loads_strict(text: str) -> Any: + def reject_constant(value: str) -> None: + raise ValueError(f"Non-standard JSON constant {value!r} is not supported.") + + return json.loads(text, parse_constant=reject_constant) + + +def _require_sequence(value: Any, *, name: str, length: int) -> list[Any]: + if isinstance(value, (str, bytes)): + raise TypeError( + f"{name} must be a one-dimensional sequence of length {length}." + ) + if isinstance(value, np.ndarray): + array = np.asarray(value) + else: + array = np.asarray(value, dtype=object) + if array.ndim == 0: + raise TypeError( + f"{name} must be a one-dimensional sequence of length {length}." + ) + if array.ndim != 1: + raise ValueError(f"{name} must be one-dimensional.") + if array.size != length: + raise ValueError(f"{name} must contain exactly {length} values.") + return [array[index] for index in range(array.size)] + + +def _validate_gap(gap: Any) -> int: + if not _is_exact_integer(gap): + raise TypeError("gap must be a non-negative integer sample count.") + gap_int = int(gap) + if gap_int < 0: + raise ValueError("gap must be non-negative.") + return gap_int + + +def _canonicalize_fraction_scalar(value: Any, *, name: str) -> CanonicalFraction: + if isinstance(value, (bool, np.bool_)) or not isinstance(value, Real): + raise TypeError(f"{name} must be a finite real number, not Boolean.") + if isinstance(value, Fraction): + fraction = value + elif isinstance(value, np.floating): + if not bool(np.isfinite(value)): + raise ValueError(f"{name} must be finite.") + fraction = Fraction(str(value)) + elif isinstance(value, float): + if not math.isfinite(value): + raise ValueError(f"{name} must be finite.") + fraction = Fraction(str(value)) + elif isinstance(value, Integral): + fraction = Fraction(int(value), 1) + else: + try: + number = float(value) + except (TypeError, ValueError, OverflowError) as exc: + raise TypeError(f"{name} must be a finite real number.") from exc + if not math.isfinite(number): + raise ValueError(f"{name} must be finite.") + fraction = Fraction(str(value)) + if fraction <= 0 or fraction >= 1: + raise ValueError(f"{name} must be strictly between zero and one.") + canonical = f"{fraction.numerator}/{fraction.denominator}" + return fraction, canonical + + +def _fraction_from_canonical(value: Any, *, name: str) -> Fraction: + if ( + type(value) is not str + or re.fullmatch( + r"[1-9][0-9]*/[1-9][0-9]*", + value, + ) + is None + ): + raise TypeError( + f"{name} must be a canonical positive numerator/denominator string." + ) + numerator_text, denominator_text = value.split("/", maxsplit=1) + fraction = Fraction(int(numerator_text), int(denominator_text)) + if fraction <= 0 or fraction >= 1: + raise ValueError( + f"{name} must represent a value strictly between zero and one." + ) + if value != f"{fraction.numerator}/{fraction.denominator}": + raise ValueError(f"{name} must use reduced canonical fraction form.") + return fraction + + +def _validate_runtime_fractions( + fractions: Any, +) -> tuple[tuple[Fraction, Fraction, Fraction], tuple[str, str, str]]: + values = _require_sequence(fractions, name="fractions", length=3) + canonicalized = tuple( + _canonicalize_fraction_scalar(value, name=f"fractions[{index}]") + for index, value in enumerate(values) + ) + exact = tuple(item[0] for item in canonicalized) + canonical = tuple(item[1] for item in canonicalized) + if abs(sum(exact, start=Fraction(0, 1)) - 1) > _FRACTION_SUM_TOLERANCE: + raise ValueError("Train, validation, and test fractions must sum to 1.0.") + return exact, canonical + + +def _validate_manifest_fractions( + fractions: Any, +) -> tuple[tuple[Fraction, Fraction, Fraction], tuple[str, str, str]]: + values = _require_sequence( + fractions, + name="parameters.fractions", + length=3, + ) + exact = tuple( + _fraction_from_canonical( + value, + name=f"parameters.fractions[{index}]", + ) + for index, value in enumerate(values) + ) + if abs(sum(exact, start=Fraction(0, 1)) - 1) > _FRACTION_SUM_TOLERANCE: + raise ValueError("Manifest fractions must sum to 1.0.") + return exact, tuple(values) + + +def _fraction_boundary_indices( + n_samples: int, + fractions: tuple[Fraction, Fraction, Fraction], +) -> tuple[int, int]: + """Resolve cumulative-floor boundaries using exact rational arithmetic.""" + train, validation, _ = fractions + validation_index = (n_samples * train.numerator) // train.denominator + cumulative = train + validation + test_index = (n_samples * cumulative.numerator) // cumulative.denominator + return validation_index, test_index + + +def _validate_index_boundary(value: Any, *, name: str, n_samples: int) -> int: + if not _is_exact_integer(value): + raise TypeError(f"{name} must be an integer index.") + index = int(value) + if index <= 0 or index >= n_samples: + raise ValueError(f"{name} must lie strictly inside [0, n_samples].") + return index + + +def _is_datetime_object(value: Any) -> bool: + return isinstance(value, (datetime, date, np.datetime64, pd.Timestamp)) + + +def _datetime_is_aware(value: Any) -> bool: + if isinstance(value, np.datetime64) or ( + isinstance(value, date) and not isinstance(value, datetime) + ): + return False + timestamp = pd.Timestamp(value) + return timestamp.tzinfo is not None and timestamp.utcoffset() is not None + + +def _datetime_to_ns(value: Any, *, name: str, aware: bool) -> int: + try: + timestamp = pd.Timestamp(value) + except (TypeError, ValueError, OverflowError) as exc: + raise ValueError( + f"{name} contains an invalid datetime value: {value!r}." + ) from exc + if pd.isna(timestamp): + raise ValueError(f"{name} contains NaT values.") + actual_aware = timestamp.tzinfo is not None and timestamp.utcoffset() is not None + if actual_aware != aware: + state = "timezone-aware" if aware else "timezone-naive" + raise ValueError(f"{name} must contain only {state} datetime values.") + if aware: + timestamp = timestamp.tz_convert("UTC") + try: + timestamp_ns = timestamp.as_unit("ns", round_ok=False) + value_ns = int(timestamp_ns.value) + except (ValueError, OverflowError) as exc: + raise ValueError( + f"{name} contains a datetime that cannot be represented exactly at " + "nanosecond resolution." + ) from exc + try: + round_trip = pd.Timestamp( + value_ns, + unit="ns", + tz="UTC" if aware else None, + ) + except (ValueError, OverflowError) as exc: + raise ValueError( + f"{name} contains a datetime outside the supported nanosecond range." + ) from exc + if round_trip != timestamp_ns: + raise ValueError( + f"{name} contains a datetime that failed the nanosecond round-trip check." + ) + return value_ns + + +def _numpy_datetime_scalar_to_ns(value: np.datetime64, *, name: str) -> int: + if np.isnat(value): + raise ValueError(f"{name} contains NaT values.") + unit, step = np.datetime_data(value.dtype) + if unit == "generic": + raise ValueError(f"{name} uses an unsupported generic datetime64 unit.") + raw = int(value.astype(np.int64)) + if unit in {"Y", "M"}: + calendar_offset = raw * step + if unit == "Y": + year = 1970 + calendar_offset + month = None + else: + year_offset, month_index = divmod(calendar_offset, 12) + year = 1970 + year_offset + month = month_index + 1 + if year < 1677 or year > 2262: + raise ValueError( + f"{name} contains a datetime outside the supported nanosecond range." + ) + text = f"{year:04d}" if month is None else f"{year:04d}-{month:02d}" + return _datetime_to_ns(text, name=name, aware=False) + scale = _FIXED_DATETIME_UNIT_TO_NS.get(unit) + if scale is None: + raise ValueError(f"{name} uses unsupported datetime64 unit {unit!r}.") + exact_ns = Fraction(raw * step, 1) * scale + if exact_ns.denominator != 1: + raise ValueError( + f"{name} contains a datetime64[{unit}] value that is not an exact " + "nanosecond multiple." + ) + value_ns = exact_ns.numerator + if value_ns < np.iinfo(np.int64).min + 1 or value_ns > np.iinfo(np.int64).max: + raise ValueError( + f"{name} contains a datetime outside the supported nanosecond range." + ) + return int(value_ns) + + +def _normalize_numpy_datetime_array( + array: np.ndarray, + *, + name: str, +) -> tuple[np.ndarray, str, tuple[str, ...]]: + integers = np.array( + [_numpy_datetime_scalar_to_ns(value, name=name) for value in array], + dtype=np.int64, + ) + return ( + integers, + "datetime64[ns]-naive", + tuple(str(int(value)) for value in integers), + ) + + +def _normalize_datetime_values( + items: list[Any], + *, + name: str, +) -> tuple[np.ndarray, str, tuple[str, ...]]: + awareness = [_datetime_is_aware(item) for item in items] + if any(awareness) and not all(awareness): + raise ValueError(f"{name} mixes timezone-aware and timezone-naive values.") + aware = all(awareness) + normalized: list[int] = [] + for item in items: + if isinstance(item, np.datetime64): + if aware: + raise ValueError( + f"{name} mixes timezone-aware values with NumPy datetime64 values." + ) + normalized.append(_numpy_datetime_scalar_to_ns(item, name=name)) + else: + normalized.append(_datetime_to_ns(item, name=name, aware=aware)) + integers = np.array(normalized, dtype=np.int64) + kind = "datetime64[ns]-aware-utc" if aware else "datetime64[ns]-naive" + return integers, kind, tuple(str(int(value)) for value in integers) + + +def _normalize_integer_values( + array: np.ndarray, + *, + name: str, +) -> tuple[np.ndarray, str, tuple[str, ...]]: + unsigned = np.issubdtype(array.dtype, np.unsignedinteger) + if unsigned: + maximum = int(np.max(array)) + if maximum > np.iinfo(np.int64).max: + raise ValueError( + f"{name} contains unsigned integers outside the supported int64 range." + ) + kind = "integer-unsigned" + else: + kind = "integer-signed" + integers = np.asarray(array, dtype=np.int64) + return integers, kind, tuple(str(int(value)) for value in integers) + + +def _normalize_time_values( + values: Any, + *, + name: str, +) -> tuple[np.ndarray, str, tuple[str, ...]]: + if isinstance(values, (str, bytes)): + raise TypeError(f"{name} must be a one-dimensional sequence, not a string.") + array = np.asarray(values) + if array.ndim == 0: + raise TypeError(f"{name} must be a one-dimensional sequence.") + if array.ndim != 1: + raise ValueError(f"{name} must be one-dimensional.") + if array.size == 0: + raise ValueError(f"{name} must not be empty.") + + if np.issubdtype(array.dtype, np.datetime64): + return _normalize_numpy_datetime_array(array, name=name) + if np.issubdtype(array.dtype, np.bool_): + raise TypeError(f"{name} must not contain Boolean values.") + if np.issubdtype(array.dtype, np.integer): + return _normalize_integer_values(array, name=name) + if np.issubdtype(array.dtype, np.floating): + floats = np.asarray(array, dtype=np.float64) + if not np.all(np.isfinite(floats)): + raise ValueError(f"{name} contains NaN or infinite values.") + return floats, "float64", tuple(float(value).hex() for value in floats) + + if array.dtype == object: + items = array.tolist() + if all(_is_datetime_object(item) for item in items): + return _normalize_datetime_values(items, name=name) + if all(_is_exact_integer(item) for item in items): + if any(int(item) < np.iinfo(np.int64).min for item in items) or any( + int(item) > np.iinfo(np.int64).max for item in items + ): + raise ValueError( + f"{name} contains integers outside the supported int64 range." + ) + unsigned = all(isinstance(item, np.unsignedinteger) for item in items) + integers = np.array([int(item) for item in items], dtype=np.int64) + kind = "integer-unsigned" if unsigned else "integer-signed" + return integers, kind, tuple(str(int(value)) for value in integers) + if all( + isinstance(item, Real) and not isinstance(item, (bool, np.bool_)) + for item in items + ): + floats = np.asarray([float(item) for item in items], dtype=np.float64) + if not np.all(np.isfinite(floats)): + raise ValueError(f"{name} contains NaN or infinite values.") + return floats, "float64", tuple(float(value).hex() for value in floats) + + raise TypeError( + f"{name} must contain real numeric values or datetime-like values; " + f"received dtype {array.dtype}." + ) + + +@dataclass(frozen=True) +class _TimeAxis: + starts: np.ndarray + ends: np.ndarray + time_kind: str + axis_mode: str + canonical_starts: tuple[str, ...] + canonical_ends: tuple[str, ...] + + @property + def n_samples(self) -> int: + return int(self.starts.shape[0]) + + +def _prepare_time_axis( + *, + sample_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None, + sample_start_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None, + sample_end_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None, +) -> _TimeAxis: + uses_points = sample_times is not None + uses_intervals = sample_start_times is not None or sample_end_times is not None + if uses_points and uses_intervals: + raise ValueError( + "Provide sample_times for point samples or both sample_start_times and " + "sample_end_times for interval samples, not both forms." + ) + if not uses_points and not uses_intervals: + raise ValueError( + "Provide sample_times or both sample_start_times and sample_end_times." + ) + if uses_intervals and (sample_start_times is None or sample_end_times is None): + raise ValueError( + "sample_start_times and sample_end_times must be provided together." + ) + + if uses_points: + starts, time_kind, canonical_starts = _normalize_time_values( + sample_times, + name="sample_times", + ) + ends = np.array(starts, copy=True) + canonical_ends = canonical_starts + axis_mode = "point" + else: + starts, start_kind, canonical_starts = _normalize_time_values( + sample_start_times, + name="sample_start_times", + ) + ends, end_kind, canonical_ends = _normalize_time_values( + sample_end_times, + name="sample_end_times", + ) + if start_kind != end_kind: + raise TypeError( + "sample_start_times and sample_end_times must use the same time kind " + "and timezone-awareness state." + ) + time_kind = start_kind + axis_mode = "interval" + if starts.shape != ends.shape: + raise ValueError( + "sample_start_times and sample_end_times must have the same length." + ) + if np.any(ends < starts): + raise ValueError( + "Every sample interval must satisfy end_time >= start_time." + ) + + if starts.size < 3: + raise ValueError("At least three chronological samples are required.") + if np.any(starts[1:] <= starts[:-1]): + raise ValueError( + "Sample start times must be strictly increasing with no duplicates." + ) + + starts_copy = np.array(starts, copy=True) + ends_copy = np.array(ends, copy=True) + starts_copy.setflags(write=False) + ends_copy.setflags(write=False) + return _TimeAxis( + starts=starts_copy, + ends=ends_copy, + time_kind=time_kind, + axis_mode=axis_mode, + canonical_starts=canonical_starts, + canonical_ends=canonical_ends, + ) + + +def _fingerprint_time_axis(axis: _TimeAxis) -> str: + payload = { + "schema": _FINGERPRINT_SCHEMA, + "axis_mode": axis.axis_mode, + "time_kind": axis.time_kind, + "starts": list(axis.canonical_starts), + "ends": list(axis.canonical_ends), + } + return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() + + +def _normalize_boundary_scalar( + value: Any, + *, + time_kind: str, + name: str, +) -> tuple[Any, str]: + if time_kind.startswith("datetime64[ns]"): + if not _is_datetime_object(value): + raise TypeError(f"{name} must be a datetime-like value.") + normalized, scalar_kind, canonical = _normalize_datetime_values( + [value], + name=name, + ) + if scalar_kind != time_kind: + raise TypeError( + f"{name} uses time kind {scalar_kind!r}; expected {time_kind!r}." + ) + return normalized[0], canonical[0] + if time_kind == "float64": + if isinstance(value, (bool, np.bool_)) or not isinstance(value, Real): + raise TypeError(f"{name} must be a finite real number.") + number = float(value) + if not math.isfinite(number): + raise ValueError(f"{name} must be finite.") + return number, number.hex() + if time_kind in {"integer-signed", "integer-unsigned"}: + if not _is_exact_integer(value): + raise TypeError(f"{name} must be an integer value.") + number = int(value) + if number < np.iinfo(np.int64).min or number > np.iinfo(np.int64).max: + raise ValueError(f"{name} lies outside the supported int64 range.") + if time_kind == "integer-unsigned" and number < 0: + raise ValueError(f"{name} must be non-negative for unsigned coordinates.") + return number, str(number) + raise ValueError(f"Unsupported time kind {time_kind!r}.") + + +def _canonical_to_scalar(value: str, *, time_kind: str, name: str) -> Any: + if type(value) is not str: + raise TypeError(f"{name} must be a canonical string value.") + try: + if time_kind == "float64": + result = float.fromhex(value) + if not math.isfinite(result): + raise ValueError + return result + if time_kind in { + "integer-signed", + "integer-unsigned", + "datetime64[ns]-naive", + "datetime64[ns]-aware-utc", + }: + result = int(value) + if result < np.iinfo(np.int64).min or result > np.iinfo(np.int64).max: + raise ValueError + if time_kind == "integer-unsigned" and result < 0: + raise ValueError + return result + except (TypeError, ValueError) as exc: + raise ValueError( + f"{name} is not a valid canonical value for {time_kind!r}." + ) from exc + raise ValueError(f"Unsupported time kind {time_kind!r}.") + + +def _base_parameters(gap: int) -> dict[str, JsonValue]: + return { + "boundary_policy": _BOUNDARY_POLICY, + "partition_closure": _PARTITION_CLOSURE, + "gap_samples_before_later_partition": gap, + } + + +def _validate_parameter_keys( + parameters: dict[str, JsonValue], + *, + required: set[str], +) -> None: + missing = sorted(required.difference(parameters)) + extra = sorted(set(parameters).difference(required)) + if missing: + raise ValueError(f"Manifest parameters are missing required fields: {missing}.") + if extra: + raise ValueError(f"Manifest parameters contain unsupported fields: {extra}.") + + +def _validate_string_list(value: Any, *, name: str) -> list[str]: + values = _require_sequence(value, name=name, length=2) + if any(type(item) is not str for item in values): + raise TypeError(f"{name} must contain canonical string values.") + return [str(item) for item in values] + + +def _validate_manifest_parameters( + method: str, + parameters: Mapping[str, Any], + *, + n_samples: int, + time_kind: str, +) -> dict[str, JsonValue]: + if not isinstance(parameters, Mapping): + raise TypeError("parameters must be a mapping.") + raw = dict(parameters) + validated_json = _validate_json_value(raw, path="parameters") + if not isinstance(validated_json, dict): + raise TypeError("parameters must be a JSON object.") + common = { + "boundary_policy", + "partition_closure", + "gap_samples_before_later_partition", + } + if method == "fractions": + required = common | { + "fractions", + "rounding_policy", + "resolved_boundary_indices", + "resolved_boundary_values", + } + elif method == "boundary_indices": + required = common | {"boundary_indices", "resolved_boundary_values"} + elif method == "boundary_times": + required = common | {"boundary_times"} + else: + raise ValueError(f"Unsupported split method: {method!r}.") + _validate_parameter_keys(validated_json, required=required) + + if validated_json["boundary_policy"] != _BOUNDARY_POLICY: + raise ValueError("Manifest boundary_policy is unsupported or inconsistent.") + if validated_json["partition_closure"] != _PARTITION_CLOSURE: + raise ValueError("Manifest partition_closure is unsupported or inconsistent.") + gap = _validate_gap(validated_json["gap_samples_before_later_partition"]) + validated_json["gap_samples_before_later_partition"] = gap + + if method == "fractions": + if validated_json["rounding_policy"] != _ROUNDING_POLICY: + raise ValueError("Manifest rounding_policy is unsupported.") + fractions, canonical_fractions = _validate_manifest_fractions( + validated_json["fractions"] + ) + validation_index, test_index = _fraction_boundary_indices( + n_samples, + fractions, + ) + resolved = _require_sequence( + validated_json["resolved_boundary_indices"], + name="parameters.resolved_boundary_indices", + length=2, + ) + if not all(_is_exact_integer(value) for value in resolved): + raise TypeError( + "parameters.resolved_boundary_indices must contain exact integers." + ) + resolved_indices = [int(value) for value in resolved] + if resolved_indices != [validation_index, test_index]: + raise ValueError( + "Manifest resolved_boundary_indices contradict the stored fractions." + ) + if ( + validation_index <= 0 + or test_index <= validation_index + or test_index >= n_samples + ): + raise ValueError( + "Manifest fractions do not produce non-empty candidate partitions." + ) + resolved_values = _validate_string_list( + validated_json["resolved_boundary_values"], + name="parameters.resolved_boundary_values", + ) + for index, value in enumerate(resolved_values): + _canonical_to_scalar( + value, + time_kind=time_kind, + name=f"parameters.resolved_boundary_values[{index}]", + ) + validated_json["fractions"] = list(canonical_fractions) + validated_json["resolved_boundary_indices"] = resolved_indices + validated_json["resolved_boundary_values"] = resolved_values + elif method == "boundary_indices": + values = _require_sequence( + validated_json["boundary_indices"], + name="parameters.boundary_indices", + length=2, + ) + validation_index = _validate_index_boundary( + values[0], + name="parameters.validation_start_index", + n_samples=n_samples, + ) + test_index = _validate_index_boundary( + values[1], + name="parameters.test_start_index", + n_samples=n_samples, + ) + if validation_index >= test_index: + raise ValueError( + "Manifest validation_start_index must be less than test_start_index." + ) + resolved_values = _validate_string_list( + validated_json["resolved_boundary_values"], + name="parameters.resolved_boundary_values", + ) + for index, value in enumerate(resolved_values): + _canonical_to_scalar( + value, + time_kind=time_kind, + name=f"parameters.resolved_boundary_values[{index}]", + ) + validated_json["boundary_indices"] = [validation_index, test_index] + validated_json["resolved_boundary_values"] = resolved_values + else: + boundary_values = _validate_string_list( + validated_json["boundary_times"], + name="parameters.boundary_times", + ) + normalized = [ + _canonical_to_scalar( + value, + time_kind=time_kind, + name=f"parameters.boundary_times[{index}]", + ) + for index, value in enumerate(boundary_values) + ] + if normalized[0] >= normalized[1]: + raise ValueError( + "Manifest validation_start_time must precede test_start_time." + ) + validated_json["boundary_times"] = boundary_values + return validated_json + + +def _classify_complete_intervals( + axis: _TimeAxis, + *, + validation_boundary: Any, + test_boundary: Any, + gap: int, +) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: + indices = np.arange(axis.n_samples, dtype=np.int64) + train = indices[axis.ends < validation_boundary] + validation = indices[ + (axis.starts >= validation_boundary) & (axis.ends < test_boundary) + ] + test = indices[axis.starts >= test_boundary] + + if gap: + if train.size <= gap: + raise ValueError("gap removes every training sample.") + if validation.size <= gap: + raise ValueError("gap removes every validation sample.") + train = train[:-gap] + validation = validation[:-gap] + + included = np.concatenate([train, validation, test]) + excluded = np.setdiff1d(indices, included, assume_unique=False) + if train.size == 0 or validation.size == 0 or test.size == 0: + raise ValueError( + "The requested boundaries, interval policy, and gap must leave non-empty " + "train, validation, and test partitions." + ) + return train, validation, test, excluded + + +def _derive_split_from_parameters( + axis: _TimeAxis, + *, + method: str, + parameters: Mapping[str, Any], +) -> tuple[ + tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray], + dict[str, JsonValue], +]: + validated = _validate_manifest_parameters( + method, + parameters, + n_samples=axis.n_samples, + time_kind=axis.time_kind, + ) + gap = int(validated["gap_samples_before_later_partition"]) + if method == "fractions": + validation_index, test_index = validated["resolved_boundary_indices"] + expected_values = [ + axis.canonical_starts[validation_index], + axis.canonical_starts[test_index], + ] + if validated["resolved_boundary_values"] != expected_values: + raise ValueError( + "Manifest resolved boundary values contradict the supplied time axis." + ) + validation_boundary = axis.starts[validation_index] + test_boundary = axis.starts[test_index] + elif method == "boundary_indices": + validation_index, test_index = validated["boundary_indices"] + expected_values = [ + axis.canonical_starts[validation_index], + axis.canonical_starts[test_index], + ] + if validated["resolved_boundary_values"] != expected_values: + raise ValueError( + "Manifest resolved boundary values contradict the supplied time axis." + ) + validation_boundary = axis.starts[validation_index] + test_boundary = axis.starts[test_index] + else: + validation_boundary, test_boundary = [ + _canonical_to_scalar( + value, + time_kind=axis.time_kind, + name=f"parameters.boundary_times[{index}]", + ) + for index, value in enumerate(validated["boundary_times"]) + ] + partitions = _classify_complete_intervals( + axis, + validation_boundary=validation_boundary, + test_boundary=test_boundary, + gap=gap, + ) + return partitions, validated + + +@dataclass(frozen=True) +class ValidationSplitManifest: + """Serializable description of a reproducible chronological split.""" + + method: str + n_samples: int + dataset_fingerprint: str + time_kind: str + axis_mode: str + parameters: Mapping[str, Any] + train_indices: tuple[int, ...] + validation_indices: tuple[int, ...] + test_indices: tuple[int, ...] + excluded_indices: tuple[int, ...] = () + schema_version: int = _SCHEMA_VERSION + + def __post_init__(self) -> None: + """Validate and freeze manifest fields after dataclass construction.""" + if type(self.schema_version) is not int: + raise TypeError("schema_version must be an exact non-Boolean integer.") + if self.schema_version != _SCHEMA_VERSION: + raise ValueError( + f"Unsupported split-manifest schema version {self.schema_version}; " + f"expected {_SCHEMA_VERSION}." + ) + if type(self.n_samples) is not int: + raise TypeError("n_samples must be an exact non-Boolean integer.") + if self.n_samples < 3: + raise ValueError("n_samples must be at least 3.") + if type(self.method) is not str: + raise TypeError("method must be an exact built-in string.") + if self.method not in { + "fractions", + "boundary_indices", + "boundary_times", + }: + raise ValueError(f"Unsupported split method: {self.method!r}.") + if type(self.axis_mode) is not str: + raise TypeError("axis_mode must be an exact built-in string.") + if self.axis_mode not in {"point", "interval"}: + raise ValueError("axis_mode must be 'point' or 'interval'.") + if type(self.time_kind) is not str: + raise TypeError("time_kind must be an exact built-in string.") + if self.time_kind not in _SUPPORTED_TIME_KINDS: + raise ValueError(f"Unsupported time_kind: {self.time_kind!r}.") + if type(self.dataset_fingerprint) is not str: + raise TypeError("dataset_fingerprint must be an exact built-in string.") + if re.fullmatch(r"[0-9a-f]{64}", self.dataset_fingerprint) is None: + raise ValueError( + "dataset_fingerprint must be a lowercase 64-character SHA-256 " + "hex digest." + ) + + validated_parameters = _validate_manifest_parameters( + self.method, + self.parameters, + n_samples=self.n_samples, + time_kind=self.time_kind, + ) + object.__setattr__( + self, + "parameters", + _freeze_json(validated_parameters), + ) + + partitions = { + "train_indices": self.train_indices, + "validation_indices": self.validation_indices, + "test_indices": self.test_indices, + "excluded_indices": self.excluded_indices, + } + normalized: dict[str, tuple[int, ...]] = {} + for name, values in partitions.items(): + normalized[name] = _validate_manifest_index_values( + values, + name=name, + n_samples=self.n_samples, + ) + object.__setattr__(self, name, normalized[name]) + + for name in ("train_indices", "validation_indices", "test_indices"): + if not normalized[name]: + raise ValueError(f"{name} must not be empty.") + seen: set[int] = set() + for name, values in normalized.items(): + overlap = seen.intersection(values) + if overlap: + raise ValueError( + f"{name} overlaps another partition at indices {sorted(overlap)}." + ) + seen.update(values) + if seen != set(range(self.n_samples)): + missing = sorted(set(range(self.n_samples)).difference(seen)) + raise ValueError( + "Split manifest must classify every sample as train, validation, test, " + f"or excluded; missing indices: {missing}." + ) + if max(self.train_indices) >= min(self.validation_indices): + raise ValueError("Training indices must precede validation indices.") + if max(self.validation_indices) >= min(self.test_indices): + raise ValueError("Validation indices must precede test indices.") + + def to_dict(self) -> dict[str, JsonValue]: + """Return a JSON-compatible dictionary with deterministic field content.""" + return { + "schema_version": self.schema_version, + "method": self.method, + "n_samples": self.n_samples, + "dataset_fingerprint": self.dataset_fingerprint, + "time_kind": self.time_kind, + "axis_mode": self.axis_mode, + "parameters": _thaw_json(self.parameters), + "train_indices": list(self.train_indices), + "validation_indices": list(self.validation_indices), + "test_indices": list(self.test_indices), + "excluded_indices": list(self.excluded_indices), + } + + def to_json(self, *, indent: int | None = 2) -> str: + """Serialize the manifest deterministically as strict JSON.""" + return _canonical_json(self.to_dict(), indent=indent) + "\n" + + def save(self, path: str | Path) -> None: + """Write the manifest to a UTF-8 JSON file.""" + destination = Path(path) + destination.write_text(self.to_json(indent=2), encoding="utf-8", newline="\n") + + @classmethod + def from_dict(cls, payload: Mapping[str, Any]) -> ValidationSplitManifest: + """Construct a validated manifest from a dictionary.""" + if not isinstance(payload, Mapping): + raise TypeError("Manifest payload must be a mapping.") + if any(type(key) is not str for key in payload): + raise TypeError("Manifest field names must be exact built-in strings.") + required = { + "schema_version", + "method", + "n_samples", + "dataset_fingerprint", + "time_kind", + "axis_mode", + "parameters", + "train_indices", + "validation_indices", + "test_indices", + "excluded_indices", + } + missing = sorted(required.difference(payload)) + extra = sorted(set(payload).difference(required)) + if missing: + raise ValueError(f"Manifest is missing required fields: {missing}.") + if extra: + raise ValueError(f"Manifest contains unsupported fields: {extra}.") + if type(payload["parameters"]) is not dict: + raise TypeError("Manifest parameters must be an exact JSON object.") + for field_name in ( + "train_indices", + "validation_indices", + "test_indices", + "excluded_indices", + ): + if type(payload[field_name]) is not list: + raise TypeError(f"Manifest {field_name} must be an exact JSON list.") + _validate_json_value(dict(payload), path="manifest") + return cls(**{key: payload[key] for key in required}) + + @classmethod + def load(cls, path: str | Path) -> ValidationSplitManifest: + """Load and validate a manifest from a UTF-8 JSON file.""" + source = Path(path) + try: + payload = _json_loads_strict(source.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise ValueError(f"Invalid split-manifest JSON: {exc}.") from exc + if not isinstance(payload, dict): + raise ValueError("Split-manifest JSON must contain one object.") + return cls.from_dict(payload) + + def to_split(self) -> ChronologicalSplit: + """Return immutable NumPy index arrays for this manifest.""" + return ChronologicalSplit( + train_indices=self.train_indices, + validation_indices=self.validation_indices, + test_indices=self.test_indices, + excluded_indices=self.excluded_indices, + manifest=self, + ) + + def validate_against( + self, + *, + sample_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None = None, + sample_start_times: ( + Sequence[Any] | np.ndarray | pd.Index | pd.Series | None + ) = None, + sample_end_times: ( + Sequence[Any] | np.ndarray | pd.Index | pd.Series | None + ) = None, + ) -> None: + """Reject changed data and manifests inconsistent with their parameters.""" + try: + axis = _prepare_time_axis( + sample_times=sample_times, + sample_start_times=sample_start_times, + sample_end_times=sample_end_times, + ) + except (TypeError, ValueError) as exc: + raise ValueError( + "The supplied time coordinates are incompatible with the manifest. " + "The dataset may have changed or been reordered: " + f"{exc}" + ) from exc + if axis.n_samples != self.n_samples: + raise ValueError( + f"Manifest expects {self.n_samples} samples, received {axis.n_samples}." + ) + if axis.time_kind != self.time_kind: + raise ValueError( + f"Manifest expects time kind {self.time_kind!r}, " + f"received {axis.time_kind!r}." + ) + if axis.axis_mode != self.axis_mode: + raise ValueError( + f"Manifest expects axis mode {self.axis_mode!r}, " + f"received {axis.axis_mode!r}." + ) + if _fingerprint_time_axis(axis) != self.dataset_fingerprint: + raise ValueError( + "The supplied time coordinates do not match the manifest fingerprint. " + "The dataset may have changed or been reordered." + ) + try: + derived, _ = _derive_split_from_parameters( + axis, + method=self.method, + parameters=_thaw_json(self.parameters), + ) + except (TypeError, ValueError) as exc: + raise ValueError( + "Manifest state contradicts the supplied time axis or do not " + f"produce a valid split: {exc}" + ) from exc + names = ( + "train_indices", + "validation_indices", + "test_indices", + "excluded_indices", + ) + for name, values in zip(names, derived): + stored = tuple(getattr(self, name)) + recomputed = tuple(int(value) for value in values) + if stored != recomputed: + raise ValueError( + f"Manifest {name} contradicts its parameters and time axis." + ) + + +@dataclass(frozen=True) +class ChronologicalSplit: + """Immutable index partitions backed by a validated manifest.""" + + train_indices: np.ndarray + validation_indices: np.ndarray + test_indices: np.ndarray + excluded_indices: np.ndarray = field( + default_factory=lambda: np.empty(0, dtype=np.int64) + ) + manifest: ValidationSplitManifest | None = field( + default=None, + repr=False, + compare=False, + ) + + def __post_init__(self) -> None: + """Normalize split indices and verify partition consistency.""" + if self.manifest is None: + raise ValueError( + "ChronologicalSplit must be constructed from a validated manifest." + ) + names = ( + "train_indices", + "validation_indices", + "test_indices", + "excluded_indices", + ) + for name in names: + object.__setattr__( + self, + name, + _readonly_int_array( + getattr(self, name), + name=name, + n_samples=self.manifest.n_samples, + ), + ) + for name in ("train_indices", "validation_indices", "test_indices"): + if getattr(self, name).size == 0: + raise ValueError(f"{name} must not be empty.") + combined = np.concatenate( + [ + self.train_indices, + self.validation_indices, + self.test_indices, + self.excluded_indices, + ] + ) + if np.unique(combined).size != combined.size: + raise ValueError( + "Split partitions must not overlap or contain repeated indices." + ) + if set(combined.tolist()) != set(range(self.manifest.n_samples)): + raise ValueError("Split partitions must classify every manifest sample.") + if self.train_indices[-1] >= self.validation_indices[0]: + raise ValueError("Training indices must precede validation indices.") + if self.validation_indices[-1] >= self.test_indices[0]: + raise ValueError("Validation indices must precede test indices.") + expected = self.manifest.to_dict() + for name in names: + if getattr(self, name).tolist() != expected[name]: + raise ValueError(f"{name} does not match the attached manifest.") + + @property + def counts(self) -> Mapping[str, int]: + """Return immutable partition counts.""" + return MappingProxyType( + { + "train": int(self.train_indices.size), + "validation": int(self.validation_indices.size), + "test": int(self.test_indices.size), + "excluded": int(self.excluded_indices.size), + } + ) + + +def split_chronologically( + *, + sample_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None = None, + sample_start_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None = None, + sample_end_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None = None, + fractions: Sequence[RealScalar] | None = None, + boundary_indices: Sequence[int] | None = None, + boundary_times: Sequence[Any] | None = None, + gap: int = 0, +) -> ChronologicalSplit: + """Create a deterministic chronological split and reproducibility manifest. + + Exactly one split specification may be supplied. When none is supplied, the + fractions are ``(0.7, 0.15, 0.15)``. Fraction boundaries are resolved with + ``floor(n * train_fraction)`` and + ``floor(n * (train_fraction + validation_fraction))``; the test partition gets + the remaining samples. + + Point partitions are half-open: train uses ``time < b1``, validation uses + ``b1 <= time < b2``, and test uses ``time >= b2``. For interval samples, train + requires ``end < b1``, validation requires ``start >= b1`` and ``end < b2``, + and test requires ``start >= b2``. Therefore an interval ending exactly on a + boundary is excluded. ``gap`` removes the final ``gap`` samples from train and + validation after interval classification. + """ + axis = _prepare_time_axis( + sample_times=sample_times, + sample_start_times=sample_start_times, + sample_end_times=sample_end_times, + ) + gap_int = _validate_gap(gap) + specifications = sum( + specification is not None + for specification in (fractions, boundary_indices, boundary_times) + ) + if specifications > 1: + raise ValueError( + "Provide only one of fractions, boundary_indices, or boundary_times." + ) + if specifications == 0: + fractions = _DEFAULT_FRACTIONS + + parameters = _base_parameters(gap_int) + if fractions is not None: + validated_fractions, canonical_fractions = _validate_runtime_fractions( + fractions + ) + validation_index, test_index = _fraction_boundary_indices( + axis.n_samples, + validated_fractions, + ) + if ( + validation_index <= 0 + or test_index <= validation_index + or test_index >= axis.n_samples + ): + raise ValueError( + "Fractions do not produce non-empty candidate partitions for this " + "dataset." + ) + method = "fractions" + parameters.update( + { + "fractions": list(canonical_fractions), + "rounding_policy": _ROUNDING_POLICY, + "resolved_boundary_indices": [validation_index, test_index], + "resolved_boundary_values": [ + axis.canonical_starts[validation_index], + axis.canonical_starts[test_index], + ], + } + ) + elif boundary_indices is not None: + values = _require_sequence( + boundary_indices, + name="boundary_indices", + length=2, + ) + validation_index = _validate_index_boundary( + values[0], + name="validation_start_index", + n_samples=axis.n_samples, + ) + test_index = _validate_index_boundary( + values[1], + name="test_start_index", + n_samples=axis.n_samples, + ) + if validation_index >= test_index: + raise ValueError( + "validation_start_index must be less than test_start_index." + ) + method = "boundary_indices" + parameters.update( + { + "boundary_indices": [validation_index, test_index], + "resolved_boundary_values": [ + axis.canonical_starts[validation_index], + axis.canonical_starts[test_index], + ], + } + ) + else: + values = _require_sequence( + boundary_times, + name="boundary_times", + length=2, + ) + validation_boundary, validation_canonical = _normalize_boundary_scalar( + values[0], + time_kind=axis.time_kind, + name="validation_start_time", + ) + test_boundary, test_canonical = _normalize_boundary_scalar( + values[1], + time_kind=axis.time_kind, + name="test_start_time", + ) + if validation_boundary >= test_boundary: + raise ValueError("validation_start_time must precede test_start_time.") + method = "boundary_times" + parameters["boundary_times"] = [ + validation_canonical, + test_canonical, + ] + + partitions, validated_parameters = _derive_split_from_parameters( + axis, + method=method, + parameters=parameters, + ) + train, validation, test, excluded = partitions + manifest = ValidationSplitManifest( + method=method, + n_samples=axis.n_samples, + dataset_fingerprint=_fingerprint_time_axis(axis), + time_kind=axis.time_kind, + axis_mode=axis.axis_mode, + parameters=validated_parameters, + train_indices=tuple(int(value) for value in train), + validation_indices=tuple(int(value) for value in validation), + test_indices=tuple(int(value) for value in test), + excluded_indices=tuple(int(value) for value in excluded), + ) + return manifest.to_split() + + +def apply_split_manifest( + manifest: ValidationSplitManifest, + *, + sample_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None = None, + sample_start_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None = None, + sample_end_times: Sequence[Any] | np.ndarray | pd.Index | pd.Series | None = None, +) -> ChronologicalSplit: + """Validate current time coordinates and reproduce a saved split exactly.""" + if not isinstance(manifest, ValidationSplitManifest): + raise TypeError("manifest must be a ValidationSplitManifest instance.") + manifest.validate_against( + sample_times=sample_times, + sample_start_times=sample_start_times, + sample_end_times=sample_end_times, + ) + return manifest.to_split() diff --git a/docs/source/index.rst b/docs/source/index.rst index 5814490..7017bd9 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -14,6 +14,7 @@ BlueMath-tk: A Python Library for Coastal Climate Hazards installation contribute modules + validation Indices and tables ================== diff --git a/docs/source/validation.rst b/docs/source/validation.rst new file mode 100644 index 0000000..1b8954b --- /dev/null +++ b/docs/source/validation.rst @@ -0,0 +1,179 @@ +Reproducible chronological validation +===================================== + +Random train/validation/test splits can leak future information into climate and +other time-series experiments. BlueMath_tk therefore provides a chronological +splitter that returns reusable indices rather than copying large data arrays. + +Basic use +--------- + +.. code-block:: python + + import numpy as np + from bluemath_tk.validation import split_chronologically + + times = np.arange("2000-01", "2020-01", dtype="datetime64[M]") + split = split_chronologically( + sample_times=times, + fractions=(0.7, 0.15, 0.15), + ) + + X_train = X[split.train_indices] + X_validation = X[split.validation_indices] + X_test = X[split.test_indices] + +The returned index arrays are read-only and can index NumPy-compatible arrays +with any trailing dimensions, including ``(n_samples, T, C, H, W)`` data. + +Fraction rounding +----------------- + +Fraction splits use cumulative flooring. For ``n`` samples and fractions +``(f_train, f_validation, f_test)``, the validation and test start indices are: + +.. code-block:: text + + validation_start = floor(n * f_train) + test_start = floor(n * (f_train + f_validation)) + +The test partition receives the remainder. Fractions must be positive finite +real scalars, must sum to one within numerical tolerance, and must leave all +three candidate partitions non-empty. Boundary resolution preserves an exact +canonical rational for every scalar. ``fractions.Fraction`` values retain their +exact numerator and denominator. Python and NumPy floating scalars use their +round-trip decimal spelling before conversion to a rational. Homogeneous NumPy +arrays retain their original scalar dtype during sequence validation, so common +values such as ``np.array([0.7, 0.1, 0.2], dtype=np.float32)`` resolve to +``(7/10, 1/10, 1/5)`` without an intermediate Python ``float`` conversion. The +manifest stores reduced ``numerator/denominator`` strings and therefore replays +with the same arithmetic after JSON round trips. + +Explicit boundaries +------------------- + +Use either two start indices or two time values. The first boundary starts the +validation partition and the second starts the test partition. + +.. code-block:: python + + split = split_chronologically( + sample_times=times, + boundary_times=(np.datetime64("2014-01"), np.datetime64("2017-01")), + ) + +Only one split specification may be supplied: fractions, boundary indices, or +boundary times. + +Half-open boundary rules +------------------------ + +For point samples with boundaries ``b1`` and ``b2``: + +* train: ``time < b1``; +* validation: ``b1 <= time < b2``; +* test: ``time >= b2``. + +A point exactly on a boundary therefore joins the later partition. + +For interval samples: + +* train: ``end < b1``; +* validation: ``start >= b1`` and ``end < b2``; +* test: ``start >= b2``. + +An interval ending exactly on a boundary is not assigned to the preceding +partition. If it started before that boundary, it crosses the boundary and is +recorded in ``excluded_indices``. + +Windowed samples and leakage prevention +--------------------------------------- + +For samples representing complete source intervals, supply both start and end +coordinates: + +.. code-block:: python + + split = split_chronologically( + sample_start_times=window_starts, + sample_end_times=window_ends, + boundary_times=(validation_start, test_start), + gap=2, + ) + +A window is assigned only when its complete interval lies within one partition. +A window crossing a boundary is recorded in ``excluded_indices`` and is never +silently assigned by its target or final time. + +``gap`` is a non-negative sample count. After interval classification, it +removes the final ``gap`` samples from the training partition and the final +``gap`` samples from the validation partition. Those samples are added to +``excluded_indices`` immediately before the later partitions. + +Datetime and numeric coordinates +-------------------------------- + +Timezone-aware datetime values are converted explicitly to UTC and represented +at checked nanosecond resolution. Timezone awareness remains part of the +manifest and fingerprint, so an aware axis cannot be replayed as a naive axis. +Values outside the supported nanosecond range, or values that cannot round-trip +exactly at that resolution, are rejected rather than wrapped or truncated. +NumPy ``datetime64`` arrays retain their dtype unit during normalization. +Sub-nanosecond units such as picoseconds and femtoseconds are accepted only +when every value is an exact nanosecond multiple; otherwise the split is +rejected. Calendar units, including stepped year and month dtypes, are checked +with arbitrary-precision offsets before NumPy renders a date, so extreme counts +cannot wrap into ordinary in-range dates. Equivalent exact +picosecond/nanosecond axes therefore share a fingerprint, while scaled, +truncated, or wrapped axes do not. + +Timezone-naive datetime values remain distinct from timezone-aware values. +Interval starts, ends, and explicit time boundaries must use the same awareness +state. Unsigned integer coordinates are supported only within the signed +64-bit range and remain distinguishable from signed integer coordinates in the +manifest fingerprint. + +Reproducibility manifests +------------------------- + +Every split includes a deterministic manifest: + +.. code-block:: python + + split.manifest.save("split_manifest.json") + + from bluemath_tk.validation import ( + ValidationSplitManifest, + apply_split_manifest, + ) + + manifest = ValidationSplitManifest.load("split_manifest.json") + reproduced = apply_split_manifest(manifest, sample_times=times) + +The manifest stores the method, strict method-specific parameters, resolved +boundaries, gap, partition indices, excluded indices, point-or-interval mode, +time-coordinate kind, sample count, schema version, and a SHA-256 fingerprint +of the ordered sample coordinates. + +Manifest JSON is strict and deterministic. NaN, infinity, non-JSON values, +unknown parameters, missing parameters, coercible non-integer indices, and +unsupported schema versions are rejected. During replay, BlueMath_tk recomputes +the split from the stored parameters and supplied coordinates and compares all +partitions. Changed, reordered, or contradictory data therefore fail clearly. +SHA-256 digests are stored as exactly 64 lowercase hexadecimal characters. + +Preprocessing +------------- + +The splitter returns indices only. Fit scalers, PCA, or other preprocessing +objects using ``train_indices`` and apply those fitted objects to validation and +test data. This module does not automatically fit preprocessing and therefore +does not introduce validation or test information into training. + +Current limitations +------------------- + +This first API implements one deterministic chronological split. Rolling-origin, +expanding-window, event-based, and grouped station/site validation are not yet +included. The gap is measured in samples rather than elapsed time. Time inputs +must be strictly increasing, and duplicate sample start times are rejected. diff --git a/tests/validation/test_chronological.py b/tests/validation/test_chronological.py new file mode 100644 index 0000000..d92c045 --- /dev/null +++ b/tests/validation/test_chronological.py @@ -0,0 +1,817 @@ +"""Tests for reproducible chronological validation splits.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from fractions import Fraction +from typing import TYPE_CHECKING + +import numpy as np +import pandas as pd +import pytest + +from bluemath_tk.validation import ( + ChronologicalSplit, + RealScalar, + ValidationSplitManifest, + apply_split_manifest, + split_chronologically, +) + + +def test_default_fraction_split_is_deterministic_and_ordered(): + times = np.arange(20) + first = split_chronologically(sample_times=times) + second = split_chronologically(sample_times=times) + + assert first.train_indices.tolist() == list(range(14)) + assert first.validation_indices.tolist() == [14, 15, 16] + assert first.test_indices.tolist() == [17, 18, 19] + assert first.excluded_indices.size == 0 + assert first.manifest.to_json() == second.manifest.to_json() + assert first.counts == {"train": 14, "validation": 3, "test": 3, "excluded": 0} + + +def test_fraction_split_uses_cumulative_floor_and_test_remainder(): + split = split_chronologically( + sample_times=np.arange(11), + fractions=(0.6, 0.2, 0.2), + ) + assert split.train_indices.tolist() == list(range(6)) + assert split.validation_indices.tolist() == [6, 7] + assert split.test_indices.tolist() == [8, 9, 10] + + +def test_explicit_index_boundaries(): + split = split_chronologically( + sample_times=np.arange(10), + boundary_indices=(5, 8), + ) + assert split.train_indices.tolist() == [0, 1, 2, 3, 4] + assert split.validation_indices.tolist() == [5, 6, 7] + assert split.test_indices.tolist() == [8, 9] + assert split.manifest.method == "boundary_indices" + + +def test_explicit_numeric_time_boundaries(): + split = split_chronologically( + sample_times=np.arange(0.0, 10.0), + boundary_times=(5.0, 8.0), + ) + assert split.train_indices.tolist() == [0, 1, 2, 3, 4] + assert split.validation_indices.tolist() == [5, 6, 7] + assert split.test_indices.tolist() == [8, 9] + assert split.manifest.method == "boundary_times" + + +def test_datetime64_boundaries_and_manifest_round_trip(tmp_path): + times = np.arange("2020-01-01", "2020-01-11", dtype="datetime64[D]") + split = split_chronologically( + sample_times=times, + boundary_times=(np.datetime64("2020-01-06"), np.datetime64("2020-01-09")), + ) + path = tmp_path / "split.json" + split.manifest.save(path) + loaded = ValidationSplitManifest.load(path) + replay = apply_split_manifest(loaded, sample_times=times) + + assert loaded.to_json() == split.manifest.to_json() + assert replay.train_indices.tolist() == [0, 1, 2, 3, 4] + assert replay.validation_indices.tolist() == [5, 6, 7] + assert replay.test_indices.tolist() == [8, 9] + + +def test_timezone_aware_times_are_normalized_to_utc(): + times = pd.date_range("2020-01-01", periods=10, freq="h", tz="Europe/London") + split = split_chronologically( + sample_times=times, + boundary_times=(times[5], times[8]), + ) + equivalent = times.tz_convert("UTC") + replay = apply_split_manifest(split.manifest, sample_times=equivalent) + assert replay.train_indices.tolist() == list(range(5)) + + +def test_mixed_timezone_awareness_is_rejected(): + values = [ + datetime(2020, 1, 1), + datetime(2020, 1, 2, tzinfo=UTC), + datetime(2020, 1, 3, tzinfo=UTC), + ] + with pytest.raises(ValueError, match="mixes timezone-aware"): + split_chronologically(sample_times=values) + + +def test_complete_interval_policy_excludes_boundary_crossing_windows(): + starts = np.arange(10) + ends = starts + 2 + split = split_chronologically( + sample_start_times=starts, + sample_end_times=ends, + boundary_indices=(5, 8), + ) + assert split.train_indices.tolist() == [0, 1, 2] + assert split.validation_indices.tolist() == [5] + assert split.test_indices.tolist() == [8, 9] + assert split.excluded_indices.tolist() == [3, 4, 6, 7] + + +def test_interval_touching_boundary_is_excluded_from_previous_partition(): + starts = np.array([0, 1, 2, 3, 4, 5]) + ends = np.array([0, 1, 3, 3, 4, 5]) + split = split_chronologically( + sample_start_times=starts, + sample_end_times=ends, + boundary_indices=(3, 5), + ) + assert 2 in split.excluded_indices + assert 2 not in split.train_indices + + +def test_gap_removes_samples_immediately_before_later_partitions(): + split = split_chronologically( + sample_times=np.arange(12), + boundary_indices=(6, 9), + gap=1, + ) + assert split.train_indices.tolist() == [0, 1, 2, 3, 4] + assert split.validation_indices.tolist() == [6, 7] + assert split.test_indices.tolist() == [9, 10, 11] + assert split.excluded_indices.tolist() == [5, 8] + + +def test_indices_are_read_only_and_directly_index_multidimensional_data(): + times = np.arange(10) + data = np.arange(10 * 2 * 3).reshape(10, 2, 3) + split = split_chronologically(sample_times=times, boundary_indices=(5, 8)) + assert np.array_equal(data[split.train_indices], data[:5]) + assert not split.train_indices.flags.writeable + with pytest.raises(ValueError): + split.train_indices[0] = 99 + + +@pytest.mark.parametrize( + "times,match", + [ + (np.array([0, 2, 1, 3]), "strictly increasing"), + (np.array([0, 1, 1, 2]), "strictly increasing"), + (np.array([0.0, np.nan, 2.0]), "NaN or infinite"), + (np.array([0.0, np.inf, 2.0]), "NaN or infinite"), + (np.array([True, False, True]), "Boolean"), + (np.array(["a", "b", "c"]), "real numeric values or datetime-like"), + ], +) +def test_invalid_time_coordinates_are_rejected(times, match): + with pytest.raises((TypeError, ValueError), match=match): + split_chronologically(sample_times=times) + + +def test_nat_is_rejected(): + times = np.array(["2020-01-01", "NaT", "2020-01-03"], dtype="datetime64[D]") + with pytest.raises(ValueError, match="NaT"): + split_chronologically(sample_times=times) + + +def test_interval_validation_rejects_missing_mismatched_or_reverse_inputs(): + with pytest.raises(ValueError, match="provided together"): + split_chronologically(sample_start_times=np.arange(4)) + with pytest.raises(ValueError, match="same length"): + split_chronologically( + sample_start_times=np.arange(4), + sample_end_times=np.arange(3), + ) + with pytest.raises(ValueError, match="end_time >= start_time"): + split_chronologically( + sample_start_times=np.arange(4), + sample_end_times=np.array([0, 0, 2, 3]), + ) + + +def test_point_and_interval_forms_are_mutually_exclusive(): + with pytest.raises(ValueError, match="not both forms"): + split_chronologically( + sample_times=np.arange(5), + sample_start_times=np.arange(5), + sample_end_times=np.arange(5), + ) + + +@pytest.mark.parametrize( + "fractions,error_type,match", + [ + ((0.7, 0.3), ValueError, "exactly"), + ((0.7, 0.2, 0.2), ValueError, "sum"), + ((0.0, 0.5, 0.5), ValueError, "strictly"), + ((True, 0.4, 0.6), TypeError, "Boolean"), + ((np.nan, 0.5, 0.5), ValueError, "finite"), + ], +) +def test_invalid_fractions_are_rejected(fractions, error_type, match): + with pytest.raises(error_type, match=match): + split_chronologically(sample_times=np.arange(10), fractions=fractions) + + +def test_multiple_split_specifications_are_rejected(): + with pytest.raises(ValueError, match="only one"): + split_chronologically( + sample_times=np.arange(10), + fractions=(0.6, 0.2, 0.2), + boundary_indices=(6, 8), + ) + + +@pytest.mark.parametrize("gap", [-1, True, 1.5, "1"]) +def test_invalid_gap_is_rejected(gap): + with pytest.raises((TypeError, ValueError), match="gap"): + split_chronologically(sample_times=np.arange(10), gap=gap) + + +def test_gap_cannot_empty_a_partition(): + with pytest.raises(ValueError, match="removes every validation"): + split_chronologically( + sample_times=np.arange(8), + boundary_indices=(4, 6), + gap=2, + ) + + +def test_manifest_rejects_reordered_or_changed_data(): + times = np.arange(10) + split = split_chronologically(sample_times=times, boundary_indices=(5, 8)) + reordered = times.copy() + reordered[[0, 1]] = reordered[[1, 0]] + with pytest.raises(ValueError, match="changed or been reordered"): + apply_split_manifest(split.manifest, sample_times=reordered) + changed = times.copy() + changed[-1] = 99 + with pytest.raises(ValueError, match="changed or been reordered"): + apply_split_manifest(split.manifest, sample_times=changed) + + +def test_manifest_rejects_different_time_kind_and_sample_count(): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + with pytest.raises(ValueError, match="samples"): + apply_split_manifest(split.manifest, sample_times=np.arange(11)) + with pytest.raises(ValueError, match="time kind"): + apply_split_manifest(split.manifest, sample_times=np.arange(10, dtype=float)) + + +def test_manifest_json_is_stable_and_has_newline(): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + first = split.manifest.to_json() + second = ValidationSplitManifest.from_dict(split.manifest.to_dict()).to_json() + assert first == second + assert first.endswith("\n") + assert '"schema_version": 2' in first + assert '"axis_mode": "point"' in first + + +def test_manifest_rejects_unknown_or_missing_fields(): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload["unknown"] = 1 + with pytest.raises(ValueError, match="unsupported fields"): + ValidationSplitManifest.from_dict(payload) + payload = split.manifest.to_dict() + del payload["method"] + with pytest.raises(ValueError, match="missing required fields"): + ValidationSplitManifest.from_dict(payload) + + +def test_split_does_not_mutate_inputs_or_global_numpy_rng(): + times = np.arange(10) + original = times.copy() + np.random.seed(1234) + before = np.random.get_state() + split_chronologically(sample_times=times, boundary_indices=(5, 8)) + after = np.random.get_state() + assert np.array_equal(times, original) + assert before[0] == after[0] + assert np.array_equal(before[1], after[1]) + assert before[2:] == after[2:] + + +def test_boundary_time_kind_must_match_axis_kind(): + with pytest.raises(TypeError, match="integer value"): + split_chronologically( + sample_times=np.arange(10), + boundary_times=(5.0, 8.0), + ) + + +def test_boundary_order_and_bounds_must_leave_nonempty_partitions(): + with pytest.raises(ValueError, match="must precede"): + split_chronologically( + sample_times=np.arange(10), + boundary_times=(8, 5), + ) + with pytest.raises(ValueError, match="non-empty"): + split_chronologically( + sample_times=np.arange(10), + boundary_times=(-1, 8), + ) + + +def test_explicit_boundaries_are_strictly_validated(): + with pytest.raises(TypeError, match="integer index"): + split_chronologically(sample_times=np.arange(10), boundary_indices=(True, 8)) + with pytest.raises(ValueError, match="less than"): + split_chronologically(sample_times=np.arange(10), boundary_indices=(8, 5)) + with pytest.raises(ValueError, match="strictly inside"): + split_chronologically(sample_times=np.arange(10), boundary_indices=(0, 8)) + + +def test_manifest_parameters_are_immutable(): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + with pytest.raises(TypeError): + split.manifest.parameters["gap_samples_before_later_partition"] = 2 + + +def test_timezone_resolution_is_canonical_and_awareness_is_bound(): + aware_us = pd.date_range( + "2024-01-01", + periods=10, + freq="h", + tz="UTC", + ).as_unit("us") + aware_ns = aware_us.as_unit("ns") + split = split_chronologically( + sample_times=aware_us, + boundary_times=(aware_us[5], aware_us[8]), + ) + replay = apply_split_manifest(split.manifest, sample_times=aware_ns) + assert replay.train_indices.tolist() == list(range(5)) + assert split.manifest.time_kind == "datetime64[ns]-aware-utc" + + naive = aware_ns.tz_localize(None) + with pytest.raises(ValueError, match="time kind"): + apply_split_manifest(split.manifest, sample_times=naive) + + +def test_datetime_fingerprint_rejects_1970_collision_candidate(): + aware = pd.date_range("2024-01-01", periods=10, freq="h", tz="UTC").as_unit("us") + split = split_chronologically(sample_times=aware, boundary_indices=(5, 8)) + collision_candidate = pd.to_datetime(aware.asi8, unit="ns", utc=True) + assert collision_candidate[0].year == 1970 + with pytest.raises(ValueError, match="fingerprint"): + apply_split_manifest(split.manifest, sample_times=collision_candidate) + + +def test_datetime_outside_nanosecond_range_is_rejected_without_wraparound(): + times = np.arange( + np.datetime64("2263-01-01", "us"), + np.datetime64("2263-01-05", "us"), + np.timedelta64(1, "D"), + ) + with pytest.raises(ValueError, match="nanosecond"): + split_chronologically(sample_times=times) + + +def test_mixed_aware_interval_endpoints_and_boundaries_are_rejected(): + starts = pd.date_range("2024-01-01", periods=10, freq="h", tz="UTC") + naive_ends = starts.tz_localize(None) + pd.Timedelta(minutes=30) + with pytest.raises(TypeError, match="timezone-awareness"): + split_chronologically( + sample_start_times=starts, + sample_end_times=naive_ends, + boundary_indices=(5, 8), + ) + with pytest.raises(TypeError, match="expected.*aware"): + split_chronologically( + sample_times=starts, + boundary_times=( + starts[5].tz_localize(None), + starts[8].tz_localize(None), + ), + ) + + +def test_point_and_interval_modes_are_bound_to_manifest(): + times = np.arange(10) + split = split_chronologically(sample_times=times, boundary_indices=(5, 8)) + with pytest.raises(ValueError, match="axis mode"): + apply_split_manifest( + split.manifest, + sample_start_times=times, + sample_end_times=times, + ) + + +def test_unsigned_values_above_int64_range_are_rejected_without_wraparound(): + values = np.array([2**63, 2**63 + 1, 2**63 + 2, 2**63 + 3], dtype=np.uint64) + with pytest.raises(ValueError, match="unsigned integers"): + split_chronologically(sample_times=values) + + +def test_unsigned_and_signed_coordinate_kinds_do_not_collide(): + unsigned = np.arange(10, dtype=np.uint64) + split = split_chronologically(sample_times=unsigned, boundary_indices=(5, 8)) + with pytest.raises(ValueError, match="time kind"): + apply_split_manifest(split.manifest, sample_times=unsigned.astype(np.int64)) + + +@pytest.mark.parametrize( + "field,value,error", + [ + ("schema_version", True, TypeError), + ("schema_version", 1.0, TypeError), + ("n_samples", True, TypeError), + ("n_samples", 10.0, TypeError), + ], +) +def test_manifest_rejects_coercible_scalar_types(field, value, error): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload[field] = value + with pytest.raises(error): + ValidationSplitManifest.from_dict(payload) + + +@pytest.mark.parametrize("value", [True, 0.9, "0", np.float64(0.0)]) +def test_manifest_rejects_coercible_index_values(value): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload["train_indices"][0] = value + with pytest.raises(TypeError, match="integer|non-JSON"): + ValidationSplitManifest.from_dict(payload) + + +@pytest.mark.parametrize( + "replacement", + [ + {}, + {"boundary_indices": [5, 8]}, + { + "boundary_policy": "complete_interval_half_open", + "partition_closure": ( + "train:end=b1,end=b2" + ), + "gap_samples_before_later_partition": 0, + "boundary_indices": [5, 8], + "resolved_boundary_values": ["5", "8"], + "extra": 1, + }, + ], +) +def test_manifest_rejects_missing_or_extra_parameter_schema(replacement): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload["parameters"] = replacement + with pytest.raises(ValueError, match="parameters"): + ValidationSplitManifest.from_dict(payload) + + +@pytest.mark.parametrize("bad_value", [{1, 2}, np.int64(1), np.nan, np.inf]) +def test_manifest_parameters_reject_non_json_or_nonfinite_values(bad_value): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload["parameters"]["gap_samples_before_later_partition"] = bad_value + with pytest.raises((TypeError, ValueError)): + ValidationSplitManifest.from_dict(payload) + + +def test_manifest_fraction_parameters_are_cross_validated(): + split = split_chronologically( + sample_times=np.arange(20), + fractions=(0.7, 0.15, 0.15), + ) + for field, value in ( + ("fractions", ["3/5", "1/5", "1/5"]), + ("resolved_boundary_indices", [13, 17]), + ("gap_samples_before_later_partition", 9), + ): + payload = split.manifest.to_dict() + payload["parameters"][field] = value + if field == "gap_samples_before_later_partition": + altered = ValidationSplitManifest.from_dict(payload) + with pytest.raises(ValueError, match="contradicts"): + apply_split_manifest(altered, sample_times=np.arange(20)) + else: + with pytest.raises(ValueError, match="contradict"): + ValidationSplitManifest.from_dict(payload) + + +def test_manifest_replay_rejects_shifted_but_structurally_valid_partitions(): + times = np.arange(20) + split = split_chronologically(sample_times=times) + payload = split.manifest.to_dict() + payload["train_indices"] = list(range(13)) + payload["validation_indices"] = list(range(13, 17)) + payload["excluded_indices"] = [13] + payload["validation_indices"] = list(range(14, 17)) + altered = ValidationSplitManifest.from_dict(payload) + with pytest.raises(ValueError, match="contradicts"): + apply_split_manifest(altered, sample_times=times) + + +def test_strict_json_loader_rejects_nan_and_infinity(tmp_path): + path = tmp_path / "invalid.json" + path.write_text('{"schema_version": NaN}', encoding="utf-8") + with pytest.raises(ValueError, match="Non-standard JSON constant"): + ValidationSplitManifest.load(path) + + +def test_direct_chronological_split_requires_a_validated_manifest(): + with pytest.raises(ValueError, match="validated manifest"): + ChronologicalSplit([0], [1], [2]) + + +@pytest.mark.parametrize( + "train,validation,test", + [ + ([1, 0], [2], [3]), + ([-2, -1], [0], [1]), + ([0.9], [1.9], [2.9]), + (["0"], [1], [2]), + ], +) +def test_direct_split_invalid_indices_cannot_bypass_manifest(train, validation, test): + with pytest.raises(ValueError, match="validated manifest"): + ChronologicalSplit(train, validation, test) + + +@pytest.mark.parametrize( + "fractions", + [ + (Fraction(7, 10), Fraction(1, 10), Fraction(1, 5)), + (Fraction(7, 10), np.float32(0.1), Fraction(1, 5)), + ], +) +def test_fraction_accepts_fraction_and_numpy_real_scalars(fractions): + split = split_chronologically( + sample_times=np.arange(10), + fractions=fractions, + ) + assert split.train_indices.tolist() == list(range(7)) + assert split.validation_indices.tolist() == [7] + assert split.test_indices.tolist() == [8, 9] + + +@pytest.mark.parametrize( + "keyword,value", + [ + ("fractions", 0.5), + ("fractions", np.array(0.5)), + ("boundary_indices", np.array(5)), + ("boundary_times", np.datetime64("2020-01-01")), + ], +) +def test_scalar_split_specifications_raise_argument_specific_errors(keyword, value): + kwargs = {keyword: value} + with pytest.raises(TypeError, match=keyword): + split_chronologically(sample_times=np.arange(10), **kwargs) + + +def test_fraction_rounding_and_half_open_touching_rules_are_explicit(): + split = split_chronologically( + sample_times=np.arange(11), + fractions=(0.6, 0.2, 0.2), + ) + assert split.manifest.parameters["rounding_policy"] == "cumulative_floor" + assert split.manifest.parameters["resolved_boundary_indices"] == (6, 8) + assert split.validation_indices[0] == 6 + assert split.test_indices[0] == 8 + + starts = np.array([0, 1, 2, 3, 4, 5, 6]) + ends = np.array([0, 1, 3, 3, 4, 5, 6]) + interval = split_chronologically( + sample_start_times=starts, + sample_end_times=ends, + boundary_times=(3, 5), + ) + assert 2 in interval.excluded_indices + assert 3 in interval.validation_indices + assert 5 in interval.test_indices + + +def test_manifest_axis_mode_and_timezone_kind_are_serialized(): + times = pd.date_range("2024-01-01", periods=10, freq="h", tz="UTC") + split = split_chronologically(sample_times=times, boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + assert payload["axis_mode"] == "point" + assert payload["time_kind"] == "datetime64[ns]-aware-utc" + + +def test_manifest_replay_recomputes_gap_partitions(): + times = np.arange(12) + split = split_chronologically( + sample_times=times, + boundary_indices=(6, 9), + gap=1, + ) + payload = split.manifest.to_dict() + payload["parameters"]["gap_samples_before_later_partition"] = 0 + altered = ValidationSplitManifest.from_dict(payload) + with pytest.raises(ValueError, match="contradicts"): + apply_split_manifest(altered, sample_times=times) + + +if TYPE_CHECKING: + + def _fraction_typing_examples(values: tuple[RealScalar, RealScalar, RealScalar]): + split_chronologically(sample_times=np.arange(10), fractions=values) + split_chronologically( + sample_times=np.arange(3), + fractions=(Fraction(1, 3), np.float32(1 / 3), 1 / 3), + ) + + +@pytest.mark.parametrize( + "n_samples,expected_validation,expected_test", + [(3, 1, 2), (6, 2, 4)], +) +def test_exact_rational_thirds_use_exact_cumulative_floor( + n_samples, + expected_validation, + expected_test, +): + split = split_chronologically( + sample_times=np.arange(n_samples), + fractions=(Fraction(1, 3), Fraction(1, 3), Fraction(1, 3)), + ) + assert split.validation_indices[0] == expected_validation + assert split.test_indices[0] == expected_test + assert split.manifest.parameters["fractions"] == ("1/3", "1/3", "1/3") + + +def test_high_denominator_fraction_near_floor_boundaries_is_exact(): + denominator = 1_000_000_007 + fractions = ( + Fraction(333_333_336, denominator), + Fraction(333_333_336, denominator), + Fraction(333_333_335, denominator), + ) + split = split_chronologically(sample_times=np.arange(3), fractions=fractions) + assert split.train_indices.tolist() == [0] + assert split.validation_indices.tolist() == [1] + assert split.test_indices.tolist() == [2] + + +def test_numpy_float32_fractions_preserve_decimal_intent(): + split = split_chronologically( + sample_times=np.arange(10), + fractions=(np.float32(0.7), np.float32(0.1), np.float32(0.2)), + ) + assert split.manifest.parameters["fractions"] == ("7/10", "1/10", "1/5") + assert split.manifest.parameters["resolved_boundary_indices"] == (7, 8) + + +def test_numpy_float32_fraction_array_preserves_intent_and_replays(): + fractions = np.array([0.7, 0.1, 0.2], dtype=np.float32) + times = np.arange(10) + split = split_chronologically(sample_times=times, fractions=fractions) + assert split.manifest.parameters["fractions"] == ("7/10", "1/10", "1/5") + assert split.manifest.parameters["resolved_boundary_indices"] == (7, 8) + + loaded = ValidationSplitManifest.from_dict(split.manifest.to_dict()) + replay = apply_split_manifest(loaded, sample_times=times) + assert replay.train_indices.tolist() == list(range(7)) + assert replay.validation_indices.tolist() == [7] + assert replay.test_indices.tolist() == [8, 9] + + +def test_exact_fraction_manifest_round_trip_replays_identically(tmp_path): + times = np.arange(6) + split = split_chronologically( + sample_times=times, + fractions=(Fraction(1, 3), Fraction(1, 3), Fraction(1, 3)), + ) + path = tmp_path / "fraction-split.json" + split.manifest.save(path) + loaded = ValidationSplitManifest.load(path) + replay = apply_split_manifest(loaded, sample_times=times) + assert loaded.parameters["fractions"] == ("1/3", "1/3", "1/3") + assert replay.validation_indices.tolist() == [2, 3] + assert replay.test_indices.tolist() == [4, 5] + + +def test_numpy_picosecond_axis_replays_equivalent_nanoseconds_only(): + picoseconds = np.array([0, 1000, 2000, 3000, 4000, 5000], dtype="datetime64[ps]") + split = split_chronologically( + sample_times=picoseconds, + boundary_indices=(2, 4), + ) + equivalent_ns = picoseconds.astype("datetime64[ns]") + replay = apply_split_manifest(split.manifest, sample_times=equivalent_ns) + assert replay.validation_indices.tolist() == [2, 3] + + scaled_collision = picoseconds.astype(np.int64).astype("datetime64[ns]") + with pytest.raises(ValueError, match="fingerprint"): + apply_split_manifest(split.manifest, sample_times=scaled_collision) + + +def test_numpy_subnanosecond_values_must_be_exact_nanosecond_multiples(): + picoseconds = np.array([0, 1001, 2000], dtype="datetime64[ps]") + with pytest.raises(ValueError, match="exact nanosecond multiple"): + split_chronologically(sample_times=picoseconds) + + +def test_numpy_femtosecond_exact_multiples_normalize_to_nanoseconds(): + femtoseconds = np.array( + [0, 1_000_000, 2_000_000, 3_000_000], + dtype="datetime64[fs]", + ) + split = split_chronologically( + sample_times=femtoseconds, + boundary_indices=(1, 3), + ) + equivalent_ns = femtoseconds.astype("datetime64[ns]") + replay = apply_split_manifest(split.manifest, sample_times=equivalent_ns) + assert replay.test_indices.tolist() == [3] + + +@pytest.mark.parametrize( + "fingerprint", + [ + "+" + "0" * 63, + " " + "0" * 63, + "g" + "0" * 63, + "A" * 64, + "0" * 63, + ], +) +def test_manifest_fingerprint_requires_lowercase_sha256_hex(fingerprint): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload["dataset_fingerprint"] = fingerprint + with pytest.raises(ValueError, match="lowercase 64-character"): + ValidationSplitManifest.from_dict(payload) + + +@pytest.mark.parametrize( + "field,value", + [ + ("axis_mode", np.str_("point")), + ("time_kind", np.str_("integer-signed")), + ("method", np.str_("boundary_indices")), + ], +) +def test_manifest_rejects_numpy_string_top_level_metadata(field, value): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload[field] = value + with pytest.raises(TypeError, match="non-JSON|built-in string"): + ValidationSplitManifest.from_dict(payload) + + +@pytest.mark.parametrize("value", [np.int64(0), np.uint64(0)]) +def test_manifest_rejects_numpy_integral_indices(value): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload["train_indices"][0] = value + with pytest.raises(TypeError, match="non-JSON|built-in integer"): + ValidationSplitManifest.from_dict(payload) + + +@pytest.mark.parametrize("value", [[], {}, ["point"], {"point": True}]) +def test_manifest_axis_mode_container_types_raise_clear_type_error(value): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload["axis_mode"] = value + with pytest.raises(TypeError, match="non-JSON|built-in string"): + ValidationSplitManifest.from_dict(payload) + + +@pytest.mark.parametrize("dtype", ["datetime64[M]", "datetime64[Y]", "datetime64[2M]"]) +def test_numpy_calendar_datetime_units_normalize_without_tolist_loss(dtype): + values = np.arange(6, dtype=np.int64).astype(dtype) + split = split_chronologically(sample_times=values, boundary_indices=(2, 4)) + equivalent_ns = values.astype("datetime64[ns]") + replay = apply_split_manifest(split.manifest, sample_times=equivalent_ns) + assert replay.validation_indices.tolist() == [2, 3] + + +@pytest.mark.parametrize("dtype", ["datetime64[2M]", "datetime64[2Y]"]) +@pytest.mark.parametrize("edge", ["high", "low"]) +def test_extreme_stepped_calendar_counts_reject_before_numpy_rendering( + dtype, + edge, +): + limits = np.iinfo(np.int64) + if edge == "high": + raw = np.array([limits.max - 2, limits.max - 1, limits.max]) + else: + raw = np.array([limits.min + 1, limits.min + 2, limits.min + 3]) + values = raw.astype(dtype) + with pytest.raises(ValueError, match="supported nanosecond range"): + split_chronologically(sample_times=values, boundary_indices=(1, 2)) + + +@pytest.mark.parametrize( + "field,value,match", + [ + ("method", [], "built-in string"), + ("time_kind", {}, "built-in string"), + ("dataset_fingerprint", [], "built-in string"), + ], +) +def test_manifest_top_level_container_scalars_raise_clear_type_errors( + field, + value, + match, +): + split = split_chronologically(sample_times=np.arange(10), boundary_indices=(5, 8)) + payload = split.manifest.to_dict() + payload[field] = value + with pytest.raises(TypeError, match=match): + ValidationSplitManifest.from_dict(payload)