ml-algorithm-performance-analysis-regime detection.py

import os
os.environ["OMP_NUM_THREADS"] = "1"

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from sklearn.dummy import DummyRegressor
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_absolute_error, mean_squared_error

class Student:
    def __init__(self, config: dict | None = None, random_state: int = 42):
        self.random_state = random_state

        # Same as Part 1
        self.n_lags = 5
        self.window_sizes = [3, 5, 10, 20]

        # Regime detection settings
        self.vol_window = 10
        self.n_clusters = 3

        # Objects created during training
        self.scaler = None
        self.kmeans = None
        self.model_ = None

        self.feature_cols_ = []
        self.fitted_ = False

        # Allow overrides
        if isinstance(config, dict):
            for k, v in config.items():
                if hasattr(self, k):
                    setattr(self, k, v)

    # -------------------- Utility Helpers --------------------

    @staticmethod
    def _finite_mean(y):
        yv = pd.Series(y.astype(float)).replace([np.inf, -np.inf], np.nan).dropna()
        return float(yv.mean()) if len(yv) else 0.0

    @staticmethod
    def _log_returns(series):
        return np.log(series / series.shift(1))

    # -------------------- Feature Engineering --------------------

    def _make_features(self, X):
        """
        Build lag + rolling features WITHOUT leakage.
        X must contain a numeric 'Close' column.
        """
        df = X.copy().sort_index()

        if "close" in df.columns:
            df["Close"] = pd.to_numeric(df["close"], errors="coerce")
        elif "Close" in df.columns:
            df["Close"] = pd.to_numeric(df["Close"], errors="coerce")
        else:
            raise ValueError(f"No close column found in dataframe. Columns: {df.columns}")

        lr = self._log_returns(df["Close"])

        feats = {}
        for lag in range(1, self.n_lags + 1):
            feats[f"lag_{lag}"] = lr.shift(lag)

        for w in self.window_sizes:
            r = lr.rolling(w, min_periods=w)
            feats[f"rolling_mean_{w}"] = r.mean()
            feats[f"rolling_std_{w}"] = r.std(ddof=0)

        F = pd.DataFrame(feats, index=df.index).dropna()
        return F, lr

    # -------------------- Regime Detection (Training) --------------------

    def _compute_regimes(self, lr_series):
        df = pd.DataFrame({
            "lr": lr_series,
            "vol": lr_series.rolling(self.vol_window, min_periods=self.vol_window).std()
        }).dropna()

        if df.empty:
            return pd.Series(0, index=lr_series.index)

        self.scaler = StandardScaler()
        scaled = self.scaler.fit_transform(df)

        self.kmeans = KMeans(
            n_clusters=self.n_clusters,
            random_state=self.random_state,
            n_init="auto"
        )

        labels = self.kmeans.fit_predict(scaled)

        regime_full = pd.Series(index=lr_series.index, dtype=float)
        regime_full[df.index] = labels

        return regime_full.bfill().ffill().fillna(0)

    # -------------------- Regime Assignment (Prediction) --------------------

    def _assign_regimes(self, lr_series):
        df = pd.DataFrame({
            "lr": lr_series,
            "vol": lr_series.rolling(self.vol_window, min_periods=self.vol_window).std()
        }).dropna()

        if df.empty or self.scaler is None:
            return pd.Series(0, index=lr_series.index)

        scaled = self.scaler.transform(df)
        labels = self.kmeans.predict(scaled)

        regime_full = pd.Series(index=lr_series.index, dtype=float)
        regime_full[df.index] = labels

        return regime_full.bfill().ffill().fillna(0)

    # -------------------- fit --------------------

    def fit(self, X_train, y_train, meta=None):
        F, lr = self._make_features(X_train)
        y = y_train.reindex(F.index)

        if F.empty or y.isna().all():
            mean_y = self._finite_mean(y_train)
            self.model_ = DummyRegressor(strategy="constant", constant=mean_y)
            self.model_.fit([[0.0]], [0.0])
            self.fitted_ = True
            return self

        regimes = self._compute_regimes(lr)
        F["regime"] = regimes.reindex(F.index)

        self.model_ = RandomForestRegressor(
            n_estimators=200,
            max_depth=3,
            min_samples_split=5,
            random_state=self.random_state,
            n_jobs=-1
        )
        self.model_.fit(F, y)

        self.feature_cols_ = F.columns
        self.fitted_ = True
        return self

    # -------------------- predict --------------------

    def predict(self, X, meta=None):
        F, lr = self._make_features(X)

        if not self.fitted_ or F.empty:
            return pd.Series(0.0, index=X.index, name="y_pred")

        regimes = self._assign_regimes(lr)
        F["regime"] = regimes.reindex(F.index)

        y_hat = self.model_.predict(F[self.feature_cols_])
        return pd.Series(y_hat, index=F.index, name="y_pred")


#main function to evaluate the model

def main():

    DATA_FILE = "data/prices2.csv"
    TICKERS = ["XLK", "XLF", "XLE"]
    START_DATE = "2015-01-01"
    END_DATE = "2019-12-31"
    HORIZON = 5       # next-H-day log return
    STEP = 10         # walk-forward step
    print("Running Student_ext with default settings...\n")
    print(f"Tickers: {TICKERS}")
    print(f"Date range: {START_DATE} → {END_DATE}")
    print(f"Horizon: {HORIZON}, Step: {STEP}\n")

    #Load dataset
    df = pd.read_csv(DATA_FILE)
    df["date"] = pd.to_datetime(df["date"])
    df = df.sort_values(["ticker", "date"])

    for TICKER in TICKERS:
        print(f"\nEvaluating {TICKER}:")

        # Filter ticker + date range
        data = df[df["ticker"] == TICKER].copy()
        data = data[(data["date"] >= START_DATE) & (data["date"] <= END_DATE)]
        data = data.set_index("date")

        # Ensure numeric
        for col in data.columns:
            if col not in ["ticker"]:
                data[col] = pd.to_numeric(data[col], errors="coerce")

        close_col = "close" if "close" in data.columns else "Close"
        data = data.dropna(subset=[close_col])

        # Target = next-H-day return
        data["target"] = np.log(data[close_col].shift(-HORIZON) / data[close_col])
        data = data.dropna()

        # Walk-forward split: 80%
        split = int(len(data) * 0.8)
        train = data.iloc[:split]
        test = data.iloc[split:]

        # Fit + predict
        model = Student()
        model.fit(train, train["target"])
        preds = model.predict(test)

        # Align prediction index
        y_true = test["target"].reindex(preds.index)

        mae = mean_absolute_error(y_true, preds)
        rmse = np.sqrt(mean_squared_error(y_true, preds))
        diracc = (np.sign(preds) == np.sign(y_true)).mean()

        print(f"MAE   : {mae:.6f}")
        print(f"RMSE  : {rmse:.6f}")
        print(f"DirAcc: {diracc:.4f}")


if __name__ == "__main__":
    main()