""" Technical indicator library — pure pandas/numpy, no TA-Lib dependency. (TA-Lib is a pain to install on most systems; this keeps the bot portable. Swap in TA-Lib later if you want C-speed on huge datasets.) Input convention: a pandas DataFrame `df` with columns: open, high, low, close, volume indexed by datetime, ascending order (oldest first). Every function returns a pandas Series (or DataFrame for multi-line indicators) aligned to df.index. """ import numpy as np import pandas as pd from config import settings as cfg # --------------------------------------------------------------------------- # Trend indicators # --------------------------------------------------------------------------- def ema(series: pd.Series, period: int) -> pd.Series: return series.ewm(span=period, adjust=False).mean() def sma(series: pd.Series, period: int) -> pd.Series: return series.rolling(period).mean() def vwap(df: pd.DataFrame) -> pd.Series: typical = (df["high"] + df["low"] + df["close"]) / 3 cum_vol = df["volume"].cumsum() cum_vol_price = (typical * df["volume"]).cumsum() return cum_vol_price / cum_vol.replace(0, np.nan) def macd(series: pd.Series, fast=None, slow=None, signal=None) -> pd.DataFrame: fast = fast or cfg.MACD_FAST slow = slow or cfg.MACD_SLOW signal = signal or cfg.MACD_SIGNAL ema_fast = ema(series, fast) ema_slow = ema(series, slow) macd_line = ema_fast - ema_slow signal_line = ema(macd_line, signal) hist = macd_line - signal_line return pd.DataFrame({"macd": macd_line, "signal": signal_line, "hist": hist}) def adx(df: pd.DataFrame, period: int = None) -> pd.DataFrame: period = period or cfg.ADX_PERIOD high, low, close = df["high"], df["low"], df["close"] up_move = high.diff() down_move = -low.diff() plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0.0) minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0.0) tr = _true_range(df) atr_ = tr.ewm(alpha=1 / period, adjust=False).mean() plus_di = 100 * pd.Series(plus_dm, index=df.index).ewm(alpha=1 / period, adjust=False).mean() / atr_.replace(0, np.nan) minus_di = 100 * pd.Series(minus_dm, index=df.index).ewm(alpha=1 / period, adjust=False).mean() / atr_.replace(0, np.nan) dx = (abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, np.nan)) * 100 adx_ = dx.ewm(alpha=1 / period, adjust=False).mean() return pd.DataFrame({"plus_di": plus_di, "minus_di": minus_di, "adx": adx_}) def ichimoku(df: pd.DataFrame) -> pd.DataFrame: high, low, close = df["high"], df["low"], df["close"] tenkan = (high.rolling(9).max() + low.rolling(9).min()) / 2 kijun = (high.rolling(26).max() + low.rolling(26).min()) / 2 senkou_a = ((tenkan + kijun) / 2).shift(26) senkou_b = ((high.rolling(52).max() + low.rolling(52).min()) / 2).shift(26) chikou = close.shift(-26) return pd.DataFrame({ "tenkan_sen": tenkan, "kijun_sen": kijun, "senkou_span_a": senkou_a, "senkou_span_b": senkou_b, "chikou_span": chikou, }) def supertrend(df: pd.DataFrame, period: int = None, multiplier: float = None) -> pd.DataFrame: period = period or cfg.SUPERTREND_PERIOD multiplier = multiplier or cfg.SUPERTREND_MULTIPLIER atr_ = atr(df, period) hl2 = (df["high"] + df["low"]) / 2 upperband = hl2 + multiplier * atr_ lowerband = hl2 - multiplier * atr_ final_upper = upperband.copy() final_lower = lowerband.copy() trend = pd.Series(index=df.index, dtype="int64") trend.iloc[0] = 1 close = df["close"] for i in range(1, len(df)): if close.iloc[i - 1] <= final_upper.iloc[i - 1]: final_upper.iloc[i] = min(upperband.iloc[i], final_upper.iloc[i - 1]) else: final_upper.iloc[i] = upperband.iloc[i] if close.iloc[i - 1] >= final_lower.iloc[i - 1]: final_lower.iloc[i] = max(lowerband.iloc[i], final_lower.iloc[i - 1]) else: final_lower.iloc[i] = lowerband.iloc[i] if close.iloc[i] > final_upper.iloc[i - 1]: trend.iloc[i] = 1 elif close.iloc[i] < final_lower.iloc[i - 1]: trend.iloc[i] = -1 else: trend.iloc[i] = trend.iloc[i - 1] supertrend_line = pd.Series( np.where(trend == 1, final_lower, final_upper), index=df.index ) return pd.DataFrame({"supertrend": supertrend_line, "trend": trend}) # --------------------------------------------------------------------------- # Momentum / oscillators # --------------------------------------------------------------------------- def rsi(series: pd.Series, period: int = None) -> pd.Series: period = period or cfg.RSI_PERIOD delta = series.diff() gain = delta.clip(lower=0) loss = -delta.clip(upper=0) avg_gain = gain.ewm(alpha=1 / period, adjust=False).mean() avg_loss = loss.ewm(alpha=1 / period, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) return 100 - (100 / (1 + rs)) def stoch_rsi(series: pd.Series, period: int = None) -> pd.Series: period = period or cfg.STOCH_RSI_PERIOD rsi_ = rsi(series, period) min_rsi = rsi_.rolling(period).min() max_rsi = rsi_.rolling(period).max() return (rsi_ - min_rsi) / (max_rsi - min_rsi).replace(0, np.nan) def cci(df: pd.DataFrame, period: int = None) -> pd.Series: period = period or cfg.CCI_PERIOD typical = (df["high"] + df["low"] + df["close"]) / 3 sma_tp = typical.rolling(period).mean() mean_dev = typical.rolling(period).apply(lambda x: np.mean(np.abs(x - x.mean())), raw=True) return (typical - sma_tp) / (0.015 * mean_dev.replace(0, np.nan)) def roc(series: pd.Series, period: int = None) -> pd.Series: period = period or cfg.ROC_PERIOD return (series - series.shift(period)) / series.shift(period).replace(0, np.nan) * 100 def momentum(series: pd.Series, period: int = None) -> pd.Series: period = period or cfg.MOMENTUM_PERIOD return series - series.shift(period) # --------------------------------------------------------------------------- # Volume indicators # --------------------------------------------------------------------------- def obv(df: pd.DataFrame) -> pd.Series: direction = np.sign(df["close"].diff().fillna(0)) return (direction * df["volume"]).cumsum() def cmf(df: pd.DataFrame, period: int = None) -> pd.Series: period = period or cfg.CMF_PERIOD mfm = ((df["close"] - df["low"]) - (df["high"] - df["close"])) / (df["high"] - df["low"]).replace(0, np.nan) mfv = mfm * df["volume"] return mfv.rolling(period).sum() / df["volume"].rolling(period).sum().replace(0, np.nan) def volume_profile(df: pd.DataFrame, bins: int = 24) -> pd.DataFrame: """Simple volume-by-price histogram over the given window.""" price_bins = pd.cut(df["close"], bins=bins) profile = df.groupby(price_bins, observed=True)["volume"].sum().sort_index() poc_bin = profile.idxmax() if not profile.empty else None return pd.DataFrame({"volume": profile}), poc_bin # POC = Point of Control # --------------------------------------------------------------------------- # Volatility / bands # --------------------------------------------------------------------------- def _true_range(df: pd.DataFrame) -> pd.Series: prev_close = df["close"].shift(1) tr = pd.concat([ df["high"] - df["low"], (df["high"] - prev_close).abs(), (df["low"] - prev_close).abs(), ], axis=1).max(axis=1) return tr def atr(df: pd.DataFrame, period: int = None) -> pd.Series: period = period or cfg.ATR_PERIOD return _true_range(df).ewm(alpha=1 / period, adjust=False).mean() def bollinger_bands(series: pd.Series, period: int = None, num_std: float = None) -> pd.DataFrame: period = period or cfg.BOLLINGER_PERIOD num_std = num_std or cfg.BOLLINGER_STD mid = sma(series, period) std = series.rolling(period).std() return pd.DataFrame({ "upper": mid + num_std * std, "mid": mid, "lower": mid - num_std * std, }) def donchian_channel(df: pd.DataFrame, period: int = None) -> pd.DataFrame: period = period or cfg.DONCHIAN_PERIOD upper = df["high"].rolling(period).max() lower = df["low"].rolling(period).min() return pd.DataFrame({"upper": upper, "mid": (upper + lower) / 2, "lower": lower}) # --------------------------------------------------------------------------- # Levels # --------------------------------------------------------------------------- def pivot_points(prev_high, prev_low, prev_close) -> dict: """Classic floor-trader pivots from the previous day's OHLC.""" pivot = (prev_high + prev_low + prev_close) / 3 r1 = 2 * pivot - prev_low s1 = 2 * pivot - prev_high r2 = pivot + (prev_high - prev_low) s2 = pivot - (prev_high - prev_low) r3 = prev_high + 2 * (pivot - prev_low) s3 = prev_low - 2 * (prev_high - pivot) return {"pivot": pivot, "r1": r1, "r2": r2, "r3": r3, "s1": s1, "s2": s2, "s3": s3} def fibonacci_retracement(swing_high: float, swing_low: float) -> dict: diff = swing_high - swing_low levels = [0, 0.236, 0.382, 0.5, 0.618, 0.786, 1] return {f"fib_{lvl}": swing_high - diff * lvl for lvl in levels} # --------------------------------------------------------------------------- # Convenience: compute everything at once # --------------------------------------------------------------------------- def compute_all(df: pd.DataFrame) -> dict: """Returns a dict of {name: Series/DataFrame} for every indicator above, computed on the given OHLCV dataframe. Used by the confidence engine.""" close = df["close"] out = {} for p in cfg.EMA_PERIODS: out[f"ema_{p}"] = ema(close, p) for p in cfg.SMA_PERIODS: out[f"sma_{p}"] = sma(close, p) out["vwap"] = vwap(df) out["macd"] = macd(close) out["adx"] = adx(df) out["ichimoku"] = ichimoku(df) out["supertrend"] = supertrend(df) out["rsi"] = rsi(close) out["stoch_rsi"] = stoch_rsi(close) out["cci"] = cci(df) out["roc"] = roc(close) out["momentum"] = momentum(close) out["obv"] = obv(df) out["cmf"] = cmf(df) out["atr"] = atr(df) out["bollinger"] = bollinger_bands(close) out["donchian"] = donchian_channel(df) return out