How to Detect RSI Divergence Programmatically

import requests
import pandas as pd

API_KEY = "your_api_key_here"
BASE_URL = "https://tickatlas.com/v1"
HEADERS = {"X-API-Key": API_KEY}

def fetch_data(symbol: str, timeframe: str, bars: int = 100):
    """Fetch OHLC data from the API."""
    resp = requests.get(f"{BASE_URL}/ohlc", params={
        "symbol": symbol,
        "timeframe": timeframe,
        "limit": bars
    }, headers=HEADERS)
    candles = resp.json()["data"]["candles"]
    df = pd.DataFrame(candles)
    df["time"] = pd.to_datetime(df["time"])
    return df

def compute_rsi(closes: pd.Series, period: int = 14) -> pd.Series:
    """Calculate RSI from a series of close prices."""
    delta = closes.diff()
    gain = delta.where(delta > 0, 0.0)
    loss = -delta.where(delta < 0, 0.0)

    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()

    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

def find_swing_lows(series: pd.Series, lookback: int = 5) -> list[tuple]:
    """Find local minima in a series.
    Returns list of (index, value) tuples."""
    swings = []
    for i in range(lookback, len(series) - lookback):
        window = series.iloc[i - lookback:i + lookback + 1]
        if series.iloc[i] == window.min():
            swings.append((i, series.iloc[i]))
    return swings

def find_swing_highs(series: pd.Series, lookback: int = 5) -> list[tuple]:
    """Find local maxima in a series."""
    swings = []
    for i in range(lookback, len(series) - lookback):
        window = series.iloc[i - lookback:i + lookback + 1]
        if series.iloc[i] == window.max():
            swings.append((i, series.iloc[i]))
    return swings

def detect_divergence(df: pd.DataFrame) -> list[dict]:
    """Detect RSI divergence in OHLC data."""
    df["rsi"] = compute_rsi(df["close"])
    divergences = []

    # Find swing lows for bullish divergence
    price_lows = find_swing_lows(df["close"])
    rsi_lows = find_swing_lows(df["rsi"])

    # Compare last two swing lows
    if len(price_lows) >= 2 and len(rsi_lows) >= 2:
        p1_idx, p1_val = price_lows[-2]
        p2_idx, p2_val = price_lows[-1]
        r1_idx, r1_val = rsi_lows[-2]
        r2_idx, r2_val = rsi_lows[-1]

        # Bullish: price lower low, RSI higher low
        if p2_val < p1_val and r2_val > r1_val:
            divergences.append({
                "type": "BULLISH",
                "price_low_1": p1_val,
                "price_low_2": p2_val,
                "rsi_low_1": r1_val,
                "rsi_low_2": r2_val,
                "bar_index": p2_idx
            })

    # Find swing highs for bearish divergence
    price_highs = find_swing_highs(df["close"])
    rsi_highs = find_swing_highs(df["rsi"])

    if len(price_highs) >= 2 and len(rsi_highs) >= 2:
        p1_idx, p1_val = price_highs[-2]
        p2_idx, p2_val = price_highs[-1]
        r1_idx, r1_val = rsi_highs[-2]
        r2_idx, r2_val = rsi_highs[-1]

        # Bearish: price higher high, RSI lower high
        if p2_val > p1_val and r2_val < r1_val:
            divergences.append({
                "type": "BEARISH",
                "price_high_1": p1_val,
                "price_high_2": p2_val,
                "rsi_high_1": r1_val,
                "rsi_high_2": r2_val,
                "bar_index": p2_idx
            })

    return divergences

# Scan for divergence across multiple pairs and timeframes
pairs = ["EURUSD", "GBPUSD", "USDJPY", "XAUUSD", "BTCUSD"]
timeframes = ["H1", "H4"]

for pair in pairs:
    for tf in timeframes:
        df = fetch_data(pair, tf, 100)
        divs = detect_divergence(df)
        for d in divs:
            print(f"{pair} ({tf}): {d['type']} divergence detected!")
            if d["type"] == "BULLISH":
                print(f"  Price: {d['price_low_1']:.5f} -> {d['price_low_2']:.5f} (lower)")
                print(f"  RSI:   {d['rsi_low_1']:.1f} -> {d['rsi_low_2']:.1f} (higher)")
            else:
                print(f"  Price: {d['price_high_1']:.5f} -> {d['price_high_2']:.5f} (higher)")
                print(f"  RSI:   {d['rsi_high_1']:.1f} -> {d['rsi_high_2']:.1f} (lower)")