Signals & Screeners — From Data to Alpha

Python — RSI Implementation (Wilder's Smoothing) import numpy as np import pandas as pd def rsi(close: pd.Series, period: int = 14) -> pd.Series: """Relative Strength Index using Wilder's smoothing. This matches TradingView's RSI exactly (not SMA-based). """ delta = close.diff() gain = delta.where(delta > 0, 0.0) loss = (-delta).where(delta < 0, 0.0) # Wilder's smoothing (exponential with alpha = 1/period) avg_gain = gain.ewm(alpha=1/period, min_periods=period).mean() avg_loss = loss.ewm(alpha=1/period, min_periods=period).mean() rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) def rsi_signal(close: pd.Series, period: int = 14) -> pd.Series: """Generate buy/sell signals from RSI.""" r = rsi(close, period) signal = pd.Series(0, index=close.index) signal[r < 30] = 1 # Oversold = buy signal signal[r > 70] = -1 # Overbought = sell signal return signal

Python — MACD with Signal Line and Histogram def macd( close: pd.Series, fast: int = 12, slow: int = 26, signal_period: int = 9 ) -> pd.DataFrame: """MACD line, signal line, and histogram.""" ema_fast = close.ewm(span=fast, adjust=False).mean() ema_slow = close.ewm(span=slow, adjust=False).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=signal_period, adjust=False).mean() histogram = macd_line - signal_line return pd.DataFrame({ "macd": macd_line, "signal": signal_line, "histogram": histogram, }) def macd_crossover(close: pd.Series) -> pd.Series: """Detect MACD crossover signals.""" m = macd(close) prev_hist = m["histogram"].shift(1) curr_hist = m["histogram"] signal = pd.Series(0, index=close.index) signal[(prev_hist < 0) & (curr_hist > 0)] = 1 # Bullish cross signal[(prev_hist > 0) & (curr_hist < 0)] = -1 # Bearish cross return signal

Python — Bollinger Bands, ATR, OBV def bollinger_bands(close: pd.Series, period: int = 20, std: float = 2.0): """Bollinger Bands: middle, upper, lower, %B, bandwidth.""" sma = close.rolling(period).mean() rolling_std = close.rolling(period).std() upper = sma + std * rolling_std lower = sma - std * rolling_std pct_b = (close - lower) / (upper - lower) bandwidth = (upper - lower) / sma return pd.DataFrame({ "bb_mid": sma, "bb_upper": upper, "bb_lower": lower, "bb_pctb": pct_b, "bb_bw": bandwidth }) def atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14): """Average True Range — measures volatility.""" tr1 = high - low tr2 = (high - close.shift(1)).abs() tr3 = (low - close.shift(1)).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) return tr.ewm(alpha=1/period, min_periods=period).mean() def obv(close: pd.Series, volume: pd.Series): """On-Balance Volume — cumulative volume pressure.""" direction = np.sign(close.diff()) return (volume * direction).cumsum()

DSL — Screener Filter Examples # Oversold bounce setup rsi(14) < 35 close > ema(200) volume > sma_volume(20) * 1.5 sector != "Utilities" # Breakout momentum close > high(52w) rsi(14) > 55 AND rsi(14) < 75 macd_histogram > 0 atr(14) / close > 0.02 # Mean reversion play bb_pctb(20,2) < 0.1 rsi(14) < 30 close > ema(50) insider_buys(90d) > 0

