rmi-backend/app/velocity_risk.py

"""
Velocity Risk Engine - Time-Series Anomaly Detection
=====================================================

Tracks how token metrics CHANGE over time, not just what they ARE.
Fast changes = higher risk than static bad metrics.

Premium feature: Catches rugs mid-flight, not just at launch.
"""

import logging
import time
from collections import defaultdict
from typing import Any

logger = logging.getLogger("sentinel.velocity")

# In-memory time-series cache (backed by Redis in production)
# { "chain:address": [(timestamp, {metrics}), ...] }
_series_cache: dict[str, list] = defaultdict(list)
MAX_SERIES_LENGTH = 10  # Keep last 10 snapshots per token


def record_snapshot(chain: str, address: str, metrics: dict[str, float]) -> None:
    """Record a metrics snapshot for velocity tracking."""
    key = f"{chain}:{address.lower()}"
    _series_cache[key].append((time.time(), metrics))
    if len(_series_cache[key]) > MAX_SERIES_LENGTH:
        _series_cache[key] = _series_cache[key][-MAX_SERIES_LENGTH:]


def analyze_velocity(chain: str, address: str, current: dict[str, float], window_seconds: int = 3600) -> dict[str, Any]:
    """Analyze how fast metrics are changing. Returns velocity scores and flags.

    Metrics tracked:
    - holder_concentration: top10% change per hour
    - lp_depth: liquidity depth change per hour
    - volume_liquidity_ratio: vol/liq ratio acceleration
    - price: price change per hour
    - tx_count: transaction velocity
    """
    key = f"{chain}:{address.lower()}"
    history = _series_cache.get(key, [])

    # Record current
    record_snapshot(chain, address, current)

    if len(history) < 2:
        return {"status": "insufficient_data", "snapshots": len(history) + 1}

    # Find snapshots within window
    now = time.time()
    window_snapshots = [(ts, m) for ts, m in history if now - ts <= window_seconds]

    if len(window_snapshots) < 2:
        return {"status": "insufficient_window_data", "snapshots": len(window_snapshots) + 1}

    oldest = window_snapshots[0][1]
    newest = current
    time_span = now - window_snapshots[0][0]
    hours = max(time_span / 3600, 0.01)

    velocities = {}
    flags = []
    risk_score = 0

    # Holder concentration velocity
    if "top10_pct" in oldest and "top10_pct" in newest:
        holder_delta = newest["top10_pct"] - oldest["top10_pct"]
        holder_velocity = holder_delta / hours
        velocities["holder_concentration_delta_per_hour"] = round(holder_velocity, 2)

        if holder_velocity > 20:
            flags.append("HOLDER_CONCENTRATION_SURGING")
            risk_score += 25
        elif holder_velocity > 10:
            flags.append("HOLDER_CONCENTRATION_RISING_FAST")
            risk_score += 15
        elif holder_velocity > 5:
            flags.append("HOLDER_CONCENTRATION_RISING")
            risk_score += 8

    # LP depth velocity
    if "liquidity_usd" in oldest and "liquidity_usd" in newest:
        old_liq = max(oldest["liquidity_usd"], 1)
        new_liq = max(newest["liquidity_usd"], 1)
        lp_delta_pct = ((new_liq - old_liq) / old_liq) * 100
        lp_velocity = lp_delta_pct / hours
        velocities["lp_depth_change_pct_per_hour"] = round(lp_velocity, 2)

        if lp_velocity < -30:
            flags.append("LP_DRAINING_RAPIDLY")
            risk_score += 30
        elif lp_velocity < -15:
            flags.append("LP_DECREASING_FAST")
            risk_score += 20
        elif lp_velocity < -5:
            flags.append("LP_DECREASING")
            risk_score += 10

    # Volume/liquidity ratio acceleration
    if all(k in oldest and k in newest for k in ["volume_24h", "liquidity_usd"]):
        old_ratio = oldest["volume_24h"] / max(oldest["liquidity_usd"], 1)
        new_ratio = newest["volume_24h"] / max(newest["liquidity_usd"], 1)
        ratio_delta = new_ratio - old_ratio
        velocities["volume_liquidity_ratio_change"] = round(ratio_delta, 3)

        if ratio_delta > 10:
            flags.append("VOLUME_SPIKE_VS_LIQUIDITY")
            risk_score += 15
        elif ratio_delta > 5:
            flags.append("VOLUME_RISING_VS_LIQUIDITY")
            risk_score += 8

    # Price velocity
    if "price_usd" in oldest and "price_usd" in newest:
        old_price = max(oldest["price_usd"], 0.000001)
        new_price = max(newest["price_usd"], 0.000001)
        price_delta_pct = ((new_price - old_price) / old_price) * 100
        price_velocity = price_delta_pct / hours
        velocities["price_change_pct_per_hour"] = round(price_velocity, 2)

        if price_velocity < -50:
            flags.append("PRICE_CRASHING")
            risk_score += 20
        elif price_velocity > 500:
            flags.append("PRICE_PUMPING_ABNORMALLY")
            risk_score += 12

    # Tx count velocity
    if "tx_count" in oldest and "tx_count" in newest:
        tx_delta = newest["tx_count"] - oldest["tx_count"]
        tx_velocity = tx_delta / hours
        velocities["tx_count_change_per_hour"] = round(tx_velocity, 1)

        if tx_velocity > 1000:
            flags.append("TX_VOLUME_SURGING")
            risk_score += 10

    return {
        "status": "ok",
        "snapshots": len(window_snapshots) + 1,
        "time_window_hours": round(hours, 2),
        "velocities": velocities,
        "flags": flags,
        "velocity_risk_score": min(risk_score, 100),
        "risk_level": "critical"
        if risk_score > 60
        else "high"
        if risk_score > 30
        else "medium"
        if risk_score > 10
        else "low",
    }


