feature_engine/training_transforms/ohlcv.py

from typing import Tuple
import numpy as np
import pandas as pd
import math
from pathlib import Path
import warnings

# Force all numpy/pandas runtime warnings to raise an exception instead
warnings.simplefilter("error", RuntimeWarning)


def compute_rsi(series, period=14):
    delta = series.diff()
    gain = delta.clip(lower=0).rolling(period).mean()
    loss = (-delta.clip(upper=0)).rolling(period).mean()
    rs = gain / (loss + 1e-9)
    return 100 - (100 / (1 + rs))


def compute_atr(high, low, close, period=14):
    high_low = high - low
    high_close = (high - close.shift(1)).abs()
    low_close = (low - close.shift(1)).abs()

    true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)

    atr = true_range.ewm(alpha=1 / period, adjust=False).mean()

    return atr


def compute_aroon(high, low, period=25):
    """

    Returns

    -------

    aroon_up : pd.Series

    aroon_down : pd.Series

    """

    days_since_high = high.rolling(period).apply(
        lambda x: period - 1 - np.argmax(x),
        raw=True,
    )

    days_since_low = low.rolling(period).apply(
        lambda x: period - 1 - np.argmin(x),
        raw=True,
    )

    aroon_up = ((period - days_since_high) / period) * 100

    aroon_down = ((period - days_since_low) / period) * 100

    return aroon_up, aroon_down


def compute_parkinson_volatility(high, low, window=20):
    """

    Parkinson volatility estimator.



    Parameters

    ----------

    high : pd.Series

    low : pd.Series

    window : int



    Returns

    -------

    pd.Series

        Rolling Parkinson volatility.

    """
    # Replaces == 0 with a tolerance check
    new_low = np.where(np.isclose(low, 0.0, atol=1e-9), 1e-9, low)
    log_hl_sq = np.log(high / new_low).pow(2)
    return np.sqrt(log_hl_sq.rolling(window).sum() / (4 * window * np.log(2)))


def compute_adx(high, low, close, period=14) -> Tuple[pd.Series, pd.Series, pd.Series]:
    """

    Compute ADX, DI+ and DI- using Wilder's smoothing.



    Parameters

    ----------

    high : pd.Series

    low : pd.Series

    close : pd.Series

    period : int, default=14



    Returns

    -------

    adx : pd.Series

    di_plus : pd.Series

    di_minus : pd.Series

    """

    # ----- Directional Movement -----
    up_move = high.diff()
    down_move = -low.diff()

    plus_dm = pd.Series(
        np.where(
            (up_move > down_move) & (up_move > 0),
            up_move,
            0.0,
        ),
        index=high.index,
    )

    minus_dm = pd.Series(
        np.where(
            (down_move > up_move) & (down_move > 0),
            down_move,
            0.0,
        ),
        index=high.index,
    )

    # ----- True Range -----
    tr = compute_atr(high, low, close, period)

    atr = tr.ewm(alpha=1 / period, adjust=False).mean()

    # ----- Wilder smoothing -----
    plus_dm_smoothed = plus_dm.ewm(
        alpha=1 / period,
        adjust=False,
    ).mean()

    minus_dm_smoothed = minus_dm.ewm(
        alpha=1 / period,
        adjust=False,
    ).mean()

    # ----- Directional Indicators -----
    di_plus = (plus_dm_smoothed / (atr + 1e-9)) * 100
    di_minus = (minus_dm_smoothed / (atr + 1e-9)) * 100

    # ----- Directional Index -----
    dx = ((di_plus - di_minus).abs() / (di_plus + di_minus + 1e-9)) * 100

    # ----- Average Directional Index -----
    adx = dx.ewm(
        alpha=1 / period,
        adjust=False,
    ).mean()

    return adx, di_plus, di_minus


def mean_absolute_deviation(x):
    return np.mean(np.abs(x - np.mean(x)))


def compute_cci(high, low, close, period=20):
    """

    Commodity Channel Index (CCI)

    """

    typical_price = (high + low + close) / 3

    sma = typical_price.rolling(period).mean()

    mean_deviation = typical_price.rolling(period).apply(
        mean_absolute_deviation,
        raw=True,
    )

    cci = (typical_price - sma) / (0.015 * (mean_deviation + 1e-9))

    return cci


def compute_stochastic_k(high, low, close, period=14):
    """

    Stochastic Oscillator %K

    """

    highest_high = high.rolling(period).max()
    lowest_low = low.rolling(period).min()

    stochastic_k = ((close - lowest_low) / (highest_high - lowest_low + 1e-9)) * 100

    return stochastic_k


