core/features.py

"""
Feature extraction from 1-min OHLCV DataFrames.

Two pipelines:
  - semantic:   Hand-crafted morning indicators (gap, VWAP deviation, momentum, etc.)
  - sequential: Raw normalised return and volume vectors for the 09:15-09:30 window.

Both return (X, y, metadata) where:
  X        : pd.DataFrame of features, indexed by date
  y        : pd.Series of binary labels (1 = price up 09:30->15:10)
  metadata : dict[date] -> { c_0930, h_0930, l_0930, v_0930, c_1510, h_1510, l_1510 }
"""

import numpy as np
import pandas as pd


# ── Helpers ──────────────────────────────────────────────────────────────────

def _safe_fill(arr):
    """Forward-fill then back-fill NaNs in a 1-D array."""
    return pd.Series(arr).ffill().bfill().values


# ── Semantic features ────────────────────────────────────────────────────────

def extract_semantic_features(df):
    """
    Compute high-level morning indicators from the 09:15-09:30 candle window.

    Features: gap, return_16m, return_first_5m, return_next_11m, std_dev,
              range_pct, upper_shadow, lower_shadow, total_vol, vwap_dev,
              price_momentum, vol_momentum, morning_trend.
    """
    df = df.copy()
    df["time"]     = df.index.time
    df["date_only"] = df.index.date
    required_times = (
        pd.date_range("09:15", "09:30", freq="min").time.tolist()
        + [pd.to_datetime("15:10").time()]
    )
    df = df[~df.index.duplicated(keep="first")]

    daily_close      = df.groupby("date_only")["close"].last()
    prev_daily_close = daily_close.shift(1)

    # Compute morning session dip/peak (09:31 to 12:00) for limit-order entries
    time_0931 = pd.to_datetime("09:31").time()
    time_1200 = pd.to_datetime("12:00").time()
    df_morning = df[(df["time"] >= time_0931) & (df["time"] <= time_1200)]
    morning_low_per_date  = df_morning.groupby("date_only")["low"].min()
    morning_high_per_date = df_morning.groupby("date_only")["high"].max()

    df_filtered = df[df["time"].isin(required_times)].copy()
    pivot_close = df_filtered.pivot(index="date_only", columns="time", values="close")
    pivot_open  = df_filtered.pivot(index="date_only", columns="time", values="open")
    pivot_high  = df_filtered.pivot(index="date_only", columns="time", values="high")
    pivot_low   = df_filtered.pivot(index="date_only", columns="time", values="low")
    pivot_vol   = df_filtered.pivot(index="date_only", columns="time", values="volume")

    time_0930 = pd.to_datetime("09:30").time()
    time_1510 = pd.to_datetime("15:10").time()

    if time_0930 not in pivot_close.columns:
        return None, None, None

    pivot_close = pivot_close.dropna(subset=[time_0930])
    valid_dates = pivot_close.index
    times_15m   = pd.date_range("09:15", "09:30", freq="min").time

    feature_dicts = []
    metadata      = {}

    for date in valid_dates:
        f = {}
        c_series = _safe_fill(pivot_close.loc[date, times_15m].values.astype(float))
        o_series = _safe_fill(pivot_open.loc[date, times_15m].values.astype(float))
        h_series = _safe_fill(pivot_high.loc[date, times_15m].values.astype(float))
        l_series = _safe_fill(pivot_low.loc[date, times_15m].values.astype(float))
        v_series = pd.Series(pivot_vol.loc[date, times_15m].values.astype(float)).fillna(0).values

        o_915 = o_series[0]
        c_930 = c_series[-1]
        c_919 = c_series[4] if len(c_series) > 4 else c_series[-1]

        pdc = prev_daily_close.get(date, np.nan)
        f["gap"]             = 0 if (pd.isna(pdc) or pdc == 0) else (o_915 / pdc) - 1.0
        f["return_16m"]      = (c_930 / o_915) - 1.0 if o_915 != 0 else 0
        f["return_first_5m"] = (c_919 / o_915) - 1.0 if o_915 != 0 else 0
        f["return_next_11m"] = (c_930 / c_919) - 1.0 if c_919 != 0 else 0
        f["std_dev"]         = np.std(c_series / (o_915 + 1e-8))

        max_h = np.max(h_series)
        min_l = np.min(l_series)
        f["range_pct"]       = (max_h - min_l) / (o_915 + 1e-8)
        f["upper_shadow"]    = (max_h - max(o_915, c_930)) / (o_915 + 1e-8)
        f["lower_shadow"]    = (min(o_915, c_930) - min_l) / (o_915 + 1e-8)
        f["total_vol"]       = np.sum(v_series)

        vwap = np.sum(((h_series + l_series + c_series) / 3.0) * v_series) / (np.sum(v_series) + 1e-8)
        f["vwap_dev"]        = (c_930 / vwap) - 1.0 if vwap != 0 else 0
        f["price_momentum"]  = (c_series[-1] - c_series[-3]) / (c_series[-3] + 1e-8)
        f["vol_momentum"]    = (v_series[-1] - v_series[-3]) / (v_series[-3] + 1e-8)
        f["morning_trend"]   = np.polyfit(np.arange(len(c_series)), c_series, 1)[0]

        feature_dicts.append(f)

        metadata[date] = {
            "c_0930": c_930,
            "h_0930": float(pivot_high.loc[date, time_0930]) if time_0930 in pivot_high.columns else c_930,
            "l_0930": float(pivot_low.loc[date, time_0930])  if time_0930 in pivot_low.columns  else c_930,
            "v_0930": float(pivot_vol.loc[date, time_0930])  if time_0930 in pivot_vol.columns  else 0,
            "c_1510": float(pivot_close.loc[date, time_1510]) if time_1510 in pivot_close.columns else c_930,
            "h_1510": float(pivot_high.loc[date, time_1510])  if time_1510 in pivot_high.columns else c_930,
            "l_1510": float(pivot_low.loc[date, time_1510])   if time_1510 in pivot_low.columns  else c_930,
            "dip_low":   float(morning_low_per_date.get(date, c_930)),
            "peak_high": float(morning_high_per_date.get(date, c_930)),
        }

    X = pd.DataFrame(feature_dicts, index=valid_dates).fillna(0)

    if time_1510 in pivot_close.columns:
        target = (pivot_close[time_1510] > pivot_close[time_0930]).astype(int)
    else:
        target = pd.Series(0, index=valid_dates)

    return X, target, metadata


