src/scanner.py

"""
Live Market Scanner — fetches price data and computes technical scores.
"""

import yfinance as yf
import pandas as pd
import ta
from datetime import datetime, timedelta
from typing import Optional


# Default watchlists
WATCHLISTS = {
    "US Tech": ["NVDA", "AAPL", "TSLA", "MSFT", "GOOGL", "META", "AMZN", "AMD"],
    "Crypto": ["BTCUSD", "ETHUSD", "SOLUSD", "XRPUSD", "ADAUSD"],
    "Forex": ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD", "USDCAD"],
    "SGX": ["D05", "O39", "U11", "Z74", "C6L"],
}


def fetch_ticker_data(ticker: str, period: str = "3mo", interval: str = "1d") -> Optional[pd.DataFrame]:
    """Fetch OHLCV data for a single ticker."""
    try:
        data = yf.download(ticker, period=period, interval=interval, progress=False)
        if data.empty:
            return None
        # Flatten multi-level columns if present
        if isinstance(data.columns, pd.MultiIndex):
            data.columns = data.columns.get_level_values(0)
        return data
    except Exception:
        return None


def compute_indicators(df: pd.DataFrame) -> dict:
    """Compute technical indicators and return a summary dict."""
    if df is None or len(df) < 20:
        return {"error": "Insufficient data"}

    close = df["Close"].astype(float)
    high = df["High"].astype(float)
    low = df["Low"].astype(float)
    volume = df["Volume"].astype(float)

    result = {}

    # Current price
    result["current_price"] = round(float(close.iloc[-1]), 2)
    result["prev_close"] = round(float(close.iloc[-2]), 2)
    result["change_pct"] = round((result["current_price"] - result["prev_close"]) / result["prev_close"] * 100, 2)

    # RSI (14)
    rsi = ta.momentum.RSIIndicator(close, window=14)
    result["rsi"] = round(float(rsi.rsi().iloc[-1]), 1) if not rsi.rsi().iloc[-1] != rsi.rsi().iloc[-1] else None

    # MACD
    macd = ta.trend.MACD(close)
    macd_line = macd.macd().iloc[-1]
    signal_line = macd.macd_signal().iloc[-1]
    result["macd"] = round(float(macd_line), 4) if pd.notna(macd_line) else None
    result["macd_signal"] = round(float(signal_line), 4) if pd.notna(signal_line) else None
    result["macd_bullish"] = bool(macd_line > signal_line) if pd.notna(macd_line) and pd.notna(signal_line) else None

    # EMAs
    ema9 = ta.trend.EMAIndicator(close, window=9).ema_indicator().iloc[-1]
    ema21 = ta.trend.EMAIndicator(close, window=21).ema_indicator().iloc[-1]
    ema50 = ta.trend.EMAIndicator(close, window=50).ema_indicator().iloc[-1] if len(close) >= 50 else None
    result["ema9"] = round(float(ema9), 2) if pd.notna(ema9) else None
    result["ema21"] = round(float(ema21), 2) if pd.notna(ema21) else None
    result["ema50"] = round(float(ema50), 2) if ema50 is not None and pd.notna(ema50) else None
    result["ema_bullish"] = bool(ema9 > ema21) if pd.notna(ema9) and pd.notna(ema21) else None

    # Volume trend
    vol_avg = volume.rolling(20).mean().iloc[-1]
    vol_current = volume.iloc[-1]
    result["volume_ratio"] = round(float(vol_current / vol_avg), 2) if pd.notna(vol_avg) and vol_avg > 0 else None

    # ATR (14) for volatility
    atr = ta.volatility.AverageTrueRange(high, low, close, window=14)
    atr_val = atr.average_true_range().iloc[-1]
    result["atr"] = round(float(atr_val), 4) if pd.notna(atr_val) else None
    result["atr_pct"] = round(float(atr_val / close.iloc[-1] * 100), 2) if pd.notna(atr_val) else None

    # Support/Resistance (simple: recent swing high/low)
    recent = df.tail(20)
    result["recent_high"] = round(float(recent["High"].max()), 2)
    result["recent_low"] = round(float(recent["Low"].min()), 2)

    # Trend direction
    if result["ema9"] and result["ema21"] and result["ema50"]:
        if ema9 > ema21 > ema50:
            result["trend"] = "strong_bullish"
        elif ema9 > ema21:
            result["trend"] = "bullish"
        elif ema9 < ema21 < ema50:
            result["trend"] = "strong_bearish"
        elif ema9 < ema21:
            result["trend"] = "bearish"
        else:
            result["trend"] = "sideways"
    elif result["ema_bullish"] is not None:
        result["trend"] = "bullish" if result["ema_bullish"] else "bearish"
    else:
        result["trend"] = "unknown"

    return result