def compute_macd(close, fast_period=12, slow_period=26, signal_period=9):
    """

    Compute MACD, Signal Line and Histogram.



    Parameters

    ----------

    close : pd.Series

        Closing prices.

    fast_period : int, default=12

    slow_period : int, default=26

    signal_period : int, default=9



    Returns

    -------

    macd : pd.Series

    signal : pd.Series

    histogram : pd.Series

    """

    ema_fast = close.ewm(
        span=fast_period,
        adjust=False,
    ).mean()

    ema_slow = close.ewm(
        span=slow_period,
        adjust=False,
    ).mean()

    macd = ema_fast - ema_slow

    signal = macd.ewm(
        span=signal_period,
        adjust=False,
    ).mean()

    histogram = macd - signal

    return macd, signal, histogram


def compute_bollinger(close, period=20, num_std=2):
    """

    Compute Bollinger Band features.



    Parameters

    ----------

    close : pd.Series

        Closing prices.

    period : int, default=20

        Rolling window for SMA and standard deviation.

    num_std : float, default=2

        Number of standard deviations for the bands.



    Returns

    -------

    bb_width : pd.Series

        Normalized Bollinger Band width.



    bb_position : pd.Series

        Position of the close within the bands.

        0   -> Lower Band

        0.5 -> Middle Band

        1   -> Upper Band



    bb_squeeze : pd.Series

        Width normalized by its rolling mean.

        <1 : Bands tighter than usual.

        >1 : Bands wider than usual.

    """

    middle = close.rolling(period).mean()
    std = close.rolling(period).std()

    upper = middle + num_std * std
    lower = middle - num_std * std

    # Normalized width
    bb_width = (upper - lower) / (middle + 1e-9)

    # Position inside the bands
    bb_position = (close - lower) / (upper - lower + 1e-9)

    # Relative squeeze
    bb_squeeze = bb_width / (bb_width.rolling(period).mean() + 1e-9)

    return bb_width, bb_position, bb_squeeze


import numpy as np
import pandas as pd


def compute_volume_features(

    high,

    low,

    close,

    volume,

    volume_ma_period=20,

    mfi_period=14,

):
    """

    Compute volume-based features.



    Returns

    -------

    volume_ma20

    volume_ratio

    obv

    vwap

    mfi

    """

    # -------------------------------------------------
    # Volume Moving Average
    # -------------------------------------------------

    volume_ma = volume.rolling(volume_ma_period).mean()

    volume_ratio = volume / (volume_ma + 1e-9)

    # -------------------------------------------------
    # OBV
    # -------------------------------------------------

    price_change = close.diff()

    obv = np.sign(price_change).fillna(0).mul(volume).cumsum()

    # -------------------------------------------------
    # VWAP (Cumulative)
    # -------------------------------------------------

    typical_price = (high + low + close) / 3

    vwap = (typical_price * volume).cumsum() / (volume.cumsum() + 1e-9)

    # -------------------------------------------------
    # Money Flow Index (MFI)
    # -------------------------------------------------

    raw_money_flow = typical_price * volume

    positive_flow = raw_money_flow.where(
        typical_price > typical_price.shift(1),
        0.0,
    )

    negative_flow = raw_money_flow.where(
        typical_price < typical_price.shift(1),
        0.0,
    )

    positive_sum = positive_flow.rolling(mfi_period).sum()

    negative_sum = negative_flow.rolling(mfi_period).sum()

    money_ratio = positive_sum / (negative_sum + 1e-9)

    mfi = 100 - (100 / (1 + money_ratio))

    return (
        volume_ma,
        volume_ratio,
        obv,
        vwap,
        mfi,
    )


