File size: 5,408 Bytes
75d9b3c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """
Technical Features — Compute all standard TA indicators using pandas-ta.
Returns a feature DataFrame that can be fed into the signal engine.
"""
import logging
import pandas as pd
import pandas_ta as ta
from config import SIGNAL_PARAMS as SP
logger = logging.getLogger(__name__)
def compute_technical_features(df: pd.DataFrame, ticker: str = "") -> pd.DataFrame:
"""
Compute comprehensive technical indicators for a single ticker's OHLCV DataFrame.
Args:
df: DataFrame with columns Open, High, Low, Close, Volume (DatetimeIndex)
ticker: For logging
Returns:
DataFrame with original OHLCV + all computed features
"""
if df.empty or len(df) < 50:
logger.warning(f"{ticker}: Insufficient data ({len(df)} rows) for technical features")
return df
features = df.copy()
# ── Trend Indicators ──
features[f"sma_{SP['sma_short']}"] = ta.sma(features["Close"], length=SP["sma_short"])
features[f"sma_{SP['sma_medium']}"] = ta.sma(features["Close"], length=SP["sma_medium"])
features[f"sma_{SP['sma_long']}"] = ta.sma(features["Close"], length=SP["sma_long"])
features[f"ema_{SP['ema_fast']}"] = ta.ema(features["Close"], length=SP["ema_fast"])
features[f"ema_{SP['ema_slow']}"] = ta.ema(features["Close"], length=SP["ema_slow"])
# MACD
macd = ta.macd(features["Close"], fast=SP["macd_fast"], slow=SP["macd_slow"], signal=SP["macd_signal"])
if macd is not None:
features = pd.concat([features, macd], axis=1)
# ADX (trend strength)
adx = ta.adx(features["High"], features["Low"], features["Close"], length=14)
if adx is not None:
features = pd.concat([features, adx], axis=1)
# ── Momentum Indicators ──
features["rsi_14"] = ta.rsi(features["Close"], length=14)
features["rsi_7"] = ta.rsi(features["Close"], length=7)
stoch = ta.stoch(features["High"], features["Low"], features["Close"])
if stoch is not None:
features = pd.concat([features, stoch], axis=1)
features["willr"] = ta.willr(features["High"], features["Low"], features["Close"])
features["roc_10"] = ta.roc(features["Close"], length=10)
features["roc_20"] = ta.roc(features["Close"], length=20)
features["cci_20"] = ta.cci(features["High"], features["Low"], features["Close"], length=20)
# ── Volatility Indicators ──
bb = ta.bbands(features["Close"], length=SP["bb_period"], std=SP["bb_std"])
if bb is not None:
features = pd.concat([features, bb], axis=1)
features["atr_14"] = ta.atr(features["High"], features["Low"], features["Close"], length=SP["atr_period"])
features["atr_pct"] = (features["atr_14"] / features["Close"]) * 100 # ATR as % of price
kc = ta.kc(features["High"], features["Low"], features["Close"])
if kc is not None:
features = pd.concat([features, kc], axis=1)
# ── Volume Indicators ──
features["obv"] = ta.obv(features["Close"], features["Volume"])
features[f"vol_sma_{SP['volume_sma_period']}"] = ta.sma(features["Volume"], length=SP["volume_sma_period"])
features["vol_ratio"] = features["Volume"] / features[f"vol_sma_{SP['volume_sma_period']}"]
# ── Price Position Relative to MAs ──
features["price_vs_sma20"] = (features["Close"] / features[f"sma_{SP['sma_short']}"]) - 1
features["price_vs_sma50"] = (features["Close"] / features[f"sma_{SP['sma_medium']}"]) - 1
features["price_vs_sma200"] = (features["Close"] / features[f"sma_{SP['sma_long']}"]) - 1
# ── Crossover Signals (binary) ──
sma_s = features[f"sma_{SP['sma_short']}"]
sma_m = features[f"sma_{SP['sma_medium']}"]
features["golden_cross"] = ((sma_s > sma_m) & (sma_s.shift(1) <= sma_m.shift(1))).astype(int)
features["death_cross"] = ((sma_s < sma_m) & (sma_s.shift(1) >= sma_m.shift(1))).astype(int)
ema_f = features[f"ema_{SP['ema_fast']}"]
ema_sl = features[f"ema_{SP['ema_slow']}"]
features["ema_bullish_cross"] = ((ema_f > ema_sl) & (ema_f.shift(1) <= ema_sl.shift(1))).astype(int)
# ── Support / Resistance (rolling highs/lows) ──
lookback = SP["breakout_lookback"]
features["resistance"] = features["High"].rolling(lookback).max()
features["support"] = features["Low"].rolling(lookback).min()
features["at_resistance"] = (features["Close"] >= features["resistance"] * 0.98).astype(int)
features["at_support"] = (features["Close"] <= features["support"] * 1.02).astype(int)
# ── Bollinger Band Position ──
bb_upper = features.get(f"BBU_{SP['bb_period']}_{SP['bb_std']}")
bb_lower = features.get(f"BBL_{SP['bb_period']}_{SP['bb_std']}")
if bb_upper is not None and bb_lower is not None:
bb_width = bb_upper - bb_lower
features["bb_pct"] = (features["Close"] - bb_lower) / bb_width.replace(0, float("nan"))
features["bb_squeeze"] = (bb_width / features["Close"] * 100) # Width as % of price
logger.info(f"{ticker}: Computed {len(features.columns) - len(df.columns)} technical features")
return features
def get_latest_features(features_df: pd.DataFrame) -> dict:
"""Extract the latest row of features as a flat dict (for signal engine)."""
if features_df.empty:
return {}
last = features_df.iloc[-1]
return {k: (round(float(v), 4) if pd.notna(v) else None) for k, v in last.items()}
|