Date Utils

Utility functions for time-related operations in retail analysis.

filter_and_label_by_periods(transactions, period_ranges, period_col='period_name')

Filter transactions to named time periods and label each row with its period.

Rows outside every period are dropped. The date column is read from the column.transaction_date option.

Parameters:

Name	Type	Description	Default
transactions	DataFrame | Table	Transactions containing the transaction date column.	required
period_ranges	Mapping[str, tuple[datetime | date | str, datetime | date | str]]	Period names mapped to inclusive (start_date, end_date) tuples. A date or "YYYY-MM-DD" string end includes that whole day. A datetime end is an exact instant on a timestamp column; on a date column it is cast to its day, so that whole day is included. A timezone-aware datetime is read as wall-clock time: its offset is ignored.	required
period_col	str	Name of the label column to add. Defaults to "period_name".	'period_name'

Returns:

Type	Description
Table	ibis.Table: Filtered table with the period_col label column.

Raises:

Type	Description
TypeError	If transactions is neither a pandas DataFrame nor an ibis Table, if its transaction date column is not a date or datetime type, if period_ranges is not a mapping, or if a period bound is not a string, date or datetime.
ValueError	If period_ranges is empty, if a period value is not a (start, end) tuple, if a bound string is not a valid YYYY-MM-DD date, if a start date is after its end date, if two periods overlap, if period_col is already a column, or if the transaction date column is missing or timezone-aware.

Examples:

>>> from openretailscience.utils import filter_and_label_by_periods
>>> periods = {"H1 2024": ("2024-01-01", "2024-06-30"), "H1 2025": ("2025-01-01", "2025-06-30")}
>>> h1 = filter_and_label_by_periods(transactions, period_ranges=periods)
>>> h1.group_by("period_name").aggregate(spend=h1.unit_spend.sum()).order_by("period_name").to_pandas()
  period_name    spend
0     H1 2024  1194.05
1     H1 2025  1350.15

Source code in openretailscience/utils/date.py

def filter_and_label_by_periods(
    transactions: pd.DataFrame | ibis.Table,
    period_ranges: Mapping[str, tuple[datetime | date | str, datetime | date | str]],
    period_col: str = "period_name",
) -> ibis.Table:
    """Filter transactions to named time periods and label each row with its period.

    Rows outside every period are dropped. The date column is read from the
    ``column.transaction_date`` option.

    Args:
        transactions (pd.DataFrame | ibis.Table): Transactions containing the transaction date column.
        period_ranges (Mapping[str, tuple[datetime | date | str, datetime | date | str]]): Period
            names mapped to inclusive ``(start_date, end_date)`` tuples. A ``date`` or
            ``"YYYY-MM-DD"`` string end includes that whole day. A datetime end is an exact instant
            on a timestamp column; on a date column it is cast to its day, so that whole day is
            included. A timezone-aware datetime is read as wall-clock time: its offset is ignored.
        period_col (str): Name of the label column to add. Defaults to ``"period_name"``.

    Returns:
        ibis.Table: Filtered table with the ``period_col`` label column.

    Raises:
        TypeError: If ``transactions`` is neither a pandas DataFrame nor an ibis Table, if its
            transaction date column is not a date or datetime type, if ``period_ranges`` is not a
            mapping, or if a period bound is not a string, date or datetime.
        ValueError: If ``period_ranges`` is empty, if a period value is not a ``(start, end)``
            tuple, if a bound string is not a valid ``YYYY-MM-DD`` date, if a start date is after
            its end date, if two periods overlap, if ``period_col`` is already a column, or if the
            transaction date column is missing or timezone-aware.

    Examples:
        >>> from openretailscience.utils import filter_and_label_by_periods
        >>> periods = {"H1 2024": ("2024-01-01", "2024-06-30"), "H1 2025": ("2025-01-01", "2025-06-30")}
        >>> h1 = filter_and_label_by_periods(transactions, period_ranges=periods)
        >>> h1.group_by("period_name").aggregate(spend=h1.unit_spend.sum()).order_by("period_name").to_pandas()
          period_name    spend
        0     H1 2024  1194.05
        1     H1 2025  1350.15
    """
    if not isinstance(period_ranges, Mapping):
        msg = (
            "period_ranges must be a mapping of period names to (start_date, end_date) tuples. "
            f"Got {type(period_ranges).__name__}."
        )
        raise TypeError(msg)
    transactions = ensure_ibis_table(transactions, "transactions")
    date_col = get_option("column.transaction_date")
    ensure_data_has_columns(transactions, [date_col])
    if period_col in transactions.columns:
        msg = f"period_col '{period_col}' is already a column in transactions. Pass a different period_col."
        raise ValueError(msg)
    ensure_tznaive_datetime(transactions, date_col)
    date_column = transactions[date_col]
    date_col_dtype = date_column.type()

    period_ranges = {
        name: tuple(_normalize_bound(bound, date_col_dtype.is_date()) for bound in bounds)
        if isinstance(bounds, tuple)
        else bounds
        for name, bounds in period_ranges.items()
    }
    _validate_and_normalize_periods(period_ranges)

    branches = []
    for period_name, date_range in period_ranges.items():
        start_date, end_date = date_range

        start = ibis.literal(start_date, type=date_col_dtype)
        if isinstance(end_date, str):
            # "< next day" rather than <= 23:59:59.999999, which a millisecond backend (e.g. Postgres
            # TIMESTAMP(3)) rounds up to the next midnight.
            next_day = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)  # noqa: DTZ007
            end = ibis.literal(next_day, type=date_col_dtype)
            period_condition = (date_column >= start) & (date_column < end)
        else:
            period_condition = date_column.between(start, ibis.literal(end_date, type=date_col_dtype))
        branches.append((period_condition, ibis.literal(period_name)))

    conditions = ibis.or_(*[condition[0] for condition in branches])
    return transactions.filter(conditions).mutate(**{period_col: ibis.cases(*branches)})

