Time Series Features

A time-series feature engineering skill. Produces lag, rolling, calendar, and decomposition features with rigorous protection against temporal data leakage.

When to use this skill

Activate when the user has a time-ordered dataset and wants to prepare features for forecasting or temporal modeling. Typical signals:

• "Create lag features for this sales data"
• "Generate rolling window features"
• "Prepare this time series for forecasting"
• "What's the seasonality in this data?"
• "Génère des features temporelles", "fais du feature engineering pour des séries temporelles"

Pre-conditions:

• The dataset MUST have a datetime column (or a column convertible to datetime)
• The dataset SHOULD be sorted chronologically (skill will verify and sort if not)
• The dataset SHOULD have a regular frequency (daily, hourly, monthly) — skill will detect and warn if irregular

Do NOT activate this skill for:

• Cross-sectional datasets without time order → use feature-engineer
• Pure exploration → use eda-explorer
• Single time series without features (just the target) → suggest feature-engineer first to add covariates

Workflow

For every dataset, follow these 7 phases in order. Never skip phase 1 or 7 — they prevent data leakage.

Phase 1 — Detect & validate temporal structure

Run these checks first:

1. Identify the datetime column. If user specified, use it. Otherwise, auto-detect by trying to parse each column as datetime; the one with most successes wins.
2. Convert to datetime if needed: pd.to_datetime(df[col], errors='coerce'). Drop NaT rows with a warning.
3. Sort chronologically: df.sort_values(datetime_col).reset_index(drop=True).
4. Detect frequency: use pd.infer_freq() on the index. Common frequencies: D (daily), H (hourly), MS (month-start), W (weekly), min (minute).
5. Check for gaps: if irregular, warn the user but proceed with caution.
6. Identify the target column (if user specified). Otherwise, ask before proceeding — features without a target are still valid but the user should confirm.

Output a summary table:

| Property | Value |
|---|---|
| Datetime column | contract_start |
| Frequency detected | D |
| First timestamp | 2018-01-01 |
| Last timestamp | 2024-11-15 |
| Total observations | 2,000 |
| Gaps detected | 12 (0.6%) |
| Target column | n_claims |

Phase 2 — Datetime decomposition

For the datetime column, extract these calendar features (always trailing — past info only):

df["year"] = df[dt_col].dt.year
df["month"] = df[dt_col].dt.month
df["day"] = df[dt_col].dt.day
df["dayofweek"] = df[dt_col].dt.dayofweek         # 0=Monday, 6=Sunday
df["dayofyear"] = df[dt_col].dt.dayofyear
df["weekofyear"] = df[dt_col].dt.isocalendar().week
df["quarter"] = df[dt_col].dt.quarter
df["is_weekend"] = (df[dt_col].dt.dayofweek >= 5).astype(int)
df["is_month_start"] = df[dt_col].dt.is_month_start.astype(int)
df["is_month_end"] = df[dt_col].dt.is_month_end.astype(int)
df["is_quarter_end"] = df[dt_col].dt.is_quarter_end.astype(int)

If frequency is hourly or finer, also add hour, minute.

Phase 3 — Business calendar features

Use the holidays package (install if needed: pip install holidays).

Default country: France (FR). Allow user to override via parameter.

import holidays
fr_holidays = holidays.France(years=range(df[dt_col].dt.year.min(), df[dt_col].dt.year.max() + 1))

df["is_holiday"] = df[dt_col].dt.date.apply(lambda d: int(d in fr_holidays))
df["days_to_christmas"] = ((pd.Timestamp(year=df[dt_col].dt.year, month=12, day=25) - df[dt_col]).dt.days).where(...)
df["days_since_year_start"] = (df[dt_col] - pd.to_datetime(df[dt_col].dt.year.astype(str) + '-01-01')).dt.days

For datasets with strong business cycles, add:

• is_payday (typically last business day of month)
• is_quarter_end
• days_since_last_holiday and days_to_next_holiday

Phase 4 — Lag features

For the target column AND any user-specified covariates:

Default lag values based on frequency:

• Daily data: [1, 7, 14, 28] (yesterday, last week, two weeks ago, four weeks ago)
• Hourly data: [1, 24, 168] (last hour, yesterday same hour, last week same hour)
• Monthly data: [1, 3, 12] (last month, last quarter, last year)
• Weekly data: [1, 4, 52]

for lag in lag_values:
    df[f"{col}_lag{lag}"] = df[col].shift(lag)

