Pair Scanner & Screener
import pandas as pd, numpy as np
from typing import Callable
class PairScanner:
@staticmethod
def scan(pairs_data: dict, conditions: list[dict]) -> list[dict]:
"""
Scan all pairs against a list of conditions.
pairs_data: {"EURUSD": df, "GBPUSD": df, ...}
conditions: [{"name": "RSI_oversold", "fn": lambda df: rsi(df) < 30}, ...]
"""
results = []
for symbol, df in pairs_data.items():
if df.empty or len(df) < 50: continue
matches = []
for cond in conditions:
try:
if cond["fn"](df):
matches.append(cond["name"])
except: continue
if matches:
results.append({
"symbol": symbol,
"conditions_met": matches,
"n_conditions": len(matches),
"score": len(matches) / len(conditions),
})
return sorted(results, key=lambda r: r["n_conditions"], reverse=True)
@staticmethod
def preset_scans() -> dict:
"""Pre-built scan conditions for common setups."""
def _rsi(df, period=14):
d = df["close"].diff()
g = d.where(d > 0, 0).rolling(period).mean()
l = (-d.where(d < 0, 0)).rolling(period).mean()
return (100 - 100 / (1 + g / l.replace(0, np.nan))).iloc[-1]
return {
"oversold_bounce": [
{"name": "RSI<30", "fn": lambda df: _rsi(df) < 30},
{"name": "above_SMA200", "fn": lambda df: df["close"].iloc[-1] > df["close"].rolling(200).mean().iloc[-1]},
{"name": "bullish_candle", "fn": lambda df: df["close"].iloc[-1] > df["open"].iloc[-1]}],
"overbought_fade": [
{"name": "RSI>70", "fn": lambda df: _rsi(df) > 70},
{"name": "below_SMA200", "fn": lambda df: df["close"].iloc[-1] < df["close"].rolling(200).mean().iloc[-1]},
{"name": "bearish_candle", "fn": lambda df: df["close"].iloc[-1] < df["open"].iloc[-1]}],
"breakout_candidate": [
{"name": "BB_squeeze", "fn": lambda df: (df["close"].rolling(20).std().iloc[-1] / df["close"].rolling(20).mean().iloc[-1]) < 0.005},
{"name": "volume_rising", "fn": lambda df: df["volume"].iloc[-1] > df["volume"].rolling(20).mean().iloc[-1] * 1.3},
{"name": "near_20bar_high", "fn": lambda df: df["close"].iloc[-1] > df["high"].rolling(20).max().iloc[-2] * 0.998}],
"trend_pullback": [
{"name": "above_EMA50", "fn": lambda df: df["close"].iloc[-1] > df["close"].ewm(span=50).mean().iloc[-1]},
{"name": "touching_EMA20", "fn": lambda df: abs(df["close"].iloc[-1] - df["close"].ewm(span=20).mean().iloc[-1]) < (df["high"] - df["low"]).rolling(14).mean().iloc[-1] * 0.5},
{"name": "RSI_40_60", "fn": lambda df: 40 < _rsi(df) < 60}],
}
@staticmethod
def quick_scan(pairs_data: dict, scan_name: str = "oversold_bounce") -> list[dict]:
presets = PairScanner.preset_scans()
conditions = presets.get(scan_name, presets["oversold_bounce"])
return PairScanner.scan(pairs_data, conditions)
1---2name: pair-scanner-screener3description: Scan all available pairs for specific technical conditions — overbought, oversold, breakout, squeeze, divergence, pattern match. Use for "scan all pairs", "screener", "find setups", "which pairs have RSI oversold", "scan for breakouts", "find squeeze setups", "market scan", "pair filter", "setup scanner", "opportunity finder", or any multi-pair screening. Works with mt5-chart-browser for data and all strategy/indicator skills for conditions.4---56# Pair Scanner & Screener78```python9import pandas as pd, numpy as np10from typing import Callable1112class PairScanner:1314 @staticmethod15 def scan(pairs_data: dict, conditions: list[dict]) -> list[dict]:16 """17 Scan all pairs against a list of conditions.18 pairs_data: {"EURUSD": df, "GBPUSD": df, ...}19 conditions: [{"name": "RSI_oversold", "fn": lambda df: rsi(df) < 30}, ...]20 """21 results = []22 for symbol, df in pairs_data.items():23 if df.empty or len(df) < 50: continue24 matches = []25 for cond in conditions:26 try:27 if cond["fn"](df):28 matches.append(cond["name"])29 except: continue30 if matches:31 results.append({32 "symbol": symbol,33 "conditions_met": matches,34 "n_conditions": len(matches),35 "score": len(matches) / len(conditions),36 })37 return sorted(results, key=lambda r: r["n_conditions"], reverse=True)3839 @staticmethod40 def preset_scans() -> dict:41 """Pre-built scan conditions for common setups."""42 def _rsi(df, period=14):43 d = df["close"].diff()44 g = d.where(d > 0, 0).rolling(period).mean()45 l = (-d.where(d < 0, 0)).rolling(period).mean()46 return (100 - 100 / (1 + g / l.replace(0, np.nan))).iloc[-1]4748 return {49 "oversold_bounce": [50 {"name": "RSI<30", "fn": lambda df: _rsi(df) < 30},51 {"name": "above_SMA200", "fn": lambda df: df["close"].iloc[-1] > df["close"].rolling(200).mean().iloc[-1]},52 {"name": "bullish_candle", "fn": lambda df: df["close"].iloc[-1] > df["open"].iloc[-1]}],53 "overbought_fade": [54 {"name": "RSI>70", "fn": lambda df: _rsi(df) > 70},55 {"name": "below_SMA200", "fn": lambda df: df["close"].iloc[-1] < df["close"].rolling(200).mean().iloc[-1]},56 {"name": "bearish_candle", "fn": lambda df: df["close"].iloc[-1] < df["open"].iloc[-1]}],57 "breakout_candidate": [58 {"name": "BB_squeeze", "fn": lambda df: (df["close"].rolling(20).std().iloc[-1] / df["close"].rolling(20).mean().iloc[-1]) < 0.005},59 {"name": "volume_rising", "fn": lambda df: df["volume"].iloc[-1] > df["volume"].rolling(20).mean().iloc[-1] * 1.3},60 {"name": "near_20bar_high", "fn": lambda df: df["close"].iloc[-1] > df["high"].rolling(20).max().iloc[-2] * 0.998}],61 "trend_pullback": [62 {"name": "above_EMA50", "fn": lambda df: df["close"].iloc[-1] > df["close"].ewm(span=50).mean().iloc[-1]},63 {"name": "touching_EMA20", "fn": lambda df: abs(df["close"].iloc[-1] - df["close"].ewm(span=20).mean().iloc[-1]) < (df["high"] - df["low"]).rolling(14).mean().iloc[-1] * 0.5},64 {"name": "RSI_40_60", "fn": lambda df: 40 < _rsi(df) < 60}],65 }6667 @staticmethod68 def quick_scan(pairs_data: dict, scan_name: str = "oversold_bounce") -> list[dict]:69 presets = PairScanner.preset_scans()70 conditions = presets.get(scan_name, presets["oversold_bounce"])71 return PairScanner.scan(pairs_data, conditions)72```737475---
mahmoud20138