File size: 1,769 Bytes
938949f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """
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
|