File size: 2,504 Bytes
b54319d | 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 65 66 67 68 69 70 | import pandas as pd
import numpy as np
from typing import Dict, Any, Optional
from sklearn.linear_model import Ridge, LogisticRegression
class RawMomentumBaseline:
"""Baseline 1: Standard 20-day return cross-sectional ranking."""
def __init__(self, lookback_col: str = "mom_20d"):
self.lookback_col = lookback_col
def fit(self, df: pd.DataFrame, target_col: str):
pass
def predict(self, df: pd.DataFrame) -> pd.Series:
if self.lookback_col in df.columns:
return df[self.lookback_col].fillna(0.0)
return df["adjusted_close"].pct_change(20).fillna(0.0)
class VolAdjMomentumBaseline:
"""Baseline 2: Volatility-Adjusted Cross-Sectional Momentum (Return_20d / Vol_20d)."""
def __init__(self, feature_col: str = "vol_adj_mom_20d"):
self.feature_col = feature_col
def fit(self, df: pd.DataFrame, target_col: str):
pass
def predict(self, df: pd.DataFrame) -> pd.Series:
if self.feature_col in df.columns:
return df[self.feature_col].fillna(0.0)
return df["adjusted_close"].pct_change(20) / (df["adjusted_close"].pct_change(1).std() * np.sqrt(252) + 1e-6)
class LinearRidgeModel:
"""Linear Model 1: Ridge Regression with L2 Regularization."""
def __init__(self, alpha: float = 1.0):
self.alpha = alpha
self.model = Ridge(alpha=alpha)
def fit(self, df: pd.DataFrame, feature_cols: list, target_col: str):
clean_df = df[feature_cols + [target_col]].dropna()
if len(clean_df) > 10:
self.model.fit(clean_df[feature_cols], clean_df[target_col])
def predict(self, df: pd.DataFrame, feature_cols: list) -> pd.Series:
X = df[feature_cols].fillna(0.0)
return pd.Series(self.model.predict(X), index=df.index)
class LogisticClassificationModel:
"""Linear Model 2: L2 Regularized Logistic Regression for Top Quantile Classification."""
def __init__(self, C: float = 1.0):
self.C = C
self.model = LogisticRegression(C=C, max_iter=500)
def fit(self, df: pd.DataFrame, feature_cols: list, target_col: str):
clean_df = df[feature_cols + [target_col]].dropna()
if len(clean_df) > 10:
self.model.fit(clean_df[feature_cols], clean_df[target_col])
def predict(self, df: pd.DataFrame, feature_cols: list) -> pd.Series:
X = df[feature_cols].fillna(0.0)
probs = self.model.predict_proba(X)[:, 1]
return pd.Series(probs, index=df.index)
|