Allow user to override with custom lag values.

⚠️ CRITICAL: NaN values appear at the start of the series (first max(lags) rows). Document this in the output. Do NOT drop them automatically — let the user decide whether to drop or impute (impacts model behavior).

Phase 5 — Rolling features

For the target and user-specified covariates, compute these aggregations on trailing windows:

Default windows based on frequency:

• Daily: [7, 14, 30]
• Hourly: [24, 168, 720]
• Monthly: [3, 6, 12]

Default aggregations: mean, std, min, max, median.

for w in windows:
    df[f"{col}_rolling_mean_{w}"] = df[col].rolling(window=w, min_periods=1).mean()
    df[f"{col}_rolling_std_{w}"] = df[col].rolling(window=w, min_periods=1).std()
    df[f"{col}_rolling_min_{w}"] = df[col].rolling(window=w, min_periods=1).min()
    df[f"{col}_rolling_max_{w}"] = df[col].rolling(window=w, min_periods=1).max()

⚠️ NEVER use `center=True` — that's leakage. Default is trailing, leave it.

⚠️ Apply ONLY after the chronological split if the user provides train and test separately. If applied before split, the rolling near the train/test boundary leaks test info into train.

Phase 6 — Decomposition & stationarity diagnostics

Decompose the target series and run ADF test.

from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.stattools import adfuller

# Decomposition (additive by default)
decomposition = seasonal_decompose(df[target].dropna(), model='additive',
                                    period=detect_period(frequency))
trend = decomposition.trend
seasonal = decomposition.seasonal
residual = decomposition.resid

# Stationarity test
adf_result = adfuller(df[target].dropna())
adf_pvalue = adf_result[1]
is_stationary = adf_pvalue < 0.05

Period selection based on frequency:

• Daily → 7 (weekly seasonality) or 365 (yearly)
• Hourly → 24 (daily seasonality)
• Monthly → 12 (yearly seasonality)
• Weekly → 52 (yearly seasonality)

Output to user:

• Trend, seasonal, residual plots saved as PNG to outputs/timeseries/
• ADF p-value with interpretation:
• p < 0.01 → HIGHLY STATIONARY — model can be applied directly
• p < 0.05 → STATIONARY — model OK
• p ≥ 0.05 → NON-STATIONARY — recommend differencing (df[target].diff())

Phase 7 — Chronological train/test split

MANDATORY step. Never use random split.

from sklearn.model_selection import TimeSeriesSplit

# Single split (default 80/20)
split_idx = int(len(df) * 0.8)
df_train = df.iloc[:split_idx]
df_test = df.iloc[split_idx:]

# OR if user wants cross-validation
tscv = TimeSeriesSplit(n_splits=5)

Confirm to user:

• Training period: <first> → <split_date>
• Testing period: <split_date+1> → <last>
• Train rows: N, Test rows: M
• Target distribution preserved (mean/std comparable)

Output format

Respond with a markdown report:

# Time Series Feature Engineering Report

## Source
- File: <name>
- Datetime column: <col>
- Frequency: <freq>
- Period: <first> → <last>
- Total observations: N
- Target: <col>

## Generated features (X total)
| Category | Count | Examples |
| Datetime | 11 | year, month, day, dayofweek, ... |
| Business | 4 | is_holiday, days_to_christmas, ... |
| Lag | 4 | target_lag1, target_lag7, ... |
| Rolling | 15 | target_rolling_mean_7, target_rolling_std_30, ... |

## Stationarity diagnosis
- ADF p-value: 0.034
- Verdict: STATIONARY — modeling can proceed
- (Or: NON-STATIONARY — recommend df[target].diff() before modeling)

## Decomposition
- Trend captured (R² = 0.87)
- Seasonality detected: weekly (period=7)
- Plots saved: outputs/timeseries/decomposition.png

## Train/Test split (chronological)
- Train: 2018-01-01 → 2023-08-15 (1600 rows, 80%)
- Test: 2023-08-16 → 2024-11-15 (400 rows, 20%)

## Artifacts
- outputs/timeseries/train_features.parquet
- outputs/timeseries/test_features.parquet
- outputs/timeseries/feature_dict.json
- outputs/timeseries/decomposition.png
- outputs/timeseries/adf_diagnostics.json

## Next steps
1. Train a baseline model (ARIMA, Prophet, or gradient boosting)
2. Use TimeSeriesSplit for cross-validation
3. Watch out for: NaN in early rows due to lags

