backend/engine.py

"""Anomaly scoring engine.

Produces a composite per-stock anomaly score by combining independent signals,
each expressed as a z-score relative to the symbol's own rolling history and its
sector peers. Scores are time-decayed so old moves fade.

Signals:
  - VOLATILITY_Z  : realized vol vs 20d rolling (regime spikes)
  - VOLUME_Z      : volume vs 20d median (liquidity surge)
  - BREAKOUT_Z    : distance from 20d/50d high-low range (momentum/breakout)
  - CORRELATION_Z : decoupling from sector peers (isolated mover)
  - SPREAD_Z      : bid-ask asymmetry (market-maker stress)

Composite = sum(weight_i * clip(z_i)) * time_decay
Weights are default, but sector-aware baseline volatility normalizes thresholds.
"""
import datetime as dt
import math
import json
import numpy as np
from database import get_conn

DEFAULT_WEIGHTS = {
    "return": 0.28,
    "volatility": 0.22,
    "volume": 0.22,
    "breakout": 0.18,
    "correlation": 0.10,
    "spread": 0.00,
}

# Sector baseline vol multipliers (tech ~2x utilities in reality; simplify).
SECTOR_VOL = {
    "Technology": 1.4, "Communication": 1.3, "Industrials": 1.1,
    "Consumer": 0.8, "Financials": 1.0, "Healthcare": 0.7, "Energy": 1.0,
}

SECTOR_NAME = {
    "Technology": "Tech", "Communication": "Comm", "Industrials": "Industrials",
    "Consumer": "Consumer", "Financials": "Financials",
    "Healthcare": "Healthcare", "Energy": "Energy",
}

# Decay: half-life of 12h for the alert importance.
DECAY_HALFLIFE_H = 12.0


def decay_multiplier(ts: dt.datetime, now: dt.datetime | None = None) -> float:
    now = now or dt.datetime.now(dt.timezone.utc)
    if ts.tzinfo is None:
        ts = ts.replace(tzinfo=dt.timezone.utc)
    age_h = max(0.0, (now - ts).total_seconds() / 3600.0)
    return 0.5 ** (age_h / DECAY_HALFLIFE_H)


def zscore(x, mean, std):
    if std is None or not math.isfinite(std) or std <= 1e-12:
        return 0.0
    return (x - mean) / std


def _clip(z, lo=-3.0, hi=3.0):
    return max(lo, min(hi, z))


def wilson_conf(p, n):
    """Wilson score confidence (for volume-confidence of signals)."""
    if n <= 0:
        return 0.0
    z = 1.96
    denom = 1 + z * z / n
    centre = (p + z * z / (2 * n)) / denom
    margin = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / denom
    return centre - margin


async def latest_quotes(symbols=None, now=None):
    """Return latest quote per symbol plus a short history for scoring."""
    conn = await get_conn()
    rows = None
    if symbols:
        placeholders = ",".join("?" for _ in symbols)
        cur = await conn.execute(
            f"SELECT symbol, ts, price, volume FROM quotes "
            f"WHERE symbol IN ({placeholders}) ORDER BY ts DESC",
            symbols,
        )
        rows = await cur.fetchall()
    else:
        cur = await conn.execute(
            "SELECT symbol, ts, price, volume FROM quotes ORDER BY ts DESC"
        )
        rows = await cur.fetchall()
    await conn.close()
    return rows


async def fetch_history(symbol, limit=200):
    conn = await get_conn()
    cur = await conn.execute(
        "SELECT ts, price, volume FROM quotes WHERE symbol=? ORDER BY ts DESC LIMIT ?",
        (symbol, limit),
    )
    rows = await cur.fetchall()
    await conn.close()
    return list(reversed(rows))


