import pandas as pd
import numpy as np
def calculate_rsi(prices, period=14):
    """
    RSI measures momentum strength (0-100)
    >70 = overbought (potential sell), <30 = oversold (potential buy)
    """
deltas = prices.diff()
    seed = deltas[:period+1]
    up = seed[seed >= 0].sum() / period
    down = -seed[seed < 0].sum() / period
rs = up / down
    rsi = np.zeros_like(prices, dtype=float)
    rsi[:period] = 100.0 - 100.0 / (1.0 + rs)
for i in range(period, len(prices)):
        delta = deltas.iloc[i]
        if delta > 0:
            up = (up * (period - 1) + delta) / period
            down = (down * (period - 1) + 0) / period
        else:
            up = (up * (period - 1) + 0) / period
            down = (down * (period - 1) + abs(delta)) / period
rs = up / down
        rsi[i] = 100.0 - 100.0 / (1.0 + rs)
return rsi
Example: Trading with RSI
prices = pd.Series([100, 101, 102, 101, 103, 105, 104, 106, 108, 107])
rsi = calculate_rsi(prices)
Buy when RSI crosses above 50 (momentum shift to upside)
Sell when RSI crosses below 50 (momentum shift to downside)
momentum_signals = (rsi > 50).astype(int)
print("RSI-based momentum signals:", momentum_signals.values)

def calculate_macd(prices, fast=12, slow=26, signal=9):
    """
    MACD measures momentum through exponential moving average divergence
    Positive MACD = upside momentum, Negative = downside momentum
    """
ema_fast = prices.ewm(span=fast, adjust=False).mean()
    ema_slow = prices.ewm(span=slow, adjust=False).mean()
macd_line = ema_fast - ema_slow
    signal_line = macd_line.ewm(span=signal, adjust=False).mean()
    histogram = macd_line - signal_line
return macd_line, signal_line, histogram
Generate momentum signals
prices = fetch_daily_prices('AAPL', years=2)
macd, signal, histogram = calculate_macd(prices['close'])
Buy when MACD crosses above signal line (momentum accelerating upward)
buy_signal = (macd > signal) & (macd.shift(1) <= signal.shift(1))
Sell when MACD crosses below signal line
sell_signal = (macd < signal) & (macd.shift(1) >= signal.shift(1))

def calculate_adx(high, low, close, period=14):
    """
    ADX measures trend strength (0-100)
    >25 = strong trend (good for momentum), <20 = weak trend (avoid trading)
    """
# True Range
    tr1 = high - low
    tr2 = abs(high - close.shift(1))
    tr3 = abs(low - close.shift(1))
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
# Directional Movement
    plus_dm = high.diff()
    minus_dm = -low.diff()
plus_dm[plus_dm < 0] = 0
    minus_dm[minus_dm < 0] = 0
# Plus and Minus DI
    tr_sum = tr.rolling(window=period).sum()
    plus_di = 100 * (plus_dm.rolling(window=period).sum() / tr_sum)
    minus_di = 100 * (minus_dm.rolling(window=period).sum() / tr_sum)
# DX and ADX
    dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
    adx = dx.rolling(window=period).mean()
return adx, plus_di, minus_di
Use ADX to filter momentum trades
adx, plus_di, minus_di = calculate_adx(data['high'], data['low'], data['close'])
Only trade momentum when ADX > 25 (strong trend)
momentum_trades = signals & (adx > 25)

class SimpleMomentumTrader:
    def __init__(self, initial_capital=100000, risk_per_trade=0.02):
        self.capital = initial_capital
        self.risk_per_trade = risk_per_trade
        self.positions = {}
        self.trades_executed = []
def identify_momentum_candidates(self, universe, lookback=20):
        """
        Identify stocks with strongest upside momentum
        Metric: Return over lookback period
        """
momentum_scores = {}
for symbol in universe:
            prices = fetch_daily_prices(symbol, days=lookback)
            returns = (prices[-1] - prices[0]) / prices[0]
            momentum_scores[symbol] = returns
# Rank by momentum (top 10%)
        ranked = sorted(momentum_scores.items(),
                       key=lambda x: x[1], reverse=True)
        top_momentum = [x[0] for x in ranked[:int(len(ranked)*0.1)]]
return top_momentum
def calculate_entry_price(self, symbol, prices):
        """
        Entry: breakout above 20-day high
        This confirms momentum and reduces false signals
        """
high_20 = prices[-20:].max()
        current_price = prices[-1]
# Buy on breakout
        if current_price > high_20 * 1.005:  # 0.5% above high
            return current_price, 'BREAKOUT_BUY'
return None, None
def calculate_stop_loss(self, entry_price, atr):
        """Stop loss 2x ATR below entry to account for volatility"""
        return entry_price - (2.0 * atr)
def calculate_profit_target(self, entry_price, atr):
        """Profit target 3x ATR above entry (1:1.5 risk-reward)"""
        return entry_price + (3.0 * atr)
def execute_momentum_trade(self, symbol, current_price, atr, signal):
        """Execute entry with proper position sizing"""
# Calculate position size
        risk_amount = self.capital * self.risk_per_trade
        stop_distance = 2.0 * atr
        position_size = risk_amount / stop_distance
trade = {
            'symbol': symbol,
            'entry_price': current_price,
            'entry_date': datetime.now(),
            'position_size': position_size,
            'stop_loss': current_price - (2.0 * atr),
            'profit_target': current_price + (3.0 * atr),
            'signal_type': signal
        }
self.positions[symbol] = trade
        self.trades_executed.append(trade)
return trade
Usage
trader = SimpleMomentumTrader(initial_capital=100000)
nasdaq100 = ['AAPL', 'MSFT', 'GOOGL']  # ... 100 symbols
Daily execution
momentum_stocks = trader.identify_momentum_candidates(nasdaq100)
for symbol in momentum_stocks[:5]:  # Trade top 5
    prices = fetch_daily_prices(symbol, days=50)
    atr = calculate_atr(prices['high'], prices['low'], prices['close'])
    entry_price, signal = trader.calculate_entry_price(symbol, prices)
if entry_price:
        trader.execute_momentum_trade(symbol, entry_price, atr[-1], signal)