spb/wp7_uqo Public

1# =============================================================================2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai4# =============================================================================5"""Realized-variance forecasting utilities (RQ2/RQ5): OLS forecasts on raw6feature levels, rolling out-of-sample evaluation, and loss functions."""78import numpy as np9import pandas as pd10from numpy.linalg import lstsq111213def ols_forecast(train: pd.DataFrame, test: pd.DataFrame,14                 features: list, target: str) -> dict:15    """Fit OLS on ``train`` and evaluate on ``test`` (non-negative forecasts).1617    Features enter in levels (no standardization), matching the original18    forecasting design. Returns MSE, MAE, out-of-sample R², QLIKE and the19    fitted coefficients.20    """21    X_train = np.column_stack([np.ones(len(train)), train[features].values])22    y_train = train[target].values23    coefs, _, _, _ = lstsq(X_train, y_train, rcond=None)2425    X_test = np.column_stack([np.ones(len(test)), test[features].values])26    y_test = test[target].values27    y_pred = np.maximum(X_test @ coefs, 0)2829    mse = np.mean((y_test - y_pred) ** 2)30    mae = np.mean(np.abs(y_test - y_pred))31    ss_res = np.sum((y_test - y_pred) ** 2)32    ss_tot = np.sum((y_test - y_test.mean()) ** 2)33    r2_oos = 1 - ss_res / ss_tot if ss_tot > 0 else 03435    y_pred_safe = np.maximum(y_pred, 1e-10)36    qlike = np.mean(y_test / y_pred_safe - np.log(y_test / y_pred_safe) - 1)3738    return {'mse': mse, 'mae': mae, 'r2_oos': r2_oos, 'qlike': qlike, 'coefs': coefs}394041def rolling_evaluation(data: pd.DataFrame, models: dict, target_col: str,42                       target_label: str, window: int = 500,43                       step: int = 250) -> pd.DataFrame:44    """Per-ticker rolling-window out-of-sample evaluation of competing models.4546    For each ticker and model, fits on ``window`` observations and evaluates47    on the following ``step`` observations, advancing by ``step``. Returns48    one row per (ticker, model) with average losses across windows.49    """50    all_results = []51    for ticker in data['ticker'].unique():52        td = data[data['ticker'] == ticker].reset_index(drop=True)53        if len(td) < window + 100:54            continue5556        for model_name, features in models.items():57            sub = td[features + [target_col]].dropna()58            if len(sub) < window + 50:59                continue6061            losses = {'mse': [], 'mae': [], 'r2': [], 'qlike': []}62            for start in range(0, len(sub) - window - 50, step):63                train = sub.iloc[start:start + window]64                test = sub.iloc[start + window:start + window + step]65                if len(test) < 10:66                    continue67                try:68                    res = ols_forecast(train, test, features, target_col)69                except np.linalg.LinAlgError:70                    continue71                losses['mse'].append(res['mse'])72                losses['mae'].append(res['mae'])73                losses['r2'].append(res['r2_oos'])74                losses['qlike'].append(res['qlike'])7576            if losses['mse']:77                all_results.append({78                    'ticker': ticker,79                    'model': model_name,80                    'target': target_label,81                    'avg_mse': np.mean(losses['mse']),82                    'avg_mae': np.mean(losses['mae']),83                    'avg_r2_oos': np.mean(losses['r2']),84                    'avg_qlike': np.mean(losses['qlike']),85                    'n_windows': len(losses['mse']),86                })87    return pd.DataFrame(all_results)888990def diebold_mariano(e1: np.ndarray, e2: np.ndarray) -> float:91    """DM statistic on squared-error differentials (positive favors model 2)."""92    d = e1 ** 2 - e2 ** 293    return np.mean(d) / (np.std(d) / np.sqrt(len(d))) if np.std(d) > 0 else 094