""" Shared utilities for time-based feature engineering. Centralises cyclical encodings for hour-of-day and day-of-year so that Preprocessor, ChronosForecaster, and LLMDataEngineer use the same logic. """ from __future__ import annotations from typing import Literal import numpy as np import pandas as pd def add_cyclical_time_features( df: pd.DataFrame, timestamp_col: str | None = None, index_is_timestamp: bool = False, day_period: float = 365.25, ) -> pd.DataFrame: """ Add hour_sin/hour_cos and doy_sin/doy_cos to a DataFrame. Parameters ---------- df : DataFrame Input data. timestamp_col : str or None Column containing timestamps; if None and index_is_timestamp=True, the index is used as the timestamp source. index_is_timestamp : bool Whether to treat the index as the timestamp source when timestamp_col is None. day_period : float Period for day-of-year cycle (default 365.25). Returns ------- DataFrame Copy of df with four extra columns: hour_sin, hour_cos, doy_sin, doy_cos. """ out = df.copy() if timestamp_col is not None and timestamp_col in out.columns: ts = pd.to_datetime(out[timestamp_col], utc=True) elif index_is_timestamp and isinstance(out.index, pd.DatetimeIndex): ts = out.index else: # No-op if we cannot resolve a timestamp return out hour = ts.dt.hour + ts.dt.minute / 60.0 doy = ts.dt.dayofyear.astype(float) out["hour_sin"] = np.sin(2 * np.pi * hour / 24.0) out["hour_cos"] = np.cos(2 * np.pi * hour / 24.0) out["doy_sin"] = np.sin(2 * np.pi * doy / day_period) out["doy_cos"] = np.cos(2 * np.pi * doy / day_period) return out