def get_market_context() -> dict[str, Any]:
    """Get current market conditions for contextualized scoring."""
    try:
        import asyncio

        import httpx

        async def _fetch():
            async with httpx.AsyncClient(timeout=5) as c:
                resp = await c.get("https://api.alternative.me/fng/?limit=1")
                if resp.status_code == 200:
                    data = resp.json()
                    fng = data.get("data", [{}])[0]
                    return {
                        "fear_greed_value": int(fng.get("value", 50)),
                        "fear_greed_classification": fng.get("value_classification", "Neutral"),
                        "timestamp": fng.get("timestamp", ""),
                    }
                return {"fear_greed_value": 50, "fear_greed_classification": "Unknown"}

        return asyncio.run(_fetch())
    except Exception as e:
        logger.warning(f"Failed to fetch market context: {e}")
        return {"fear_greed_value": 50, "fear_greed_classification": "Unknown"}


def contextualize_score(base_safety: int, market_context: dict[str, Any]) -> dict[str, Any]:
    """Adjust safety score based on market conditions.

    During Extreme Greed (>75): scammers launch more - raise sensitivity
    During Extreme Fear (<25): fewer scams launch - lower sensitivity
    """
    fng = market_context.get("fear_greed_value", 50)

    # Market pressure: how much to adjust
    if fng > 75:  # Extreme Greed - more scams
        pressure = 1.3
        context = "Scam alert elevated - market in Extreme Greed"
    elif fng > 60:  # Greed
        pressure = 1.15
        context = "Elevated scam risk - market in Greed"
    elif fng < 25:  # Extreme Fear - fewer new scams
        pressure = 0.85
        context = "Reduced scam activity - market in Extreme Fear"
    elif fng < 40:  # Fear
        pressure = 0.95
        context = "Slightly reduced risk - market in Fear"
    else:  # Neutral
        pressure = 1.0
        context = "Normal market conditions"

    adjusted = max(0, min(100, round(base_safety / pressure)))

    return {
        "base_safety": base_safety,
        "contextualized_safety": adjusted,
        "market_pressure": round(pressure, 2),
        "fear_greed": fng,
        "classification": market_context.get("fear_greed_classification", "Neutral"),
        "context": context,
    }