| """Macro Features - FRED data, yield curve, VIX, credit spreads""" |
| import numpy as np |
| import pandas as pd |
| from typing import List |
|
|
| class MacroFeatures: |
| """Macroeconomic and sentiment overlay features""" |
| |
| @staticmethod |
| def fetch_fred_data(series_ids, start='2020-01-01', end='2024-01-01'): |
| """Fetch macro data from FRED. Series: DGS10, DGS2, T10Y2Y, VIXCLS, UNRATE, BAA10YM""" |
| try: |
| from fredapi import Fred |
| import os |
| fred = Fred(api_key=os.environ.get('FRED_API_KEY', '')) |
| data = {} |
| for sid in series_ids: |
| try: |
| data[sid] = fred.get_series(sid, observation_start=start, observation_end=end) |
| except Exception as e: |
| print(f"FRED {sid}: {e}") |
| return pd.DataFrame(data) |
| except ImportError: |
| return MacroFeatures._synthetic_macro(start, end) |
| |
| @staticmethod |
| def _synthetic_macro(start='2020-01-01', end='2024-01-01'): |
| dates = pd.bdate_range(start=start, end=end) |
| n = len(dates) |
| np.random.seed(42) |
| level = np.cumsum(np.random.normal(0, 0.01, n)) + 2.0 |
| return pd.DataFrame({ |
| 'DGS10': level + np.random.normal(0, 0.05, n), |
| 'DGS2': level * 0.6 + np.random.normal(0, 0.03, n), |
| 'T10Y2Y': level * 0.4 + np.random.normal(0, 0.02, n), |
| 'VIXCLS': 18 + np.random.normal(0, 3, n).clip(10, 45), |
| 'BAA10YM': 2.5 + np.random.normal(0, 0.2, n), |
| }, index=dates) |
| |
| @staticmethod |
| def yield_curve_features(treasury_10y, treasury_2y): |
| features = pd.DataFrame(index=treasury_10y.index) |
| features['yc_spread'] = treasury_10y - treasury_2y |
| features['yc_slope'] = features['yc_spread'].diff(21) |
| features['yc_inversion'] = (features['yc_spread'] < 0).astype(float) |
| features['yc_zscore'] = (features['yc_spread'] - features['yc_spread'].rolling(252).mean()) / features['yc_spread'].rolling(252).std().replace(0, 1) |
| return features |
| |
| @staticmethod |
| def vix_features(vix): |
| features = pd.DataFrame(index=vix.index) |
| features['vix_level'] = vix |
| features['vix_change'] = vix.pct_change() |
| features['vix_zscore'] = (vix - vix.rolling(63).mean()) / vix.rolling(63).std().replace(0, 1) |
| features['vix_term'] = vix.rolling(5).mean() - vix.rolling(21).mean() |
| return features |
| |
| @staticmethod |
| def credit_spread_features(spread): |
| features = pd.DataFrame(index=spread.index) |
| features['credit_spread'] = spread |
| features['credit_change'] = spread.diff(21) |
| features['credit_zscore'] = (spread - spread.rolling(252).mean()) / spread.rolling(252).std().replace(0, 1) |
| return features |
