spb/wp7_uqo Public

1# =============================================================================2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai4# =============================================================================5"""Portfolio construction: daily cross-sectional quintile sorts, double6sorts, and performance statistics (annualized returns, Sharpe ratios)."""78import numpy as np9import pandas as pd101112def portfolio_sort(data: pd.DataFrame, sort_var: str, ret_var: str,13                   n_quantiles: int = 5):14    """Daily equal-weighted quantile portfolios sorted on ``sort_var``.1516    Returns a dict keyed by quantile (1..n) plus ``'LS_5_1'`` (top minus17    bottom), each holding performance statistics, or ``None`` when the18    sample is too small. Annualization: 252 for 1-day returns, 52 otherwise.19    """20    data = data[[sort_var, ret_var, 'ticker', 'trade_date']].dropna()21    if len(data) < 1000:22        return None2324    data['quantile'] = data.groupby('trade_date')[sort_var].transform(25        lambda x: pd.qcut(x, n_quantiles, labels=False, duplicates='drop') + 126        if len(x) >= n_quantiles else np.nan27    )28    data = data.dropna(subset=['quantile'])29    data['quantile'] = data['quantile'].astype(int)3031    port_ret = data.groupby(['trade_date', 'quantile'])[ret_var].mean().reset_index()32    port_ret = port_ret.pivot(index='trade_date', columns='quantile', values=ret_var)3334    # Long-short: top quantile minus bottom (key 'LS_5_1' for quintiles,35    # 'LS_10_1' for deciles, ...)36    ls_key = f'LS_{n_quantiles}_1'37    if n_quantiles in port_ret.columns and 1 in port_ret.columns:38        port_ret[ls_key] = port_ret[n_quantiles] - port_ret[1]39    else:40        return None4142    results = {}43    for q in list(range(1, n_quantiles + 1)) + [ls_key]:44        if q not in port_ret.columns:45            continue46        s = port_ret[q].dropna()47        ann_factor = 252 if '1d' in ret_var or ret_var == 'ret_1d' else 5248        results[q] = {49            'mean_daily': s.mean(),50            'std_daily': s.std(),51            'annualized_return': s.mean() * ann_factor,52            'annualized_vol': s.std() * np.sqrt(ann_factor),53            'sharpe': (s.mean() / s.std() * np.sqrt(ann_factor)) if s.std() > 0 else 0,54            't_stat': (s.mean() / (s.std() / np.sqrt(len(s)))) if s.std() > 0 else 0,55            'n_days': len(s),56            'pct_positive': (s > 0).mean(),57            'max_drawdown': (s.cumsum() - s.cumsum().cummax()).min(),58        }59    return results606162def double_sort(data: pd.DataFrame, var1: str, var2: str, ret_var: str,63                n1: int = 3, n2: int = 3) -> pd.DataFrame:64    """Independent daily double sort; mean returns and t-stats per cell."""65    data = data.copy()66    for var, col, n in ((var1, 'q1', n1), (var2, 'q2', n2)):67        data[col] = data.groupby('trade_date')[var].transform(68            lambda x, n=n: pd.qcut(x, n, labels=False, duplicates='drop') + 169            if len(x) >= n else np.nan70        )71    data = data.dropna(subset=['q1', 'q2'])72    data['q1'] = data['q1'].astype(int)73    data['q2'] = data['q2'].astype(int)7475    results = data.groupby(['q1', 'q2'])[ret_var].agg(['mean', 'std', 'count']).reset_index()76    results['mean_bps'] = results['mean'] * 1000077    results['t_stat'] = results['mean'] / (results['std'] / np.sqrt(results['count']))78    return results79