# ── Sequential features ──────────────────────────────────────────────────────

def extract_sequential_features(df):
    """
    Compute normalised return and raw volume vectors for each minute
    in the 09:15-09:30 window, relative to the 09:30 close.
    """
    df = df.copy()
    df["time"]      = df.index.time
    df["date_only"] = df.index.date
    required_times = (
        pd.date_range("09:15", "09:30", freq="min").time.tolist()
        + [pd.to_datetime("15:10").time()]
    )
    df = df[~df.index.duplicated(keep="first")]

    # Compute morning session dip/peak (09:31 to 12:00) for limit-order entries
    time_0931 = pd.to_datetime("09:31").time()
    time_1200 = pd.to_datetime("12:00").time()
    df_morning = df[(df["time"] >= time_0931) & (df["time"] <= time_1200)]
    morning_low_per_date  = df_morning.groupby("date_only")["low"].min()
    morning_high_per_date = df_morning.groupby("date_only")["high"].max()

    df_filtered = df[df["time"].isin(required_times)].copy()
    pivot_close = df_filtered.pivot(index="date_only", columns="time", values="close")
    pivot_high  = df_filtered.pivot(index="date_only", columns="time", values="high")
    pivot_low   = df_filtered.pivot(index="date_only", columns="time", values="low")
    pivot_vol   = df_filtered.pivot(index="date_only", columns="time", values="volume")

    time_0930 = pd.to_datetime("09:30").time()
    time_1510 = pd.to_datetime("15:10").time()

    if time_0930 not in pivot_close.columns:
        return None, None, None

    pivot_close = pivot_close.dropna(subset=[time_0930])
    valid_dates = pivot_close.index
    times_15m   = pd.date_range("09:15", "09:30", freq="min").time

    feature_dicts = []
    metadata      = {}

    for date in valid_dates:
        f = {}
        c_series = _safe_fill(pivot_close.loc[date, times_15m].values.astype(float))
        v_series = pd.Series(pivot_vol.loc[date, times_15m].values.astype(float)).fillna(0).values
        c_ref    = c_series[-1]

        for i, t in enumerate(times_15m):
            f[f"ret_c_{i}"]   = (c_series[i] / (c_ref + 1e-8)) - 1.0
            f[f"raw_vol_{i}"] = v_series[i]
        feature_dicts.append(f)

        metadata[date] = {
            "c_0930": c_ref,
            "h_0930": float(pivot_high.loc[date, time_0930]) if time_0930 in pivot_high.columns else c_ref,
            "l_0930": float(pivot_low.loc[date, time_0930])  if time_0930 in pivot_low.columns  else c_ref,
            "v_0930": float(pivot_vol.loc[date, time_0930])  if time_0930 in pivot_vol.columns  else 0,
            "c_1510": float(pivot_close.loc[date, time_1510]) if time_1510 in pivot_close.columns else c_ref,
            "h_1510": float(pivot_high.loc[date, time_1510])  if time_1510 in pivot_high.columns else c_ref,
            "l_1510": float(pivot_low.loc[date, time_1510])   if time_1510 in pivot_low.columns  else c_ref,
            "dip_low":   float(morning_low_per_date.get(date, c_ref)),
            "peak_high": float(morning_high_per_date.get(date, c_ref)),
        }

    X = pd.DataFrame(feature_dicts, index=valid_dates).fillna(0)

    if time_1510 in pivot_close.columns:
        target = (pivot_close[time_1510] > pivot_close[time_0930]).astype(int)
    else:
        target = pd.Series(0, index=valid_dates)

    return X, target, metadata