check_continuity¶

def check_continuity(
    df_p: pl.DataFrame,
    df_n: pl.DataFrame,
    expected_interval: str | None,
    check_intervals: bool = True,
) -> None:
    """Validate temporal continuity between consecutive DataFrames.

    Ensures that two DataFrames representing consecutive time periods have no gaps
    or overlaps in their time indices. Used when appending new data to existing
    time series.

    Parameters
    ----------
    df_p : pl.DataFrame
        Previous (earlier) time series DataFrame with "time" column.

    df_n : pl.DataFrame
        Next (later) time series DataFrame with "time" column.

    expected_interval : str | None
        Expected time interval between consecutive observations.
        Examples: "1d", "1h", "1mo", "3mo", "1y"
        If None, skip interval validation (used for single-step predictions).

    check_intervals : bool, default=True
        If True, validates that both DataFrames have consistent internal intervals.
        Single-row DataFrames cannot have an internal interval and are skipped by
        this check (relevant for single-step observation updates).

    Raises
    ------
    ValueError
        If there is a gap or overlap between df_p and df_n, or if internal
        intervals don't match expected_interval.
    ValueError
        If df_p and df_n each have internally consistent but mutually
        incompatible intervals (only checked when both have at least 2 rows).

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime, timedelta
    >>> # Continuous time series
    >>> df1 = pl.DataFrame({
    ...     "time": pl.datetime_range(datetime(2020, 1, 1), datetime(2020, 1, 3), "1d", eager=True),
    ...     "value": [10, 20, 30],
    ... })
    >>> df2 = pl.DataFrame({
    ...     "time": pl.datetime_range(datetime(2020, 1, 4), datetime(2020, 1, 6), "1d", eager=True),
    ...     "value": [40, 50, 60],
    ... })
    >>> check_continuity(df1, df2, "1d")

    See Also
    --------
    - [`check_interval_consistency`][yohou.utils.validation.check_interval_consistency] : Validates uniform time spacing

    """
    # Skip validation if expected_interval is None (e.g., single-step prediction)
    if expected_interval is None:
        return

    if check_intervals:
        if len(df_p) > 1:
            interval_p = check_interval_consistency(df_p)
            # Normalize intervals for comparison (e.g., "31d" -> "1mo")
            interval_p_norm = _normalize_interval(interval_p)
            expected_interval_norm = _normalize_interval(expected_interval)

            if interval_p_norm != expected_interval_norm:
                raise ValueError(
                    f"Previous DataFrame has interval {interval_p}, but expected interval is {expected_interval}."
                )

        if len(df_n) > 1:
            interval_n = check_interval_consistency(df_n)
            interval_n_norm = _normalize_interval(interval_n)

            if len(df_p) > 1 and interval_p_norm != interval_n_norm:
                raise ValueError(
                    "Interval mismatch between DataFrames: previous DataFrame has interval "
                    f"{interval_p}, but next DataFrame has interval {interval_n}."
                )

            expected_interval_norm = _normalize_interval(expected_interval)
            if interval_n_norm != expected_interval_norm:
                raise ValueError(
                    f"Next DataFrame has interval {interval_n}, but expected interval is {expected_interval}."
                )

    time_p = df_p.select(cs.by_name("time")).tail(1)
    time_n = df_n.select(cs.by_name("time")).head(1)

    time = pl.concat([time_p, time_n])

    time_change = time.select(pl.col("time").diff()).fill_null(strategy="backward")
    interval_td = time_change[0, 0]

    # Convert expected_interval string to timedelta for comparison
    expected_interval_td = interval_to_timedelta(expected_interval)

    # For variable-length intervals (monthly, quarterly, yearly), we need to compute
    # the expected timedelta using calendar arithmetic
    if expected_interval_td is None:
        # Use add_interval to compute what the expected next time should be
        last_time = time_p["time"].item()  # Extract scalar from single-row DataFrame
        expected_next_time = add_interval(last_time, expected_interval, 1)
        first_time = time_n["time"].item()  # Extract scalar from single-row DataFrame

        if first_time != expected_next_time:
            if first_time > expected_next_time:
                raise ValueError(
                    f"Gap detected between DataFrames: previous DataFrame ends at {last_time}, "
                    f"next DataFrame starts at {first_time}, expected {expected_next_time} "
                    f"(interval {expected_interval})."
                )
            else:
                raise ValueError(
                    f"Overlap detected between DataFrames: previous DataFrame ends at {last_time}, "
                    f"next DataFrame starts at {first_time}, expected {expected_next_time} "
                    f"(interval {expected_interval})."
                )
    # Fixed interval - can use timedelta comparison
    elif interval_td != expected_interval_td:
        last_time_p = time_p[0, 0]
        first_time_n = time_n[0, 0]
        if interval_td > expected_interval_td:
            raise ValueError(
                f"Gap detected between DataFrames: previous DataFrame ends at {last_time_p}, "
                f"next DataFrame starts at {first_time_n}, creating a gap of {interval_td} "
                f"(expected {expected_interval})."
            )
        else:
            raise ValueError(
                f"Overlap detected between DataFrames: previous DataFrame ends at "
                f"{last_time_p}, next DataFrame starts at {first_time_n}, with interval "
                f"{interval_td} (expected {expected_interval})."
            )