Python — Screener Engine Core from dataclasses import dataclass from typing import Callable @dataclass class Filter: """A single screening condition.""" name: str fn: Callable[[pd.DataFrame], pd.Series] # returns boolean mask weight: float = 1.0 class ScreenerEngine: """DSL-based stock screener with scoring.""" def __init__(self, db): self.db = db self.filters: list[Filter] = [] self.universe: list[str] = [] def set_universe(self, symbols: list[str]): self.universe = symbols return self def add_filter(self, name: str, fn, weight: float = 1.0): self.filters.append(Filter(name=name, fn=fn, weight=weight)) return self async def run(self, top_n: int = 20) -> pd.DataFrame: """Execute all filters and return scored results.""" # Load data for entire universe data = await self.db.get_bars(self.universe, period="1Y") # Compute indicators for each symbol results = [] for sym in self.universe: df = data[data["symbol"] == sym].copy() if len(df) < 50: continue # Apply each filter, track pass/fail score = 0 passes = 0 details = {} for f in self.filters: try: mask = f.fn(df) passed = bool(mask.iloc[-1]) if passed: score += f.weight passes += 1 details[f.name] = passed except: details[f.name] = None if passes == len(self.filters): # all filters pass results.append({ "symbol": sym, "score": score, "close": df["close"].iloc[-1], **details }) return (pd.DataFrame(results) .sort_values("score", ascending=False) .head(top_n)) # Usage screener = ScreenerEngine(db) screener.set_universe(sp500_symbols) screener.add_filter("RSI oversold", lambda df: rsi(df["close"]) < 35, weight=2.0) screener.add_filter("Above 200 EMA", lambda df: df["close"] > df["close"].ewm(span=200).mean(), weight=1.5) screener.add_filter("Volume surge", lambda df: df["volume"] > df["volume"].rolling(20).mean() * 1.5) results = await screener.run(top_n=10)

Python — Support/Resistance Detection def find_support_resistance( df: pd.DataFrame, lookback: int = 120, touches: int = 3, tolerance: float = 0.02 ) -> dict: """Find support/resistance levels from price history. A level is valid if price touches it >= `touches` times within `tolerance` (2% default). """ recent = df.tail(lookback) highs = recent["high"].values lows = recent["low"].values closes = recent["close"].values # Find local pivots (swing highs and lows) pivot_highs = [] pivot_lows = [] for i in range(2, len(recent) - 2): if highs[i] > highs[i-1] and highs[i] > highs[i+1]: pivot_highs.append(highs[i]) if lows[i] < lows[i-1] and lows[i] < lows[i+1]: pivot_lows.append(lows[i]) # Cluster pivots into levels def cluster(values, tol): if not values: return [] sorted_v = sorted(values) clusters = [[sorted_v[0]]] for v in sorted_v[1:]: if abs(v - np.mean(clusters[-1])) / np.mean(clusters[-1]) < tol: clusters[-1].append(v) else: clusters.append([v]) return [ {"level": np.mean(c), "touches": len(c), "strength": len(c) / lookback} for c in clusters if len(c) >= touches ] current_price = closes[-1] resistance = [l for l in cluster(pivot_highs, tolerance) if l["level"] > current_price] support = [l for l in cluster(pivot_lows, tolerance) if l["level"] < current_price] return {"support": support, "resistance": resistance, "price": current_price}