def compute_candlestick_features(

    open_,

    high,

    low,

    close,

    doji_threshold=0.1,

):
    """

    Compute candlestick-based features.



    Parameters

    ----------

    open_ : pd.Series

    high : pd.Series

    low : pd.Series

    close : pd.Series

    doji_threshold : float, default=0.1

        Maximum body percentage to classify as a Doji.



    Returns

    -------

    body_percent

    upper_shadow_percent

    lower_shadow_percent

    gap_up

    gap_down

    inside_day

    outside_day

    doji

    """

    candle_range = (high - low).replace(0, np.nan)

    # ---------------------------------------------------------
    # Body
    # ---------------------------------------------------------

    body = (close - open_).abs()

    body_percent = body / candle_range

    # ---------------------------------------------------------
    # Upper Shadow
    # ---------------------------------------------------------

    upper_shadow = high - np.maximum(open_, close)

    upper_shadow_percent = upper_shadow / candle_range

    # ---------------------------------------------------------
    # Lower Shadow
    # ---------------------------------------------------------

    lower_shadow = np.minimum(open_, close) - low

    lower_shadow_percent = lower_shadow / candle_range

    # ---------------------------------------------------------
    # Gap Up / Gap Down
    # ---------------------------------------------------------

    previous_high = high.shift(1)
    previous_low = low.shift(1)

    gap_up = (low > previous_high).astype(int)

    gap_down = (high < previous_low).astype(int)

    # ---------------------------------------------------------
    # Inside / Outside Day
    # ---------------------------------------------------------

    inside_day = ((high < previous_high) & (low > previous_low)).astype(int)

    outside_day = ((high > previous_high) & (low < previous_low)).astype(int)

    # ---------------------------------------------------------
    # Doji
    # ---------------------------------------------------------

    doji = (body_percent <= doji_threshold).astype(int)

    return (
        body_percent,
        upper_shadow_percent,
        lower_shadow_percent,
        gap_up,
        gap_down,
        inside_day,
        outside_day,
        doji,
    )


def compute_relative_position(high, low, close, period=252):
    """

    Compute relative position features.



    Parameters

    ----------

    high : pd.Series

    low : pd.Series

    close : pd.Series

    period : int, default=252

        Number of trading days representing one year.



    Returns

    -------

    distance_from_52w_high : pd.Series

    distance_from_52w_low : pd.Series

    rolling_drawdown : pd.Series

    """

    # -------------------------------------------------
    # 52-week High / Low
    # -------------------------------------------------

    rolling_high = high.rolling(period).max()
    rolling_low = low.rolling(period).min()

    distance_from_52w_high = (close - rolling_high) / (rolling_high + 1e-9)

    distance_from_52w_low = (close - rolling_low) / (rolling_low + 1e-9)

    # -------------------------------------------------
    # Rolling Drawdown
    # -------------------------------------------------

    rolling_drawdown = (close - rolling_high) / (rolling_high + 1e-9)

    return (
        distance_from_52w_high,
        distance_from_52w_low,
        rolling_drawdown,
    )


