feature_engine/engine.py

"""Build and read model features from the Parquet feature layer."""

from __future__ import annotations

import json
import pandas as pd

from app.config import settings
from database.parquet_store import ParquetFeatureStore
from .correlations import add_correlation_features
from .us_market import build_us_wide


class FeatureEngine:
    def __init__(self):
        with settings.METADATA_PATH.open() as handle:
            self.required_features = set(json.load(handle)["feature_cols"])

    @staticmethod
    def _read_store(store: ParquetFeatureStore, name: str) -> pd.DataFrame:
        frame = store.read(name)
        if "timestamp" in frame:
            frame["timestamp"] = pd.to_datetime(frame["timestamp"])
        return frame

    def _merge(self, ohlcv_df, nifty_df, vix_df, macro_df, us_features_df) -> pd.DataFrame:
        """Attach only information available on or before each NSE session."""
        us_wide = build_us_wide(us_features_df)
        base = ohlcv_df.sort_values(["timestamp", "symbol"]).copy()

        def attach(left, right, columns):
            if not columns:
                return left
            right = right[["timestamp", *columns]].sort_values("timestamp")
            return pd.merge_asof(left.sort_values("timestamp"), right, on="timestamp", direction="backward")

        nifty_cols = [c for c in nifty_df if c != "timestamp" and f"nifty_{c}" in self.required_features]
        nifty = nifty_df.rename(columns={c: f"nifty_{c}" for c in nifty_cols})
        base = attach(base, nifty, [f"nifty_{c}" for c in nifty_cols])

        vix_cols = [c for c in vix_df if c != "timestamp" and c in self.required_features]
        if "close_1" in self.required_features and "close" in vix_df:
            vix_df = vix_df.rename(columns={"close": "close_1"})
            vix_cols.append("close_1")
        base = attach(base, vix_df, list(dict.fromkeys(vix_cols)))
        base = attach(base, macro_df, [c for c in macro_df if c != "timestamp" and c in self.required_features])
        base = attach(base, us_wide, [c for c in us_wide if c != "timestamp" and c in self.required_features])
        return add_correlation_features(base.sort_values(["symbol", "timestamp"]).reset_index(drop=True))

    def build_all_features(self) -> pd.DataFrame:
        store = ParquetFeatureStore(settings.DATABASE_URL)
        frames = [self._read_store(store, name) for name in (
                settings.OHLCV_FEATURES_TABLE, settings.NIFTY_FEATURES_TABLE,
                settings.VIX_FEATURES_TABLE, settings.MACRO_FEATURES_TABLE,
                settings.US_FEATURES_TABLE,
        )]
        return self._merge(*frames)

    def build_features_for_date(self, prediction_date: pd.Timestamp) -> pd.DataFrame:
        prediction_date = pd.Timestamp(prediction_date).normalize()
        df = self._read_store(ParquetFeatureStore(settings.DATABASE_URL), "merged_features")
        df = df[df["timestamp"].dt.normalize().eq(prediction_date)].copy()
        missing = sorted(self.required_features - set(df.columns))
        if missing:
            raise RuntimeError(f"Feature-store merge is missing model columns: {missing}")
        return df