async def compute_stock_score(symbol, now=None):
    """Compute composite + per-signal z-scores for a single symbol."""
    now = now or dt.datetime.now(dt.timezone.utc)
    hist = await fetch_history(symbol, 200)
    if len(hist) < 30:
        return None

    prices = np.array([r["price"] for r in hist], dtype=float)
    volumes = np.array([r["volume"] for r in hist], dtype=float)
    ts = [r["ts"] for r in hist]
    dur_h = max(1.0, (now - _iso(ts[-1])).total_seconds() / 3600.0)

    # --- volatility signal -------------------------------------------------
    rets = np.diff(prices)
    log_rets = np.log(prices[1:] / prices[:-1])
    # realized vol last window vs prior
    look = min(20, len(log_rets))
    recent_vol = np.std(log_rets[-look:])
    prev_vol = np.std(log_rets[-2 * look: -look]) if len(log_rets) > 2 * look else recent_vol
    vol_mean = max(prev_vol, 1e-6)
    vol_std = max(vol_mean * 0.5, 1e-6)
    vol_z = zscore(recent_vol, vol_mean, vol_std)

    # --- return magnitude signal -------------------------------------------
    # Directly score the latest move vs the stock's own history of moves,
    # expressed as a z-score. This is the most intuitive definition of "something
    # happened today" and is not diluted by the (already high) volatility context.
    if len(log_rets) > 10:
        ret_mean = float(np.mean(log_rets[-40:])) if len(log_rets) >= 40 else float(np.mean(log_rets))
        ret_std = float(np.std(log_rets[-40:])) if len(log_rets) >= 40 else float(np.std(log_rets))
        latest_ret = log_rets[-1]
        return_z = zscore(latest_ret, ret_mean, ret_std)
    else:
        latest_ret = 0.0
        return_z = 0.0

    # --- volume signal ------------------------------------------------------
    vol_recent = volumes[-5:].mean()
    vol_median = np.median(volumes[-60:]) if len(volumes) >= 20 else volumes.mean()
    vol_base = max(vol_median, 1e-6)
    vol_ratio = vol_recent / vol_base
    vol_std = max(vol_base * 0.5, 1e-6)
    vol_z = zscore(vol_recent, vol_base, vol_std)

    # --- breakout signal -----------------------------------------------------
    hi20 = prices[-20:].max() if len(prices) >= 20 else prices.max()
    lo20 = prices[-20:].min() if len(prices) >= 20 else prices.min()
    rng = max(hi20 - lo20, 1e-6)
    latest = prices[-1]
    # distance from midpoint normalized by range; +1 = at high, -1 = at low
    breakout_ratio = (latest - (hi20 + lo20) / 2) / (rng / 2)
    historical_median = (hi20 + lo20) / 2
    hist_std = np.std(prices[-20:]) if len(prices) >= 20 else 1.0
    breakout_z = zscore(latest, historical_median, max(hist_std, 1e-6))

    signals = {
        "return": _clip(return_z),
        "volatility": _clip(vol_z),
        "volume": _clip(vol_z),
        "breakout": _clip(breakout_z),
        "spread": 0.0,
    }

    composite = 0.0
    for key, w in DEFAULT_WEIGHTS.items():
        composite += w * max(0.0, abs(signals.get(key, 0.0)))

    # normalize by sector baseline vol so same threshold isn't applied to all
    conn = await get_conn()
    cur = await conn.execute("SELECT sector FROM stocks WHERE symbol=?", (symbol,))
    srow = await cur.fetchone()
    await conn.close()
    sector = srow["sector"] if srow else "Technology"
    sector_norm = SECTOR_VOL.get(sector, 1.0)
    # Higher sector vol => same absolute move is less unusual => scale down
    composite_norm = composite / (sector_norm ** 0.7)
    composite_norm = max(0.0, min(10.0, composite_norm))

    return {
        "symbol": symbol,
        "sector": sector,
        "sector_short": SECTOR_NAME.get(sector, sector),
        "price": float(latest),
        "change_pct": float((prices[-1] / prices[-2] - 1) * 100) if len(prices) > 1 else 0.0,
        "volume_ratio": float(vol_ratio),
        "signals": signals,
        "signal_values": {
            "return_z": float(return_z), "vol_z": float(vol_z),
            "volume_z": float(vol_z), "breakout_z": float(breakout_z),
        },
        "composite": round(float(composite_norm), 3),
        "decay": round(decay_multiplier(_iso(ts[-1]), now), 3),
        "freshness_sec": round((now - _iso(ts[-1])).total_seconds()),
        "series_prices": [float(p) for p in prices[-24:]],
        # Relative-strength context (trend regime, not just today's move)
        "rs_20d": round(float((prices[-1] / prices[-20] - 1) * 100), 2)
                  if len(prices) >= 20 and prices[-20] > 0 else round(float((prices[-1] / prices[0] - 1) * 100), 2),
        "high_52w": float(prices.max()),
        "low_52w": float(prices.min()),
        "pos_in_52w": round(float((prices[-1] - prices.min()) / max(prices.max() - prices.min(), 1e-6) * 100), 1),
    }


async def sector_movers(symbols, now=None):
    """Compute scores for all symbols, then attach sector peer medians &
    correlation decoupling so relative moves are shown, not absolute only."""
    scores = []
    for s in symbols:
        sc = await compute_stock_score(s, now)
        if sc:
            scores.append(sc)
    # group by sector
    by_sector = {}
    for sc in scores:
        by_sector.setdefault(sc["sector"], []).append(sc)
    for sc in scores:
        peers = by_sector.get(sc["sector"], [sc])
        peer_median = float(np.median([p["change_pct"] for p in peers]))
        peer_vol = float(np.median([abs(p["composite"]) for p in peers]))
        sc["peer_median_change"] = peer_median
        sc["relative_change"] = round(sc["change_pct"] - peer_median, 2)
        # Correlation/demand signal: isolated mover if |relative| big vs peers
        corr_std = max(np.std([p["change_pct"] for p in peers]) if len(peers) > 1 else 1.0, 1e-6)
        sc["correlation_z"] = _clip(zscore(sc["change_pct"], peer_median, corr_std))
        sc["signals"]["correlation"] = sc["correlation_z"]
        # Recompute composite including correlation
        composite = 0.0
        for key, w in DEFAULT_WEIGHTS.items():
            composite += w * max(0.0, abs(sc["signals"].get(key, 0.0)))
        sector_norm = SECTOR_VOL.get(sc["sector"], 1.0)
        sc["composite"] = round(max(0.0, min(10.0, composite / (sector_norm ** 0.7))), 3)
    return scores


def traffic_light(composite: float) -> str:
    if composite >= 1.8:
        return "unusual"
    if composite >= 0.8:
        return "watch"
    return "normal"


def _iso(ts):
    if isinstance(ts, dt.datetime):
        return ts
    try:
        return dt.datetime.fromisoformat(ts)
    except (TypeError, ValueError):
        return dt.datetime.now(dt.timezone.utc)