def build_features_ohlcv(ohlcv_df) -> pd.Series:
    """

    this will build all the features we can build from ohlcv.

    Input:  raw OHLCV per stock + market data

    Output: feature matrix, one row per (symbol, date)

    """
    features = []
    symbols = ohlcv_df["symbol"].unique() if "symbol" in ohlcv_df.columns else None

    if symbols is None or len(symbols) == 0:
        print("No tickers were found, exiting")
        return []

    for symbol in symbols:
        df = (
            ohlcv_df[ohlcv_df["symbol"] == symbol].copy() if symbol else ohlcv_df.copy()
        )
        df = df.sort_index()
        # price based features ----------------------------------------------------------
        # log returns
        df["log_ret_1d"] = np.log(df["close"] / df["close"].shift(1))
        df["log_ret_3d"] = np.log(df["close"] / df["close"].shift(3))
        df["log_ret_5d"] = np.log(df["close"] / df["close"].shift(5))
        df["log_ret_10d"] = np.log(df["close"] / df["close"].shift(10))
        df["log_ret_20d"] = np.log(df["close"] / df["close"].shift(20))
        df["log_ret_60d"] = np.log(df["close"] / df["close"].shift(60))

        # simple returns -----------------------------------------------------------------
        for period in [1, 3, 5, 10, 20, 60, 120]:
            df[f"ret_{period}d"] = df["close"].pct_change(period)

        # simple moving average
        df["sma_20d"] = df["close"].rolling(20).mean()
        df["sma_50d"] = df["close"].rolling(50).mean()
        df["sma_200d"] = df["close"].rolling(200).mean()

        # Exponential Moving Average (EMA) 20,50
        df["ema_20d"] = df["close"].ewm(span=20, adjust=False).mean()
        df["ema_50d"] = df["close"].ewm(span=50, adjust=False).mean()

        # price position
        # close_sma20_ratio close_sma50_ratio close_sma200_ratio close_ema20_ratio close_ema50_ratio high_20_position low_20_position
        # distance_from_52w_high distance_from_52w_low
        df["close_sma20_ratio"] = df["close"] / df["sma_20d"]
        df["close_sma50_ratio"] = df["close"] / df["sma_50d"]
        df["close_sma200_ratio"] = df["close"] / df["sma_200d"]
        df["close_ema20_ratio"] = df["close"] / df["ema_20d"]
        df["high_20"] = df["high"].rolling(20).max()
        df["high_20_position"] = df["close"] / (df["high_20"] + 1e-9)
        df["low_20"] = df["low"].rolling(20).min()
        df["low_20_position"] = (df["close"] - df["low_20"]) / (df["low_20"] + 1e-9)
        df["high_52w"] = df["high"].rolling(252).max()
        df["close_to_52w_high"] = (df["close"] - df["high_52w"]) / (
            df["high_52w"] + 1e-9
        )
        df["low_52w"] = df["low"].rolling(252).min()
        df["close_to_52w_low"] = (df["close"] - df["low_52w"]) / (df["low_52w"] + 1e-9)
        df["position_in_20d_range"] = (df["close"] - df["low_20"]) / (
            df["high_20"] - df["low_20"] + 1e-9
        )
        # Momentum — ROC: yes for 10, 20, 60. Momentum 10, 20 yes. PPO yes, APO no (PPO is just normalized APO, keep one)
        # ROC
        for period in [10, 20, 60]:
            df[f"roc_{period}"] = df["close"].pct_change(period) * 100

        # Momentum
        for period in [10, 20]:
            df[f"momentum_{period}"] = df["close"] - df["close"].shift(period)
        # PPO
        df["ppo"] = ((df["ema_20d"] - df["ema_50d"]) / (df["ema_50d"] + 1e-9)) * 100

        # Distance from moving averages (normalized)
        df["dist_sma20"] = (df["close"] - df["sma_20d"]) / df["sma_20d"]
        df["dist_sma50"] = (df["close"] - df["sma_50d"]) / df["sma_50d"]
        df["dist_sma200"] = (df["close"] - df["sma_200d"]) / df["sma_200d"]

        # Volatility
        # Volatility — rolling_std: yes for 10, 20, 60. Skip 5 (noise).
        # ATR(Average True Range) 14 yes, ATR21 no (redundant). ATR_percent yes.
        # Parkinson yes (uses high/low, genuinely different from close-to-close std).
        # True Range no (ATR already captures it)
        for period in [10, 20, 60]:
            df[f"vol_{period}"] = df["ret_1d"].rolling(period).std()

        df["vol_ratio"] = df["vol_10"] / df["vol_20"]  # vol regime

        df["atr_14"] = compute_atr(df["high"], df["low"], df["close"])

        df["atr_percent"] = (df["atr_14"] / (df["close"] + 1e-9)) * 100

        df["parkinson_volatility"] = compute_parkinson_volatility(df["high"], df["low"])

        #  Trend Strength — ADX14 yes, ADX20 no (redundant). DI+ and DI- both yes. Aroon Up and Down both yes.
        df["adx_14"], df["di_plus"], df["di_minus"] = compute_adx(
            df["high"],
            df["low"],
            df["close"],
        )

        df["aroon_up"], df["aroon_down"] = compute_aroon(
            df["high"],
            df["low"],
        )
        # Oscillators — RSI14 yes, RSI7
        # CCI20, Stochastic K
        # RSI
        df["rsi_14"] = compute_rsi(df["close"], 14)
        df["rsi_7"] = compute_rsi(df["close"], 7)
        df["cci_20"] = compute_cci(
            df["high"],
            df["low"],
            df["close"],
        )
        df["stochastic_k"] = compute_stochastic_k(
            df["high"],
            df["low"],
            df["close"],
        )

        #  MACD — MACD yes, Signal yes, Histogram yes (all three — histogram is the most predictive of the three)
        df["macd"], df["macd_signal"], df["macd_histogram"] = compute_macd(df["close"])
        # Bollinger — BB Width yes, BB Position yes, BB Squeeze yes.
        # Skip Upper and Lower raw values (BB Position already captures where price sits, raw levels aren't meaningful cross-sectionally)
        df["bb_width"], df["bb_position"], df["bb_squeeze"] = compute_bollinger(
            df["close"]
        )
        # Range features
        df["intraday_range"] = (df["high"] - df["low"]) / df["close"]
        df["gap"] = (df["open"] - df["close"].shift(1)) / df["close"].shift(1)
        df["close_position"] = (df["close"] - df["low"]) / (
            df["high"] - df["low"] + 1e-9
        )  # 0=low, 1=high

        # Volume — Volume MA20 yes, MA50 no (redundant).
        # Volume Ratio yes. OBV yes.
        # VWAP yes.
        # MFI yes (combines price + volume, genuinely different).
        # Chaikin Money Flow no (redundant with MFI)
        # ── Volume features ───────────────────────────────────
        (
            df["volume_ma20"],
            df["volume_ratio"],
            df["obv"],
            df["vwap"],
            df["mfi"],
        ) = compute_volume_features(
            df["high"],
            df["low"],
            df["close"],
            df["volume"],
        )

        # Candlestick — Body % yes, Upper Shadow % yes, Lower Shadow % yes, Gap Up yes, Gap Down yes,
        # Inside Day yes, Outside Day yes. Doji yes.
        # Skip Hammer, Shooting Star, Bullish/Bearish Engulfing — these are rule-based patterns XGBoost will reconstruct itself from body/shadow/gap features anyway.
        # Adding them explicitly is redundant.
        (
            df["body_percent"],
            df["upper_shadow_percent"],
            df["lower_shadow_percent"],
            df["gap_up"],
            df["gap_down"],
            df["inside_day"],
            df["outside_day"],
            df["doji"],
        ) = compute_candlestick_features(
            df["open"],
            df["high"],
            df["low"],
            df["close"],
        )
        # Relative Position — 52w high yes, 52w low yes, Rolling Max no (52w high covers it), Rolling Min no (same). Rolling Drawdown yes.
        (
            df["distance_from_52w_high"],
            df["distance_from_52w_low"],
            df["rolling_drawdown"],
        ) = compute_relative_position(
            df["high"],
            df["low"],
            df["close"],
        )

        # ── Calendar features ─────────────────────────────────
        # FIX: Direct column-level datetime extraction
        df["day_of_week"] = df["timestamp"].dt.dayofweek
        df["month"] = df["timestamp"].dt.month
        df["is_month_end"] = df["timestamp"].dt.is_month_end.astype(int)

        df["symbol"] = symbol
        features.append(df)

    features_df = pd.concat(features)
    return features_df


