Building a Market Scanner: Find Oversold Forex Pairs Instantly

import requests
from dataclasses import dataclass

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

ALL_PAIRS = [
    "EURUSD", "GBPUSD", "USDJPY", "AUDUSD", "USDCHF",
    "USDCAD", "NZDUSD", "EURGBP", "EURJPY", "GBPJPY",
    "AUDJPY", "XAUUSD", "BTCUSD", "ETHUSD"
]

@dataclass
class ScanResult:
    symbol: str
    score: int  # 0-100, higher = more oversold
    rsi: float
    bb_position: str
    stoch_k: float
    price: float

def scan_oversold(timeframe: str = "H1") -> list[ScanResult]:
    """Scan all pairs for oversold conditions."""
    results = []

    for symbol in ALL_PAIRS:
        resp = requests.get(f"{BASE_URL}/indicators", params={
            "symbol": symbol,
            "timeframe": timeframe,
        }, headers=HEADERS)

        if resp.status_code != 200:
            continue

        data = resp.json()["data"]
        ind = data["indicators"]
        price = data.get("ohlcv", {}).get("close", 0)

        rsi = ind["RSI_14"]
        stoch = ind["Stochastic_K"]

        # Calculate oversold score (0-100)
        score = 0

        # RSI contribution (max 40 points)
        if rsi < 20:
            score += 40
        elif rsi < 30:
            score += 30
        elif rsi < 40:
            score += 15

        # Bollinger Band contribution (max 30 points)
        if price <= ind["BB_lower"]:
            score += 30
        elif price <= ind["BB_lower"] * 1.005:
            score += 20
        elif price < ind["BB_middle"]:
            score += 5

        # Stochastic contribution (max 30 points)
        if stoch < 10:
            score += 30
        elif stoch < 20:
            score += 25
        elif stoch < 30:
            score += 15

        bb_pos = "below" if price <= ind["BB_lower"] else \
                 "near_lower" if price < ind["BB_middle"] else "middle"

        results.append(ScanResult(
            symbol=symbol,
            score=score,
            rsi=rsi,
            bb_position=bb_pos,
            stoch_k=stoch,
            price=price
        ))

    # Sort by oversold score (highest first)
    results.sort(key=lambda r: r.score, reverse=True)
    return results

def print_scan_results(results: list[ScanResult],
                       min_score: int = 30):
    """Display scanner results in a formatted table."""
    filtered = [r for r in results if r.score >= min_score]

    if not filtered:
        print("No oversold pairs found.")
        return

    print(f"\n{'Symbol':<10} {'Score':>5} {'RSI':>6} {'Stoch':>6} {'BB Position':<12}")
    print("-" * 50)

    for r in filtered:
        print(f"{r.symbol:<10} {r.score:>5} {r.rsi:>6.1f} "
              f"{r.stoch_k:>6.1f} {r.bb_position:<12}")

# Scan and display
results = scan_oversold("H1")
print_scan_results(results)

# Example output:
# Symbol     Score    RSI  Stoch BB Position
# --------------------------------------------------
# AUDJPY        75   24.1   12.3 below
# NZDUSD        60   28.7   18.5 near_lower
# GBPJPY        45   33.2   22.1 near_lower

def multi_timeframe_scan() -> list[dict]:
    """Find pairs that are oversold on multiple timeframes."""
    timeframes = ["H1", "H4", "D1"]
    pair_scores = {}

    for tf in timeframes:
        results = scan_oversold(tf)
        for r in results:
            if r.symbol not in pair_scores:
                pair_scores[r.symbol] = {"scores": {}, "details": {}}
            pair_scores[r.symbol]["scores"][tf] = r.score
            pair_scores[r.symbol]["details"][tf] = {
                "rsi": r.rsi, "stoch": r.stoch_k
            }

    # Filter: must be oversold on at least 2 timeframes
    confirmed = []
    for symbol, data in pair_scores.items():
        oversold_tfs = [tf for tf, s in data["scores"].items() if s >= 40]
        if len(oversold_tfs) >= 2:
            confirmed.append({
                "symbol": symbol,
                "avg_score": sum(data["scores"].values()) / len(data["scores"]),
                "oversold_timeframes": oversold_tfs,
                "details": data["details"]
            })

    confirmed.sort(key=lambda x: x["avg_score"], reverse=True)
    return confirmed

# Find multi-timeframe oversold pairs
strong_signals = multi_timeframe_scan()
for sig in strong_signals:
    tfs = ", ".join(sig["oversold_timeframes"])
    print(f"{sig['symbol']}: oversold on {tfs} (avg score: {sig['avg_score']:.0f})")

import time
from datetime import datetime

def run_scanner_loop():
    """Run scanner on a schedule and alert on new signals."""
    previous_signals = set()

    while True:
        results = scan_oversold("H1")
        current_signals = set()

        for r in results:
            if r.score >= 50:  # Only high-confidence signals
                current_signals.add(r.symbol)

                if r.symbol not in previous_signals:
                    print(f"\n[{datetime.utcnow()}] NEW: {r.symbol} "
                          f"(score: {r.score}, RSI: {r.rsi:.1f})")
                    # send_telegram_alert(r)  # Optional

        # Report any pairs that recovered
        recovered = previous_signals - current_signals
        for symbol in recovered:
            print(f"[{datetime.utcnow()}] RECOVERED: {symbol} no longer oversold")

        previous_signals = current_signals
        time.sleep(300)  # Scan every 5 minutes