Constraints

• Always sort chronologically before any feature engineering
• Always use rolling() with default trailing window (no center=True)
• Always split chronologically — never random
• Always apply rolling/lag features after the train/test split if user provides them separately
• Always run ADF test and report stationarity verdict
• Never drop NaN rows from lags automatically — document and let user decide
• Never suggest random shuffling or random split for time series data
• Output language: match user (FR or EN)
• Default country for holidays: France (FR). User can override.

Code skeleton

import pandas as pd
import numpy as np
from pathlib import Path
import json
import joblib

import holidays
from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.stattools import adfuller
from sklearn.model_selection import TimeSeriesSplit
import matplotlib.pyplot as plt

# Setup
OUTPUT_DIR = Path("outputs/timeseries")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
SKILL_VERSION = "0.1.0"

# Phase 1: Detect & validate
def detect_datetime_column(df):
    """Try to parse each column as datetime; return the best match."""
    candidates = {}
    for col in df.columns:
        try:
            converted = pd.to_datetime(df[col], errors='coerce')
            success_rate = converted.notna().mean()
            if success_rate > 0.9:
                candidates[col] = success_rate
        except Exception:
            continue
    return max(candidates, key=candidates.get) if candidates else None

def validate_and_sort(df, dt_col):
    df[dt_col] = pd.to_datetime(df[dt_col], errors='coerce')
    df = df.dropna(subset=[dt_col]).sort_values(dt_col).reset_index(drop=True)
    freq = pd.infer_freq(df[dt_col])
    return df, freq

# Phase 2-3: Calendar features
def add_datetime_features(df, dt_col, country='FR'):
    s = df[dt_col]
    df["year"] = s.dt.year
    df["month"] = s.dt.month
    df["day"] = s.dt.day
    df["dayofweek"] = s.dt.dayofweek
    df["weekofyear"] = s.dt.isocalendar().week.astype(int)
    df["quarter"] = s.dt.quarter
    df["is_weekend"] = (s.dt.dayofweek >= 5).astype(int)
    df["is_month_start"] = s.dt.is_month_start.astype(int)
    df["is_month_end"] = s.dt.is_month_end.astype(int)
    df["is_quarter_end"] = s.dt.is_quarter_end.astype(int)

    # Holidays
    years = range(s.dt.year.min(), s.dt.year.max() + 1)
    country_holidays = holidays.country_holidays(country, years=years)
    df["is_holiday"] = s.dt.date.apply(lambda d: int(d in country_holidays))

    return df

# Phase 4: Lag features
def add_lag_features(df, target, lags):
    for lag in lags:
        df[f"{target}_lag{lag}"] = df[target].shift(lag)
    return df

# Phase 5: Rolling features
def add_rolling_features(df, target, windows, aggs=("mean", "std", "min", "max")):
    for w in windows:
        for agg in aggs:
            df[f"{target}_rolling_{agg}_{w}"] = df[target].rolling(window=w, min_periods=1).agg(agg)
    return df

# Phase 6: Decomposition + ADF
def diagnose_stationarity(df, target, period):
    series = df[target].dropna()
    decomposition = seasonal_decompose(series, model='additive', period=period)
    adf_result = adfuller(series)
    adf_pvalue = adf_result[1]

    fig, axes = plt.subplots(4, 1, figsize=(10, 8))
    axes[0].plot(series.values); axes[0].set_title("Original")
    axes[1].plot(decomposition.trend.values); axes[1].set_title("Trend")
    axes[2].plot(decomposition.seasonal.values); axes[2].set_title("Seasonal")
    axes[3].plot(decomposition.resid.values); axes[3].set_title("Residual")
    fig.tight_layout()
    fig.savefig(OUTPUT_DIR / "decomposition.png", dpi=80)
    plt.close(fig)

    return {
        "adf_pvalue": float(adf_pvalue),
        "is_stationary": bool(adf_pvalue < 0.05),
        "interpretation": (
            "HIGHLY STATIONARY" if adf_pvalue < 0.01 else
            "STATIONARY" if adf_pvalue < 0.05 else
            "NON-STATIONARY — consider differencing"
        ),
    }

# Phase 7: Chronological split
def chronological_split(df, dt_col, test_size=0.2):
    split_idx = int(len(df) * (1 - test_size))
    train = df.iloc[:split_idx].copy()
    test = df.iloc[split_idx:].copy()
    return train, test