find_overlapping_periods(start_date, end_date, return_str=True)

Split a date range into year-aligned overlapping periods.

The first period starts at start_date; each subsequent period starts on the same month and day one year later. Each period ends on end_date's month and day in the following year, so the last period ends exactly at end_date. Both dates in the same year yield an empty list.

Note

No leap-year adjustment: a Feb 29 start or end date raises ValueError in a non-leap year.

Parameters:

Name	Type	Description	Default
start_date	datetime | str	Range start, a datetime or YYYY-MM-DD string.	required
end_date	datetime | str	Range end, a datetime or YYYY-MM-DD string.	required
return_str	bool	If True, dates are returned as ISO-formatted strings; if False, as datetime objects. Defaults to True.	True

Returns:

Type	Description
list[tuple[str | datetime, str | datetime]]	list[tuple[str | datetime, str | datetime]]: (start, end) pairs; datetime outputs preserve the input timezone-awareness (string inputs give naive outputs).

Raises:

Type	Description
TypeError	If start_date and end_date have mismatched timezone awareness.
ValueError	If the start date is after the end date.

Examples:

>>> from openretailscience.utils import find_overlapping_periods
>>> find_overlapping_periods("2023-10-01", "2025-12-31")
[('2023-10-01', '2024-12-31'), ('2024-10-01', '2025-12-31')]

Source code in openretailscience/utils/date.py

def find_overlapping_periods(
    start_date: datetime | str,
    end_date: datetime | str,
    return_str: bool = True,
) -> list[tuple[str | datetime, str | datetime]]:
    """Split a date range into year-aligned overlapping periods.

    The first period starts at ``start_date``; each subsequent period starts on the same
    month and day one year later. Each period ends on ``end_date``'s month and day in the
    following year, so the last period ends exactly at ``end_date``. Both dates in the
    same year yield an empty list.

    Note:
        No leap-year adjustment: a Feb 29 start or end date raises ``ValueError`` in a
        non-leap year.

    Args:
        start_date (datetime | str): Range start, a datetime or ``YYYY-MM-DD`` string.
        end_date (datetime | str): Range end, a datetime or ``YYYY-MM-DD`` string.
        return_str (bool): If True, dates are returned as ISO-formatted strings; if
            False, as datetime objects. Defaults to True.

    Returns:
        list[tuple[str | datetime, str | datetime]]: ``(start, end)`` pairs; datetime
            outputs preserve the input timezone-awareness (string inputs give naive outputs).

    Raises:
        TypeError: If ``start_date`` and ``end_date`` have mismatched timezone awareness.
        ValueError: If the start date is after the end date.

    Examples:
        >>> from openretailscience.utils import find_overlapping_periods
        >>> find_overlapping_periods("2023-10-01", "2025-12-31")
        [('2023-10-01', '2024-12-31'), ('2024-10-01', '2025-12-31')]
    """
    # Track whether outputs should be tz-naive to preserve backward compatibility.
    # String inputs and naive datetime inputs both produced naive outputs before.
    start_is_naive = _is_naive(start_date)
    end_is_naive = _is_naive(end_date)

    if start_is_naive != end_is_naive:
        msg = "start_date and end_date must have matching timezone awareness. Got naive and aware (or vice versa)."
        raise TypeError(msg)

    start_date = _normalize_datetime(start_date)
    end_date = _normalize_datetime(end_date)

    if start_date > end_date:
        raise ValueError("Start date must be before end date")

    if start_date.year == end_date.year:
        return []

    if start_is_naive:
        output_tz = None
        start_date = start_date.replace(tzinfo=None)
    else:
        output_tz = start_date.tzinfo

    period_starts = [
        start_date if year == start_date.year else datetime(year, start_date.month, start_date.day, tzinfo=output_tz)
        for year in range(start_date.year, end_date.year)
    ]
    period_ends = [
        datetime(year + 1, end_date.month, end_date.day, tzinfo=output_tz)
        for year in range(start_date.year, end_date.year)
    ]

    if return_str:
        return [
            (s.strftime("%Y-%m-%d"), e.strftime("%Y-%m-%d")) for s, e in zip(period_starts, period_ends, strict=True)
        ]

    return list(zip(period_starts, period_ends, strict=True))