def clean_ohlcv(df):
    """

    Apply before any feature engineering.

    """
    original_len = len(df)

    # ── Layer 1: Drop zero/negative prices ───────────────────
    # Any OHLCV value of 0 is invalid
    price_cols = ["open", "high", "low", "close"]
    zero_mask = (df[price_cols] <= 0).any(axis=1)
    df = df[~zero_mask]
    print(f"Dropped {zero_mask.sum()} rows with zero/negative prices")

    # ── Layer 2: Drop impossible OHLC relationships ───────────
    invalid_ohlc = (
        (df["high"] < df["low"])  # high below low
        | (df["high"] < df["close"])  # high below close
        | (df["high"] < df["open"])  # high below open
        | (df["low"] > df["close"])  # low above close
        | (df["low"] > df["open"])  # low above open
    )
    df = df[~invalid_ohlc]
    print(f"Dropped {invalid_ohlc.sum()} rows with invalid OHLC relationships")

    # ── Layer 3: Drop symbols with insufficient history ───────
    # A symbol needs at least 250 rows (≈1 year) for 200d SMA to warm up
    symbol_counts = df.groupby("symbol")["close"].count()
    valid_symbols = symbol_counts[symbol_counts >= 250].index
    dropped_symbols = symbol_counts[symbol_counts < 250].index.tolist()
    df = df[df["symbol"].isin(valid_symbols)]
    print(f"Dropped {len(dropped_symbols)} symbols with < 250 trading days")
    print(
        f"Dropped symbols: {dropped_symbols[:10]}{'...' if len(dropped_symbols) > 10 else ''}"
    )

    print(
        f"\nTotal rows: {original_len:,} → {len(df):,} "
        f"({original_len - len(df):,} removed)"
    )

    return df


def sanity_check(df):
    print(f"Rows: {len(df):,}")
    print(f"Symbols: {df['symbol'].nunique()}")
    print(f"Date range: {df.index.min()} → {df.index.max()}")
    print(f"Null counts:\n{df[['open','high','low','close','volume']].isnull().sum()}")
    print(f"Min close: {df['close'].min()}")
    print(f"Any zero close: {(df['close'] <= 0).any()}")


def identify_bad_df(df):
    bad = (df["close"] <= 0) | (df["close"].shift(1) <= 0)

    print(df.loc[bad, ["symbol", "open", "high", "low", "close"]])


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


def start_save_feature_store():
    ohlcv_df = pd.read_parquet(INPUT_PATH)
    clean_df = clean_ohlcv(ohlcv_df)
    ohlcv_feature_store = build_features_ohlcv(clean_df)
    # ohlcv_feature_store = pd.concat(ohlcv_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
    ohlcv_feature_store.to_parquet(OUTPUT_PATH, index=False)

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


if __name__ == "__main__":
    start_save_feature_store()