""" data/loader.py Loads master_data.parquet from HF Dataset. Engineers rich feature set from raw price/macro columns. No external pings — all data from HF Dataset only. Supports both FI and Equity ETF modules. """ import pandas as pd import numpy as np import streamlit as st from huggingface_hub import hf_hub_download from datetime import datetime, timedelta import pytz try: import pandas_market_calendars as mcal NYSE_CAL_AVAILABLE = True except ImportError: NYSE_CAL_AVAILABLE = False DATASET_REPO = "P2SAMAPA/fi-etf-macro-signal-master-data" PARQUET_FILE = "master_data.parquet" # ── Module-specific ETF definitions ─────────────────────────────────────────── FI_ETF_COLS = ["TLT", "VNQ", "SLV", "GLD", "LQD", "HYG", "VCIT"] EQUITY_ETF_COLS = ["QQQ", "XLK", "XLF", "XLE", "XLV", "XLI", "XLY", "XLP", "XLU", "XME", "GDX", "IWM"] BENCHMARK_COLS = ["SPY", "AGG"] TBILL_COL = "TBILL_3M" MACRO_COLS = ["VIX", "DXY", "T10Y2Y", "IG_SPREAD", "HY_SPREAD"] # Minimum non-NaN fraction a feature column must have to be included in dropna. MIN_COVERAGE = 0.80 # ── NYSE calendar ───────────────────────────────────────────────────────────── def get_last_nyse_trading_day(as_of=None): est = pytz.timezone("US/Eastern") if as_of is None: as_of = datetime.now(est) today = as_of.date() if NYSE_CAL_AVAILABLE: try: nyse = mcal.get_calendar("NYSE") sched = nyse.schedule(start_date=today - timedelta(days=10), end_date=today) if len(sched) > 0: return sched.index[-1].date() except Exception: pass candidate = today while candidate.weekday() >= 5: candidate -= timedelta(days=1) return candidate # ── Data loading ────────────────────────────────────────────────────────────── @st.cache_data(ttl=3600, show_spinner=False) def load_dataset(hf_token: str) -> pd.DataFrame: try: path = hf_hub_download( repo_id=DATASET_REPO, filename=PARQUET_FILE, repo_type="dataset", token=hf_token, ) df = pd.read_parquet(path) if not isinstance(df.index, pd.DatetimeIndex): for col in ["Date", "date", "DATE", "__index_level_0__"]: if col in df.columns: df = df.set_index(col) break df.index = pd.to_datetime(df.index) return df.sort_index() except Exception as e: st.error(f"❌ Failed to load dataset: {e}") return pd.DataFrame() # ── Freshness check ─────────────────────────────────────────────────────────── def check_data_freshness(df: pd.DataFrame) -> dict: if df.empty: return {"fresh": False, "message": "Dataset is empty."} last = df.index[-1].date() expect = get_last_nyse_trading_day() fresh = last >= expect msg = ( f"✅ Dataset up to date through {last}." if fresh else f"⚠️ {expect} data not yet updated. Latest: {last}. " f"Dataset updates daily after market close." ) return {"fresh": fresh, "last_date_in_data": last, "expected_date": expect, "message": msg} # ── Price → returns ─────────────────────────────────────────────────────────── def _to_returns(series: pd.Series) -> pd.Series: """Convert price series to daily pct returns. If already returns, pass through.""" clean = series.dropna() if len(clean) == 0: return series if abs(clean.median()) > 2: # price series return series.pct_change() return series # already returns # ── Feature engineering ─────────────────────────────────────────────────────── def _engineer_features(df: pd.DataFrame, ret_cols: list, module_type: str = "fi") -> pd.DataFrame: """ Build a rich feature set from raw macro + ETF return columns. Args: df: DataFrame with raw data ret_cols: List of return column names to use as targets module_type: "fi" or "equity" - determines feature emphasis """ feat = pd.DataFrame(index=df.index) # ── ETF return features ─────────────────────────────────────────────────── for col in ret_cols: r = df[col] feat[f"{col}_lag1"] = r.shift(1) feat[f"{col}_lag5"] = r.shift(5) feat[f"{col}_lag21"] = r.shift(21) feat[f"{col}_vol5"] = r.rolling(5).std() feat[f"{col}_vol21"] = r.rolling(21).std() feat[f"{col}_mom5"] = r.rolling(5).sum() feat[f"{col}_mom21"] = r.rolling(21).sum() # ── Macro features ──────────────────────────────────────────────────────── for col in MACRO_COLS: if col not in df.columns: continue s = df[col].ffill() roll_mean = s.rolling(252, min_periods=63).mean() roll_std = s.rolling(252, min_periods=63).std() feat[f"{col}_z"] = (s - roll_mean) / (roll_std + 1e-9) feat[f"{col}_chg5"] = s.diff(5) feat[f"{col}_lag1"] = s.shift(1) # ── TBILL level ─────────────────────────────────────────────────────────── if TBILL_COL in df.columns: tbill = df[TBILL_COL].ffill() feat["TBILL_level"] = tbill feat["TBILL_chg5"] = tbill.diff(5) # ── Derived cross-asset signals (module-specific) ───────────────────────── if module_type == "fi": # FI-specific spreads if "TLT_Ret" in df.columns and "AGG_Ret" in df.columns: feat["TLT_AGG_spread_mom5"] = ( df["TLT_Ret"].rolling(5).sum() - df["AGG_Ret"].rolling(5).sum() ) if "IG_SPREAD" in df.columns and "HY_SPREAD" in df.columns: feat["credit_ratio"] = ( df["HY_SPREAD"].ffill() / (df["IG_SPREAD"].ffill() + 1e-9) ) else: # Equity-specific signals if "QQQ_Ret" in df.columns and "SPY_Ret" in df.columns: feat["QQQ_SPY_spread_mom5"] = ( df["QQQ_Ret"].rolling(5).sum() - df["SPY_Ret"].rolling(5).sum() ) if "XLE_Ret" in df.columns and "XLK_Ret" in df.columns: feat["XLE_XLK_rotation"] = ( df["XLE_Ret"].rolling(5).sum() - df["XLK_Ret"].rolling(5).sum() ) if "XLF_Ret" in df.columns and "XLV_Ret" in df.columns: feat["XLF_XLV_spread"] = ( df["XLF_Ret"].rolling(5).sum() - df["XLV_Ret"].rolling(5).sum() ) # ── Universal regime indicators ─────────────────────────────────────────── if "VIX" in df.columns: vix = df["VIX"].ffill() feat["VIX_regime"] = (vix > 25).astype(float) feat["VIX_mom5"] = vix.diff(5) if "T10Y2Y" in df.columns: feat["YC_inverted"] = (df["T10Y2Y"].ffill() < 0).astype(float) return feat # ── Main extraction function (module-aware) ─────────────────────────────────── def get_features_and_targets(df: pd.DataFrame, module_type: str = "fi"): """ Build return columns for target ETFs and engineer a rich feature set. Args: df: Raw DataFrame module_type: "fi" or "equity" - determines which ETFs to use Returns: input_features, target_etfs, tbill_rate, df_out, col_info """ target_cols = FI_ETF_COLS if module_type == "fi" else EQUITY_ETF_COLS missing = [c for c in target_cols if c not in df.columns] if missing: raise ValueError( f"Missing {module_type.upper()} ETF columns: {missing}. " f"Found: {list(df.columns)}" ) col_info = {} rows_start = len(df) # ── Build ETF return columns ────────────────────────────────────────────── target_etfs = [] for col in target_cols: ret_col = f"{col}_Ret" df[ret_col] = _to_returns(df[col]) med = abs(df[col].dropna().median()) col_info[col] = ( f"price→pct_change (median={med:.2f})" if med > 2 else f"used as-is (median={med:.4f})" ) target_etfs.append(ret_col) # ── Build benchmark return columns ──────────────────────────────────────── for col in BENCHMARK_COLS: if col in df.columns: df[f"{col}_Ret"] = _to_returns(df[col]) # ── Drop NaN from first pct_change row ──────────────────────────────────── df = df.dropna(subset=target_etfs).copy() rows_after_ret = len(df) # ── Engineer features ───────────────────────────────────────────────────── feat_df = _engineer_features(df, target_etfs, module_type=module_type) for col in feat_df.columns: df[col] = feat_df[col].values feat_cols = list(feat_df.columns) # ── Smart dropna ────────────────────────────────────────────────────────── n = len(df) strict_cols = [] ffill_cols = [] for col in feat_cols: coverage = df[col].notna().sum() / n if n > 0 else 0 if coverage >= MIN_COVERAGE: strict_cols.append(col) else: ffill_cols.append(col) if ffill_cols: df[ffill_cols] = df[ffill_cols].ffill() if strict_cols: df = df.dropna(subset=strict_cols).copy() rows_after_feat = len(df) col_info["_diagnostics"] = ( f"rows: {rows_start} raw → {rows_after_ret} after ret dropna → " f"{rows_after_feat} after feature dropna | " f"strict_cols={len(strict_cols)} ffill_cols={len(ffill_cols)}" ) # ── T-bill rate ─────────────────────────────────────────────────────────── tbill_rate = 0.045 if TBILL_COL in df.columns: raw = df[TBILL_COL].dropna() if len(raw) > 0: v = float(raw.iloc[-1]) tbill_rate = v / 100 if v > 1 else v # ── Input features ──────────────────────────────────────────────────────── exclude = set( target_cols + BENCHMARK_COLS + target_etfs + [f"{c}_Ret" for c in BENCHMARK_COLS] + [TBILL_COL] + list(MACRO_COLS) ) input_features = [c for c in feat_cols if c not in exclude] if len(df) == 0: raise ValueError( f"No rows remain after feature engineering. " f"Diagnostics: {col_info['_diagnostics']}" ) return input_features, target_etfs, tbill_rate, df, col_info # ── Dataset summary (module-aware) ──────────────────────────────────────────── def dataset_summary(df: pd.DataFrame, module_type: str = "fi") -> dict: if df.empty: return {} target_cols = FI_ETF_COLS if module_type == "fi" else EQUITY_ETF_COLS return { "rows": len(df), "columns": len(df.columns), "start_date": df.index[0].strftime("%Y-%m-%d"), "end_date": df.index[-1].strftime("%Y-%m-%d"), "etfs_found": [c for c in target_cols if c in df.columns], "benchmarks": [c for c in BENCHMARK_COLS if c in df.columns], "macro_found": [c for c in MACRO_COLS if c in df.columns], "tbill_found": TBILL_COL in df.columns, "all_cols": list(df.columns), "module": module_type, }