check_continuity

yohou.utils.validation.check_continuity(df_p, df_n, expected_interval, check_intervals=True)

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

Name	Type	Description	Default
df_p	DataFrame	Previous (earlier) time series DataFrame with "time" column.	required
df_n	DataFrame	Next (later) time series DataFrame with "time" column.	required
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).	required
check_intervals	bool	If True, validates that both DataFrames have consistent internal intervals.	True

Raises

Type	Description
ValueError	If there is a gap or overlap between df_p and df_n, or if internal intervals don't match expected_interval.

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 : Validates uniform time spacing

Source Code

View on GitHub

Show/Hide source

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.

    Raises
    ------
    ValueError
        If there is a gap or overlap between df_p and df_n, or if internal
        intervals don't match expected_interval.

    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:
                interval_p_norm = _normalize_interval(interval_p)
                if 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})."
            )