def compute_score(indicators: dict, absorption: dict = None, scalper: dict = None) -> float:
    """Compute a 0-10 confluence score from indicators + advanced signals."""
    if "error" in indicators:
        return 0.0

    score = 5.0  # Start neutral

    # Trend alignment (+/- 2)
    trend = indicators.get("trend", "unknown")
    if trend == "strong_bullish":
        score += 2.0
    elif trend == "bullish":
        score += 1.0
    elif trend == "strong_bearish":
        score -= 2.0
    elif trend == "bearish":
        score -= 1.0

    # RSI (+/- 1.5)
    rsi = indicators.get("rsi")
    if rsi is not None:
        if 40 <= rsi <= 60:
            score += 0.5  # Neutral, room to move
        elif rsi < 30:
            score += 1.5  # Oversold = potential buy
        elif rsi > 70:
            score -= 1.5  # Overbought = caution

    # MACD alignment (+/- 1)
    if indicators.get("macd_bullish") is True:
        score += 1.0
    elif indicators.get("macd_bullish") is False:
        score -= 1.0

    # Volume confirmation (+/- 1)
    vol_ratio = indicators.get("volume_ratio")
    if vol_ratio is not None:
        if vol_ratio > 1.5:
            score += 1.0  # High volume confirms move
        elif vol_ratio < 0.5:
            score -= 0.5  # Low volume = weak move

    # Price vs EMAs (+/- 1)
    price = indicators.get("current_price", 0)
    ema9 = indicators.get("ema9")
    ema21 = indicators.get("ema21")
    if price and ema9 and ema21:
        if price > ema9 > ema21:
            score += 1.0
        elif price < ema9 < ema21:
            score -= 1.0

    # ── Absorption Bubbles bonus (+/- 0.75) ──
    if absorption and absorption.get("absorption_detected"):
        bias = absorption.get("signal_bias", "neutral")
        if bias == "bullish":
            score += 0.75  # Buying absorption = support forming
        elif bias == "bearish":
            score -= 0.75  # Selling absorption = resistance forming

    # ── Pro Scalper bonus (+/- 1.0) ──
    if scalper:
        scalper_signal = scalper.get("signal", "neutral")
        scalper_conf = scalper.get("confidence", 0)
        if scalper_signal == "buy" and scalper_conf >= 0.5:
            score += 1.0 * scalper_conf
        elif scalper_signal == "sell" and scalper_conf >= 0.5:
            score -= 1.0 * scalper_conf

        # Reversal signals
        reversal = scalper.get("reversal")
        if reversal == "bullish_reversal":
            score += 0.5
        elif reversal == "bearish_reversal":
            score -= 0.5

    # Clamp to 0-10
    return round(max(0.0, min(10.0, score)), 1)


def generate_signal(score: float, indicators: dict) -> str:
    """Generate GO/NO GO/WAIT signal from score."""
    if score >= 7.0:
        return "GO"
    elif score >= 5.0:
        return "WAIT"
    else:
        return "NO GO"


def scan_watchlist(tickers: list[str], period: str = "3mo", interval: str = "1d") -> list[dict]:
    """Scan a list of tickers and return scored results."""
    try:
        from .advanced_indicators import compute_absorption_bubbles, compute_pro_scalper
        has_advanced = True
    except Exception:
        has_advanced = False

    results = []

    for ticker in tickers:
        df = fetch_ticker_data(ticker, period=period, interval=interval)
        indicators = compute_indicators(df)

        if "error" in indicators:
            results.append({
                "ticker": ticker,
                "score": 0.0,
                "signal": "ERROR",
                "indicators": indicators,
                "absorption": {"absorption_detected": False, "events": [], "absorption_score": 0, "signal_bias": "neutral"},
                "scalper": {"signal": "neutral", "direction": "neutral", "zone": "neutral", "reversal": None, "confidence": 0},
            })
            continue

        # Compute advanced indicators (full OHLCV available)
        absorption = {"absorption_detected": False, "events": [], "absorption_score": 0, "signal_bias": "neutral"}
        scalper = {"signal": "neutral", "direction": "neutral", "zone": "neutral", "reversal": None, "confidence": 0}

        if has_advanced:
            try:
                absorption = compute_absorption_bubbles(df)
                scalper = compute_pro_scalper(df)
            except Exception:
                pass

        score = compute_score(indicators, absorption=absorption, scalper=scalper)
        signal = generate_signal(score, indicators)

        results.append({
            "ticker": ticker,
            "score": score,
            "signal": signal,
            "indicators": indicators,
            "absorption": absorption,
            "scalper": scalper,
        })

    # Sort by score descending
    results.sort(key=lambda x: x["score"], reverse=True)
    return results