A strategy is only as good as its signals. We'll build a complete signal library covering momentum, mean reversion, volume, and volatility — then combine them into multi-factor models that capture different market regimes.
Momentum is the most persistent anomaly in financial markets. Stocks that have been going up tend to keep going up (over 3-12 month horizons). Here are the core momentum signals.
import pandas as pd import numpy as np import talib class MomentumSignals: """Momentum-based signal generators.""" @staticmethod def roc(prices: pd.Series, period: int = 20) -> pd.Series: """Rate of Change — simplest momentum measure.""" return prices.pct_change(period) @staticmethod def rsi(prices: pd.Series, period: int = 14) -> pd.Series: """RSI using TA-Lib (exponential smoothing).""" return pd.Series( talib.RSI(prices.values, timeperiod=period), index=prices.index ) @staticmethod def macd(prices: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame: """MACD — trend direction and strength.""" macd_line, signal_line, histogram = talib.MACD( prices.values, fastperiod=fast, slowperiod=slow, signalperiod=signal ) return pd.DataFrame({ 'macd': macd_line, 'signal': signal_line, 'histogram': histogram }, index=prices.index) @staticmethod def adx(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series: """ADX — trend strength (not direction). ADX > 25 = trending, ADX < 20 = ranging.""" return pd.Series( talib.ADX(high.values, low.values, close.values, period), index=close.index ) @staticmethod def dual_momentum(prices: pd.Series, benchmark: pd.Series, lookback: int = 252) -> pd.Series: """Dual momentum: long only if both absolute and relative momentum are positive.""" abs_mom = prices.pct_change(lookback) # Absolute rel_mom = abs_mom - benchmark.pct_change(lookback) # Relative signal = pd.Series(0, index=prices.index) signal[(abs_mom > 0) & (rel_mom > 0)] = 1 return signal
Short-term (1-4 weeks): mean-reverting. Stocks that dropped last week tend to bounce. Medium-term (3-12 months): momentum. Winners keep winning. Long-term (3-5 years): mean-reverting again. This is the "momentum anomaly" — one of the most robust findings in academic finance, documented since Jegadeesh & Titman (1993).
| Signal | Type | Best Market | Typical Lookback | Strength |
|---|---|---|---|---|
| ROC | Rate of Change | Trending | 20-60 days | Simple, versatile |
| RSI | Oscillator | Range-bound | 14 days | Overbought/oversold |
| MACD | Trend + momentum | Trending | 12/26/9 | Direction + strength |
| ADX | Trend strength | Both | 14 days | Filter (trend vs range) |
| Dual Momentum | Cross-asset | Trending | 12 months | Regime filter |
If momentum says "the trend is your friend," mean reversion says "rubber bands snap back." These signals identify when price has deviated too far from its norm.
class ReversionSignals: """Mean reversion signal generators.""" @staticmethod def bollinger_bands(prices: pd.Series, period: int = 20, num_std: float = 2.0) -> pd.DataFrame: """Bollinger Bands — price relative to volatility envelope.""" sma = prices.rolling(period).mean() std = prices.rolling(period).std() upper = sma + num_std * std lower = sma - num_std * std # %B: where price sits within the bands (0=lower, 1=upper) pct_b = (prices - lower) / (upper - lower) return pd.DataFrame({ 'upper': upper, 'middle': sma, 'lower': lower, 'pct_b': pct_b, 'bandwidth': (upper - lower) / sma }, index=prices.index) @staticmethod def zscore(prices: pd.Series, period: int = 20) -> pd.Series: """Z-score: how many std devs from the rolling mean. |z| > 2 = extreme. Buy at z < -2, sell at z > +2.""" mean = prices.rolling(period).mean() std = prices.rolling(period).std() return (prices - mean) / std @staticmethod def pairs_spread(price_a: pd.Series, price_b: pd.Series, lookback: int = 60) -> pd.DataFrame: """Pairs trading spread with z-score. Trade the spread between two correlated assets.""" # Log price ratio (hedge ratio = 1 for simplicity) spread = np.log(price_a / price_b) z = (spread - spread.rolling(lookback).mean()) / \ spread.rolling(lookback).std() # Signal: buy spread when z < -2, sell when z > +2 signal = pd.Series(0, index=price_a.index) signal[z < -2] = 1 # Long A, short B signal[z > 2] = -1 # Short A, long B return pd.DataFrame({ 'spread': spread, 'zscore': z, 'signal': signal }, index=price_a.index)
Use ADX as a regime filter: when ADX > 25, the market is trending — use momentum signals. When ADX < 20, the market is range-bound — use mean reversion. This simple regime switch can dramatically improve performance. We'll formalize this in the multi-factor section below.
Price tells you WHAT happened. Volume tells you WHO was behind it. A breakout on 3x average volume is real. A breakout on thin volume is suspect.
class VolumeSignals: """Volume-based signal generators.""" @staticmethod def obv(close: pd.Series, volume: pd.Series) -> pd.Series: """On-Balance Volume — cumulative volume flow. Rising OBV with flat price = accumulation (bullish). Falling OBV with flat price = distribution (bearish).""" direction = np.sign(close.diff()) return (direction * volume).cumsum() @staticmethod def vwap_deviation(close: pd.Series, volume: pd.Series, period: int = 20) -> pd.Series: """Price deviation from rolling VWAP. Positive = above VWAP (bullish), negative = below (bearish).""" typical_price = close # Simplified; ideally (H+L+C)/3 vwap = (typical_price * volume).rolling(period).sum() / \ volume.rolling(period).sum() return (close - vwap) / vwap @staticmethod def volume_spike(volume: pd.Series, period: int = 20, threshold: float = 2.0) -> pd.Series: """Detect volume spikes (> threshold * average). Volume spikes often precede significant moves.""" avg_vol = volume.rolling(period).mean() vol_ratio = volume / avg_vol return vol_ratio @staticmethod def mfi(high: pd.Series, low: pd.Series, close: pd.Series, volume: pd.Series, period: int = 14) -> pd.Series: """Money Flow Index — volume-weighted RSI. MFI < 20 = oversold, MFI > 80 = overbought.""" return pd.Series( talib.MFI(high.values, low.values, close.values, volume.values, period), index=close.index )
Volatility is mean-reverting: high-volatility periods compress, low-volatility periods explode. Understanding volatility regimes is key to position sizing (Part 4) and strategy selection.
class VolatilitySignals: """Volatility-based signal generators.""" @staticmethod def atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series: """Average True Range — measures volatility in price units. Used for stop losses and position sizing.""" return pd.Series( talib.ATR(high.values, low.values, close.values, period), index=close.index ) @staticmethod def bollinger_squeeze(prices: pd.Series, bb_period: int = 20, kc_period: int = 20, kc_mult: float = 1.5) -> pd.Series: """Bollinger Squeeze — BB inside Keltner Channel. Squeeze = low volatility, breakout imminent. 1 = squeeze on, 0 = squeeze off.""" # Bollinger Bands bb_mid = prices.rolling(bb_period).mean() bb_std = prices.rolling(bb_period).std() bb_upper = bb_mid + 2 * bb_std bb_lower = bb_mid - 2 * bb_std # Keltner Channel (simplified with rolling range) kc_mid = prices.rolling(kc_period).mean() kc_range = prices.rolling(kc_period).apply( lambda x: np.mean(np.abs(np.diff(x))) ) * kc_mult kc_upper = kc_mid + kc_range kc_lower = kc_mid - kc_range # Squeeze: BB inside KC squeeze = ((bb_lower > kc_lower) & (bb_upper < kc_upper)).astype(int) return squeeze @staticmethod def realized_vs_implied(close: pd.Series, vix: pd.Series, period: int = 20) -> pd.Series: """Volatility Risk Premium = Implied Vol - Realized Vol. Positive = options overpriced (sell vol). Negative = options underpriced (buy vol).""" realized = close.pct_change().rolling(period).std() * np.sqrt(252) * 100 return vix - realized
Individual signals are noisy. Combining them with proper weighting creates a more robust composite signal. The key insight: signals from different families (momentum, volume, volatility) are less correlated, so their combination is more powerful than using three momentum signals.
class CompositeSignal: """Combine multiple signals into a single score.""" def __init__(self, weights: dict = None): self.weights = weights or { 'momentum': 0.30, 'reversion': 0.25, 'volume': 0.20, 'volatility': 0.15, 'trend': 0.10, } def normalize(self, signal: pd.Series) -> pd.Series: """Normalize signal to [-1, +1] range using z-score.""" z = (signal - signal.rolling(252).mean()) / \ signal.rolling(252).std() return z.clip(-3, 3) / 3 # Scale to [-1, 1] def score(self, data: pd.DataFrame) -> pd.Series: """ Compute composite score from OHLCV data. Returns: pd.Series of scores in [-1, +1] > +0.3 = bullish, < -0.3 = bearish """ close = data['Close'] high = data['High'] low = data['Low'] volume = data['Volume'] # Compute individual signals mom = MomentumSignals rev = ReversionSignals vol = VolumeSignals vlt = VolatilitySignals # Momentum: RSI normalized rsi = mom.rsi(close) / 100 * 2 - 1 # [0,100] -> [-1,1] # Reversion: Bollinger %B bb = rev.bollinger_bands(close) bb_sig = -(bb['pct_b'] * 2 - 1) # Inverted: low %B = buy # Volume: VWAP deviation vwap = vol.vwap_deviation(close, volume) vwap_sig = self.normalize(vwap) # Volatility: squeeze as filter (1 = ready to move) squeeze = vlt.bollinger_squeeze(close) # Trend: ADX-filtered MACD macd_data = mom.macd(close) adx = mom.adx(high, low, close) trend_sig = self.normalize(macd_data['histogram']) trend_sig[adx < 20] = 0 # Kill trend signal in ranging market # Weighted composite composite = ( self.weights['momentum'] * rsi + self.weights['reversion'] * bb_sig + self.weights['volume'] * vwap_sig + self.weights['volatility'] * squeeze * trend_sig.abs() + self.weights['trend'] * trend_sig ) return composite.clip(-1, 1) def generate_signals(self, data: pd.DataFrame, threshold: float = 0.3) -> pd.Series: """Convert composite score to trading signals.""" score = self.score(data) signals = pd.Series(0, index=data.index) signals[score > threshold] = 1 # Long signals[score < -threshold] = -1 # Short return signals
def signal_quality(signal: pd.Series, forward_returns: pd.Series) -> dict: """ Measure how predictive a signal is. Args: signal: signal values (continuous, e.g., z-score) forward_returns: next-day returns """ # Information Coefficient: rank correlation between # signal and forward returns ic = signal.corr(forward_returns, method='spearman') # Hit rate: % of times signal direction matches return direction hits = ((signal > 0) == (forward_returns > 0)).mean() # Expectancy: average return when signal is positive long_ret = forward_returns[signal > 0].mean() short_ret = forward_returns[signal < 0].mean() # Decay: IC over different forward horizons decay = {} for days in [1, 5, 10, 20]: fwd = forward_returns.rolling(days).sum().shift(-days) decay[f'{days}d'] = signal.corr(fwd, method='spearman') return { 'ic': ic, 'hit_rate': hits, 'long_return': long_ret, 'short_return': short_ret, 'long_short_spread': long_ret - short_ret, 'ic_decay': decay, }
When you move to ML models (Part 6), your signals become features. Good features capture different aspects of market behavior and have low mutual correlation.
def build_features(data: pd.DataFrame, lookbacks: list = [5, 10, 20, 60]) -> pd.DataFrame: """ Generate ML features from OHLCV data. Returns DataFrame with ~40 features per row. """ close = data['Close'] volume = data['Volume'] features = pd.DataFrame(index=data.index) for lb in lookbacks: # Price-based features[f'ret_{lb}d'] = close.pct_change(lb) features[f'vol_{lb}d'] = close.pct_change().rolling(lb).std() features[f'zscore_{lb}d'] = (close - close.rolling(lb).mean()) / \ close.rolling(lb).std() features[f'high_dist_{lb}d'] = close / close.rolling(lb).max() - 1 features[f'low_dist_{lb}d'] = close / close.rolling(lb).min() - 1 # Volume-based features[f'vol_ratio_{lb}d'] = volume / volume.rolling(lb).mean() # Technical indicators features['rsi_14'] = MomentumSignals.rsi(close, 14) / 100 features['macd_hist'] = MomentumSignals.macd(close)['histogram'] # Cross-sectional rank (useful for stock selection) # features['rank_ret_20'] = close.pct_change(20).rank(pct=True) # Day of week, month (calendar effects) features['dow'] = data.index.dayofweek / 4 # [0, 1] features['month'] = data.index.month / 12 # [0, 1] return features.dropna()
Never use forward-looking features: all features must be computed from past data only. Normalize everything: ML models struggle with features on different scales (price vs volume vs RSI). Remove highly correlated features: if two features have >0.9 correlation, drop one. Label encoding: for classification, use next-day return sign (+1/-1), not the return itself.