feature_engine/training_transforms/macro.py

import pandas as pd
import numpy as np
from functools import reduce
from pathlib import Path
from typing import Literal
from functools import reduce

# Tickers mapped to your requested identifier names
TICKER_MAP = {
    "BZ=F": "brent_crude_oil",
    "GC=F": "gold",
    "INR=X": "usd_inr",
}


def compute_zscore(series: pd.Series, period=252):
    mean = series.rolling(period).mean()
    std = series.rolling(period).std()

    return (series - mean) / (std + 1e-9)


def build_macro_features(df, prefix) -> pd.Series:
    """
    Generic — call for gold, brent, usd_inr with their prefix.
    df must have a 'close' column.
    """
    print(f"feature store {prefix} has length of {len(df)}")
    df = df.sort_index().copy()
    p = prefix  # e.g. "gold", "brent", "usd_inr"

    # ── Returns (keep) ────────────────────────────────────────
    df[f"{p}_ret_1d"] = df["close"].pct_change(1)
    df[f"{p}_ret_5d"] = df["close"].pct_change(5)
    df[f"{p}_ret_20d"] = df["close"].pct_change(20)

    # ── Trend (keep, fix momentum to be % not absolute) ───────
    df[f"{p}_sma_20"] = df["close"].rolling(20).mean()
    df[f"{p}_sma_50"] = df["close"].rolling(50).mean()
    df[f"{p}_sma_200"] = df["close"].rolling(200).mean()

    df[f"{p}_sma20_ratio"] = df["close"] / (df[f"{p}_sma_20"] + 1e-9) - 1
    df[f"{p}_sma200_ratio"] = df["close"] / (df[f"{p}_sma_200"] + 1e-9) - 1

    # Momentum as % not absolute (absolute is not cross-sectionally comparable)
    df[f"{p}_momentum_10"] = df["close"].pct_change(10)
    df[f"{p}_momentum_20"] = df["close"].pct_change(20)

    # ── Volatility (keep) ─────────────────────────────────────
    df[f"{p}_volatility_20d"] = df[f"{p}_ret_1d"].rolling(20).std()

    # ── ADD: Trend direction ──────────────────────────────────
    # Is asset in uptrend? Model knows level but not direction
    df[f"{p}_trend"] = (df[f"{p}_sma_20"] > df[f"{p}_sma_50"]).astype("int8")

    # ── ADD: Z-score (regime context) ────────────────────────
    df[f"{p}_zscore"] = compute_zscore(df["close"])

    # ── ADD: Volatility regime ────────────────────────────────
    df[f"{p}_vol_zscore"] = compute_zscore(df[f"{p}_volatility_20d"])
    df[f"{p}_high_vol"] = (df[f"{p}_vol_zscore"] > 1).astype("int8")

    return pd.concat([df])


MergeHow = Literal["left", "right", "outer", "inner", "cross"]


def merge_feature_frames_asof(
    dfs: dict[str, pd.DataFrame],
    timestamp_col: str = "timestamp",
    tolerance: str = "2D",
    prefix_columns: bool = True,
) -> pd.DataFrame:
    """
    Merge feature frames by NEAREST timestamp within `tolerance`, not exact match.
    Use this when sources may label the same trading day under slightly
    different timestamps (common when mixing futures and FX data).
    """
    prepped = {}
    for name, df in dfs.items():
        df = df.copy()
        df[timestamp_col] = pd.to_datetime(df[timestamp_col], utc=True).dt.tz_convert(
            None
        )
        df = df.sort_values(timestamp_col).reset_index(drop=True)

        dupes = df[timestamp_col].duplicated().sum()
        if dupes:
            print(f"WARNING: {name} has {dupes} duplicate timestamps")

        if prefix_columns:
            df = df.rename(
                columns={c: f"{name}_{c}" for c in df.columns if c != timestamp_col}
            )

        prepped[name] = df
        print(
            f"{name}: {df[timestamp_col].min().date()} -> {df[timestamp_col].max().date()} ({len(df)} rows)"
        )

    def _merge(left, right):
        return pd.merge_asof(
            left,
            right,
            on=timestamp_col,
            direction="nearest",
            tolerance=pd.Timedelta(tolerance),
        )

    merged = reduce(_merge, prepped.values())
    merged = merged.sort_values(timestamp_col).reset_index(drop=True)

    overlap_pct = merged.drop(columns=[timestamp_col]).notna().all(axis=1).mean() * 100
    print(
        f"\nmerged: {merged[timestamp_col].min().date()} -> {merged[timestamp_col].max().date()} "
        f"({len(merged)} rows, {overlap_pct:.1f}% fully-populated rows)"
    )

    return merged


def normalize_timestamp(df: pd.DataFrame, col: str = "timestamp") -> pd.DataFrame:
    df = df.copy()
    ts = pd.to_datetime(df[col], utc=True)
    df[col] = ts.dt.tz_convert(None).dt.normalize()
    return df


