File size: 2,805 Bytes
12fb8be | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | """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 |