Python — Breakout Detection def detect_breakout( df: pd.DataFrame, consolidation_days: int = 20, volume_multiplier: float = 2.0, range_threshold: float = 0.08 # 8% range = tight consolidation ) -> dict | None: """Detect breakout from consolidation range.""" recent = df.tail(consolidation_days + 1) consolidation = recent.iloc[:-1] today = recent.iloc[-1] # Check if consolidation range is tight enough range_high = consolidation["high"].max() range_low = consolidation["low"].min() range_pct = (range_high - range_low) / range_low if range_pct > range_threshold: return None # Not consolidating avg_vol = consolidation["volume"].mean() # Bullish breakout if today["close"] > range_high and today["volume"] > avg_vol * volume_multiplier: return { "type": "bullish_breakout", "entry": today["close"], "stop": range_low, "target": today["close"] + (range_high - range_low), "volume_ratio": today["volume"] / avg_vol, } return None

Python — Composite Signal Scorer class CompositeScorer: """Multi-signal scoring engine with category weighting.""" CATEGORIES = { "momentum": 0.25, # RSI, Stochastics "trend": 0.30, # EMA cross, MACD, ADX "volatility": 0.15, # ATR squeeze, Bollinger %B "volume": 0.15, # OBV trend, volume spike "fundamental": 0.15, # Insider buys, earnings surprise } def score(self, df: pd.DataFrame) -> dict: scores = {} # Momentum (0-100) r = rsi(df["close"]).iloc[-1] scores["momentum"] = max(0, min(100, (70 - abs(r - 50)) * 2.5 # Peak at RSI 50, fade at extremes )) # Trend (0-100) m = macd(df["close"]) ema50 = df["close"].ewm(span=50).mean().iloc[-1] ema200 = df["close"].ewm(span=200).mean().iloc[-1] price = df["close"].iloc[-1] trend_score = 0 if price > ema200: trend_score += 40 if price > ema50: trend_score += 30 if m["histogram"].iloc[-1] > 0: trend_score += 30 scores["trend"] = trend_score # Volatility (0-100) — prefer moderate volatility bb = bollinger_bands(df["close"]) bw = bb["bb_bw"].iloc[-1] scores["volatility"] = max(0, min(100, 100 - abs(bw - 0.08) * 500 )) # Volume (0-100) vol_ratio = df["volume"].iloc[-1] / df["volume"].rolling(20).mean().iloc[-1] scores["volume"] = min(100, vol_ratio * 50) # Weighted composite composite = sum( scores[cat] * weight for cat, weight in self.CATEGORIES.items() if cat in scores ) return { "composite": round(composite, 1), "components": scores, "grade": self._grade(composite), } def _grade(self, score: float) -> str: if score >= 85: return "A+" if score >= 75: return "A" if score >= 65: return "B+" if score >= 55: return "B" return "C"

Python — Walk-Forward Backtester from dataclasses import dataclass, field @dataclass class Trade: symbol: str entry_date: str entry_price: float exit_date: str = "" exit_price: float = 0.0 pnl_pct: float = 0.0 hit_tp: bool = False hit_stop: bool = False class WalkForwardBacktester: """Walk-forward backtest with fixed R/R.""" def __init__(self, stop_pct: float = 0.05, tp_pct: float = 0.10, max_hold: int = 20): self.stop_pct = stop_pct self.tp_pct = tp_pct self.max_hold = max_hold def backtest(self, df: pd.DataFrame, signals: pd.Series) -> list[Trade]: """Run backtest on buy signals (1 = buy, 0 = no signal).""" trades = [] i = 0 while i < len(df): if signals.iloc[i] != 1: i += 1 continue entry = df.iloc[i] stop = entry["close"] * (1 - self.stop_pct) target = entry["close"] * (1 + self.tp_pct) trade = Trade( symbol=entry.get("symbol", ""), entry_date=str(df.index[i]), entry_price=entry["close"], ) # Simulate forward for j in range(i + 1, min(i + self.max_hold, len(df))): bar = df.iloc[j] if bar["low"] <= stop: trade.exit_price = stop trade.hit_stop = True break if bar["high"] >= target: trade.exit_price = target trade.hit_tp = True break else: trade.exit_price = df.iloc[min(i+self.max_hold, len(df)-1)]["close"] trade.exit_date = str(df.index[j if 'j' in dir() else i]) trade.pnl_pct = (trade.exit_price - trade.entry_price) / trade.entry_price trades.append(trade) i = j + 1 if trade.hit_tp or trade.hit_stop else i + self.max_hold return trades def summary(self, trades: list[Trade]) -> dict: if not trades: return {} wins = [t for t in trades if t.pnl_pct > 0] return { "total_trades": len(trades), "win_rate": len(wins) / len(trades), "avg_pnl": np.mean([t.pnl_pct for t in trades]), "hit_tp_rate": sum(t.hit_tp for t in trades) / len(trades), "hit_stop_rate": sum(t.hit_stop for t in trades) / len(trades), "max_drawdown": self._max_drawdown(trades), }