""" 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