def feature_building_macro(macro_df: pd.Series):
    gold_df = (
        macro_df[macro_df["feature_type"] == TICKER_MAP["GC=F"]]
        .sort_values("timestamp")
        .reset_index(drop=True)
    )
    brent_df = (
        macro_df[macro_df["feature_type"] == TICKER_MAP["BZ=F"]]
        .sort_values("timestamp")
        .reset_index(drop=True)
    )
    usdinr_df = (
        macro_df[macro_df["feature_type"] == TICKER_MAP["INR=X"]]
        .sort_values("timestamp")
        .reset_index(drop=True)
    )

    gold_features = build_macro_features(gold_df, "gold")
    brent_features = build_macro_features(brent_df, "brent")
    usdinr_features = build_macro_features(usdinr_df, "usd_inr")

    gold_features = normalize_timestamp(gold_features)
    brent_features = normalize_timestamp(brent_features)
    usdinr_features = normalize_timestamp(usdinr_features)

    print(gold_features["timestamp"].dtype)
    print(gold_features["timestamp"].head(3).tolist())

    print(brent_features["timestamp"].dtype)
    print(brent_features["timestamp"].head(3).tolist())

    print(usdinr_features["timestamp"].dtype)
    print(usdinr_features["timestamp"].head(3).tolist())

    gold_dates = set(gold_features["timestamp"])
    brent_dates = set(brent_features["timestamp"])
    usdinr_dates = set(usdinr_features["timestamp"])

    print("gold ∩ brent ∩ usdinr:", len(gold_dates & brent_dates & usdinr_dates))
    print("gold ∩ brent:", len(gold_dates & brent_dates))
    print("gold ∩ usdinr:", len(gold_dates & usdinr_dates))
    print("brent ∩ usdinr:", len(brent_dates & usdinr_dates))

    for name, df in [
        ("gold", gold_features),
        ("brent", brent_features),
        ("usdinr", usdinr_features),
    ]:
        ts = pd.to_datetime(df["timestamp"], utc=True).dt.tz_convert(None)
        diffs = ts.sort_values().diff().dropna()
        print(f"\n{name}:")
        print(f"  rows: {len(df)}")
        print(f"  date range: {ts.min().date()} -> {ts.max().date()}")
        print(f"  gap distribution (days between consecutive rows):")
        print(diffs.dt.days.value_counts().sort_index().head(10))

    # USD/INR only — currency regime matters differently
    usdinr_features["usd_inr_appreciation"] = (
        usdinr_features["usd_inr_ret_5d"] > 0
    ).astype(
        "int8"
    )  # rupee weakening = capital outflow risk for equities

    # Brent only — supply shock detection
    brent_features["brent_spike"] = (
        brent_features["brent_ret_1d"].abs() > 0.03
    ).astype(
        "int8"
    )  # >3% single day move = supply shock event

    # merged = pd.concat(
    #     [gold_features, brent_features, usdinr_features], ignore_index=True
    # )
    # print(f"before gold has following columns {gold_features.columns}")
    gold_features = gold_features.drop(
        columns=["open", "high", "low", "close", "volume", "feature_type"]
    )
    # print(f"after gold has following columns {gold_features.columns}")

    # print(f"before brent has following columns {brent_features.columns}")
    brent_features = brent_features.drop(
        columns=["open", "high", "low", "close", "volume", "feature_type"]
    )
    # print(f"after brent has following columns {brent_features.columns}")

    # print(f"before usdinr has following columns {usdinr_features.columns}")
    usdinr_features = usdinr_features.drop(
        columns=["open", "high", "low", "close", "volume", "feature_type"]
    )
    # print(f"after usdinr has following columns {usdinr_features.columns}")

    # usage with your actual date ranges
    merged = merge_feature_frames_asof(
        {
            "gold": gold_features,  # 2000-08-30 -> 2026-06-29
            "brent": brent_features,  # 2007-07-31 -> 2026-06-30
            "usdinr": usdinr_features,  # 2003-12-02 -> 2026-07-01
        },
        "timestamp",
        "2D",
        False,
    )

    merged = merged.sort_values("timestamp").reset_index(drop=True)

    return pd.concat([merged])


# input path for OHLCV
INPUT_PATH = Path(
    "H:/Developer/stock_model/Dataset/Processed_dataset/feature_stores/macro_store.parquet"
)
# output path
OUTPUT_PATH = Path(
    "H:/Developer/stock_model/Dataset/Processed_dataset/feature_stores/macro_feature_store.parquet"
)


def start_save_feature_store():
    from OHLCV_to_store import get_start_end_date_for_df
    from feature_building_vix import clean_nifty, identify_bad_df

    ohlcv_df = pd.read_parquet(INPUT_PATH)
    clean_df = clean_nifty(ohlcv_df)
    identify_bad_df(clean_df)
    macro_feature_store = feature_building_macro(clean_df)
    macro_feature_store = pd.concat([macro_feature_store], ignore_index=True)
    # save the new store
    # Ensure destination tracking directory paths exist natively
    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)

    # Export out to structural Parquet architecture
    macro_feature_store.to_parquet(OUTPUT_PATH, index=False)

    print(f"\n✅ Success! Macro Feature Store created at: {OUTPUT_PATH}")
    print(f"Total rows recorded: {len(macro_feature_store)}")
    print(
        f"Timeline Range: {macro_feature_store['timestamp'].min().strftime('%Y-%m-%d')} to {macro_feature_store['timestamp'].max().strftime('%Y-%m-%d')}"
    )


if __name__ == "__main__":
    start_save_feature_store()