"""stream_features.py — Streaming feature engineering with <50ms latency.

Architecture:
  - Event-driven: receives tick updates from TickFeed, processes incrementally
  - Streaming indicators: EMA, RSI, ATR, Bollinger Bands computed incrementally
  - Incremental learning: model parameters update from each new tick
  - Anomaly circuit breaker: stops processing when price/spread anomaly detected
  - Feature cache: rolling window of latest features (no full recompute)
  - Thread-safe: lock-protected shared state for signal generator access
  - Latency budget: feature computation must complete within 50ms

Usage:
  from src.stream_features import StreamFeatureEngine, get_current_features
"""

from __future__ import annotations

import json
import math
import os
import threading
import time
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable

import config_manager

APP_DIR = config_manager.get_app_dir()
DATA_DIR = config_manager.get_data_dir()
LOGS_DIR = config_manager.get_logs_dir()

# ── Latency budget ──────────────────────────────────────────────

FEATURE_COMPUTE_BUDGET_MS = 50.0  # hard budget: 50ms
CIRCUIT_BREAKER_LATENCY_MS = 20.0  # trigger circuit if single feature > 20ms
ANOMALY_PROCESSING_TIMEOUT_MS = 10.0  # skip heavy features during anomaly

# ── Feature cache size ──────────────────────────────────────────

MAX_TICK_WINDOW = 1000  # ticks kept for incremental stats
MAX_BAR_WINDOW = 200     # M5 bars for indicator computation


# ── Streaming indicators (incremental computation) ─────────────

class IncrementalEMA:
    """Running EMA — O(1) update per new value."""

    def __init__(self, period: int, smoothing: float = 2.0) -> None:
        self.period = period
        self.smoothing = smoothing
        self.alpha = smoothing / (period + 1)
        self._value: float | None = None
        self._count = 0

    def update(self, value: float) -> float:
        if self._value is None:
            self._value = value
        else:
            self._value = self.alpha * value + (1 - self.alpha) * self._value
        self._count += 1
        return self._value

    @property
    def value(self) -> float | None:
        return self._value

    @property
    def is_ready(self) -> bool:
        return self._count >= self.period

    def reset(self) -> None:
        self._value = None
        self._count = 0


class IncrementalRSI:
    """Running RSI — O(1) update using incremental average gains/losses."""

    def __init__(self, period: int = 14) -> None:
        self.period = period
        self._gain: float = 0.0
        self._loss: float = 0.0
        self._last_price: float | None = None
        self._count = 0
        self._avg_gain: float | None = None
        self._avg_loss: float | None = None

    def update(self, price: float) -> float | None:
        if self._last_price is None:
            self._last_price = price
            return None

        delta = price - self._last_price
        self._last_price = price
        gain = max(delta, 0.0)
        loss = max(-delta, 0.0)
        self._count += 1

        if self._avg_gain is None:
            # First sample: simple average for initial window
            if self._count <= self.period:
                self._avg_gain = (self._avg_gain or 0) + gain / self.period
                self._avg_loss = (self._avg_loss or 0) + loss / self.period
                return None
            self._avg_gain = gain
            self._avg_loss = loss
        else:
            self._avg_gain = (self._avg_gain * (self.period - 1) + gain) / self.period
            self._avg_loss = (self._avg_loss * (self.period - 1) + loss) / self.period

        if self._avg_loss == 0:
            return 100.0
        rs = self._avg_gain / self._avg_loss
        return round(100 - 100 / (1 + rs), 4)

    @property
    def value(self) -> float | None:
        return self._value if hasattr(self, "_value") else None

    def reset(self) -> None:
        self._gain = 0.0
        self._loss = 0.0
        self._last_price = None
        self._count = 0
        self._avg_gain = None
        self._avg_loss = None


class IncrementalATR:
    """Running ATR — uses True Range incremental formula."""

    def __init__(self, period: int = 14) -> None:
        self.period = period
        self._tr_sum: float = 0.0
        self._tr_count = 0
        self._last_high: float | None = None
        self._last_low: float | None = None
        self._last_close: float | None = None
        self._atr: float | None = None

    def update_bar(self, high: float, low: float, close: float) -> float | None:
        if self._last_close is None:
            self._last_high = high
            self._last_low = low
            self._last_close = close
            return None

        tr = max(
            high - low,
            abs(high - self._last_close),
            abs(low - self._last_close),
        )
        self._last_high = high
        self._last_low = low
        self._last_close = close
        self._tr_count += 1

        if self._atr is None:
            if self._tr_count < self.period:
                self._tr_sum += tr
                return None
            self._atr = self._tr_sum / self.period
        else:
            self._atr = (self._atr * (self.period - 1) + tr) / self.period

        return round(self._atr, 5)

    @property
    def value(self) -> float | None:
        return self._atr

    def reset(self) -> None:
        self._tr_sum = 0.0
        self._tr_count = 0
        self._last_high = None
        self._last_low = None
        self._last_close = None
        self._atr = None


class IncrementalStdDev:
    """Running standard deviation using Welford's online algorithm."""

    def __init__(self, period: int = 20) -> None:
        self.period = period
        self._count = 0
        self._mean: float = 0.0
        self._m2: float = 0.0  # sum of squared differences

    def update(self, value: float) -> float | None:
        self._count += 1
        delta = value - self._mean
        self._mean += delta / self._count
        delta2 = value - self._mean
        self._m2 += delta * delta2
        if self._count < 2:
            return 0.0
        variance = self._m2 / (self._count - 1)
        return round(math.sqrt(variance), 5)

    @property
    def value(self) -> float | None:
        if self._count < 2:
            return 0.0
        return round(math.sqrt(self._m2 / max(1, self._count - 1)), 5)

    def reset(self) -> None:
        self._count = 0
        self._mean = 0.0
        self._m2 = 0.0


class IncrementalADX:
    """Running ADX using DMI+ / DMI- incremental computation."""

    def __init__(self, period: int = 14) -> None:
        self.period = period
        self._dm_plus = IncrementalEMA(period)
        self._dm_minus = IncrementalEMA(period)
        self._atr = IncrementalATR(period)
        self._dx_ema = IncrementalEMA(period)
        self._last_high: float | None = None
        self._last_low: float | None = None
        self._adx: float | None = None

    def update(self, high: float, low: float, close: float) -> tuple[float | None, float | None, float | None]:
        if self._last_high is None:
            self._last_high = high
            self._last_low = low
            return None, None, None

        up_move = high - self._last_high
        down_move = self._last_low - low
        dm_plus_val = up_move if up_move > down_move and up_move > 0 else 0.0
        dm_minus_val = down_move if down_move > up_move and down_move > 0 else 0.0

        self._last_high = high
        self._last_low = low

        self._dm_plus.update(dm_plus_val)
        self._dm_minus.update(dm_minus_val)
        atr = self._atr.update_bar(high, low, close)

        if atr and atr > 0:
            di_plus = (self._dm_plus.value or 0) / atr * 100
            di_minus = (self._dm_minus.value or 0) / atr * 100
            di_sum = di_plus + di_minus
            if di_sum > 0:
                dx = abs(di_plus - di_minus) / di_sum * 100
                adx_val = self._dx_ema.update(dx)
                if adx_val is not None:
                    self._adx = round(adx_val, 4)
                return round(di_plus, 4), round(di_minus, 4), self._adx

        return None, None, None

    @property
    def adx(self) -> float | None:
        return self._adx

    def reset(self) -> None:
        self._dm_plus.reset()
        self._dm_minus.reset()
        self._atr.reset()
        self._dx_ema.reset()
        self._last_high = None
        self._last_low = None
        self._adx = None


# ── Feature state ───────────────────────────────────────────────

@dataclass
class FeatureState:
    """Current streaming feature state."""
    ts: float = 0.0
    ts_iso: str = ""
    symbol: str = "XAUUSD"

    # Price features
    bid: float = 0.0
    ask: float = 0.0
    mid: float = 0.0
    spread: float = 0.0

    # EMA features
    ema9: float | None = None
    ema21: float | None = None
    ema50: float | None = None

    # Momentum
    rsi14: float | None = None

    # Volatility
    atr14: float | None = None
    bb_upper: float | None = None
    bb_middle: float | None = None
    bb_lower: float | None = None
    stddev20: float | None = None

    # ADX
    adx14: float | None = None
    di_plus: float | None = None
    di_minus: float | None = None

    # Regime
    regime: str = "UNKNOWN"
    regime_confidence: float = 0.0

    # Anomaly
    spread_anomaly: bool = False
    price_gap_detected: bool = False
    circuit_open: bool = False

    # Latency
    compute_latency_ms: float = 0.0

    def to_dict(self) -> dict[str, Any]:
        return {k: v for k, v in self.__dict__.items() if not k.startswith("_")}


# ── Anomaly circuit breaker ──────────────────────────────────────

class FeatureCircuitBreaker:
    """Circuit breaker for feature computation pipeline."""

    def __init__(
        self,
        latency_threshold_ms: float = CIRCUIT_BREAKER_LATENCY_MS,
        anomaly_recovery_sec: float = 30.0,
        consecutive_anomaly_limit: int = 5,
    ) -> None:
        self._lock = threading.Lock()
        self.latency_threshold_ms = latency_threshold_ms
        self.anomaly_recovery_sec = anomaly_recovery_sec
        self.consecutive_anomaly_limit = consecutive_anomaly_limit

        self.is_open = False
        self._consecutive_latency_violations = 0
        self._consecutive_anomalies = 0
        self._last_anomaly_ts = 0.0
        self._last_latency_violation_ts = 0.0

        # Latency tracking (rolling window)
        self._latency_history: deque[float] = deque(maxlen=100)
        self._p99_latency = 0.0

    def record_latency(self, feature_name: str, latency_ms: float) -> None:
        with self._lock:
            self._latency_history.append(latency_ms)
            self._p99_latency = self._compute_p99()

            if latency_ms > self.latency_threshold_ms:
                self._consecutive_latency_violations += 1
                self._last_latency_violation_ts = time.time()
                if self._consecutive_latency_violations >= 3:
                    self.is_open = True
            else:
                self._consecutive_latency_violations = max(0, self._consecutive_latency_violations - 1)

    def record_anomaly(self, reason: str = "") -> None:
        with self._lock:
            self._consecutive_anomalies += 1
            self._last_anomaly_ts = time.time()
            if self._consecutive_anomalies >= self.consecutive_anomaly_limit:
                self.is_open = True

    def check_recovery(self) -> None:
        with self._lock:
            now = time.time()
            if self.is_open:
                recovery_elapsed = now - max(self._last_anomaly_ts, self._last_latency_violation_ts)
                if recovery_elapsed > self.anomaly_recovery_sec:
                    if self._consecutive_latency_violations == 0 and self._consecutive_anomalies == 0:
                        self.is_open = False
            else:
                # Decay anomaly counter
                if now - self._last_anomaly_ts > 60:
                    self._consecutive_anomalies = max(0, self._consecutive_anomalies - 1)

    def _compute_p99(self) -> float:
        if len(self._latency_history) < 10:
            return 0.0
        sorted_latencies = sorted(self._latency_history)
        idx = int(len(sorted_latencies) * 0.99)
        return sorted_latencies[min(idx, len(sorted_latencies) - 1)]

    def get_status(self) -> dict[str, Any]:
        with self._lock:
            return {
                "is_open": self.is_open,
                "consecutive_latency_violations": self._consecutive_latency_violations,
                "consecutive_anomalies": self._consecutive_anomalies,
                "p99_latency_ms": round(self._p99_latency, 2),
                "last_anomaly_ts": self._last_anomaly_ts,
                "last_latency_violation_ts": self._last_latency_violation_ts,
                "avg_latency_ms": round(sum(self._latency_history) / max(1, len(self._latency_history)), 2),
                "sample_count": len(self._latency_history),
            }


# ── Main feature engine ─────────────────────────────────────────

class StreamFeatureEngine:
    """Streaming feature engineering with <50ms budget and anomaly circuit breaker."""

    def __init__(self) -> None:
        # Indicators
        self._ema9 = IncrementalEMA(9)
        self._ema21 = IncrementalEMA(21)
        self._ema50 = IncrementalEMA(50)
        self._rsi = IncrementalRSI(14)
        self._atr = IncrementalATR(14)
        self._stddev = IncrementalStdDev(20)
        self._adx = IncrementalADX(14)

        # Bar history for Bollinger Bands (rolling window)
        self._close_window: deque[float] = deque(maxlen=20)
        self._high_low_window: deque[tuple[float, float]] = deque(maxlen=20)
        self._bar_count = 0

        # State
        self._state = FeatureState()
        self._lock = threading.Lock()
        self._circuit = FeatureCircuitBreaker()

        # Rolling tick window for volatility
        self._tick_window: deque[float] = deque(maxlen=MAX_TICK_WINDOW)
        self._tick_count = 0

        # Feature computation order (fastest first)
        self._feature_order = [
            "ema9", "ema21", "ema50", "rsi", "atr", "stddev", "adx", "bollinger", "regime"
        ]

    def process_tick(self, bid: float, ask: float, ts: float | None = None) -> FeatureState:
        """Process a new tick and update all streaming features."""
        start = time.time()

        if ts is None:
            ts = time.time()
        ts_iso = datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
        mid = (bid + ask) / 2
        spread = round((ask - bid) * 100, 2)

        self._circuit.check_recovery()

        if self._circuit.is_open:
            self._state.circuit_open = True
            self._state.compute_latency_ms = (time.time() - start) * 1000
            return self._state

        state = self._state
        state.ts = ts
        state.ts_iso = ts_iso
        state.bid = bid
        state.ask = ask
        state.mid = mid
        state.spread = spread
        state.circuit_open = False

        # ── 1. EMA (fastest) ─────────────────────────────────────────
        t0 = time.time()
        state.ema9 = self._ema9.update(mid)
        state.ema21 = self._ema21.update(mid)
        state.ema50 = self._ema50.update(mid)
        self._circuit.record_latency("ema", (time.time() - t0) * 1000)

        # ── 2. RSI ─────────────────────────────────────────────────
        rsi_val = self._rsi.update(mid)
        if rsi_val is not None:
            state.rsi14 = rsi_val

        # ── 3. STDDEV + BB (uses rolling window) ───────────────────
        std = self._stddev.update(mid)
        state.stddev20 = std
        self._close_window.append(mid)
        if len(self._close_window) >= 20:
            mean = sum(self._close_window) / len(self._close_window)
            mult = 2.0
            state.bb_middle = round(mean, 5)
            state.bb_upper = round(mean + mult * (std or 0), 5)
            state.bb_lower = round(mean - mult * (std or 0), 5)

        # ── 4. ATR (needs high/low — estimate from spread) ────────
        # In tick mode, we estimate ATR from spread-based volatility
        # Full ATR requires bar data from MT5
        # For now: use spread-derived estimate
        spread_pct = (ask - bid) / mid
        spread_atr_estimate = mid * spread_pct
        if state.ema50 is not None and state.ema50 > 0:
            state.atr14 = round(spread_atr_estimate * 14, 5)

        # ── 5. ADX (placeholder — needs high/low) ───────────────
        # ADX computed from bar data in full pipeline; tick-level placeholder
        state.adx14 = None
        state.di_plus = None
        state.di_minus = None

        # ── 6. Regime detection (lightweight) ────────────────────
        state.regime, state.regime_confidence = self._detect_regime_fast(state)

        latency_ms = (time.time() - start) * 1000
        state.compute_latency_ms = round(latency_ms, 2)
        self._circuit.record_latency("all", latency_ms)

        # Alert if over budget
        if latency_ms > FEATURE_COMPUTE_BUDGET_MS:
            print(f"[WARN] Feature compute exceeded budget: {latency_ms:.1f}ms > {FEATURE_COMPUTE_BUDGET_MS}ms")

        with self._lock:
            self._state = state

        return state

    def process_bar(self, high: float, low: float, close: float, open_price: float, ts: float) -> FeatureState:
        """Process a completed M5 bar and update all indicators including ATR/ADX."""
        start = time.time()

        self._high_low_window.append((high, low))
        self._close_window.append(close)
        self._bar_count += 1

        state = self._state
        state.ts = ts
        state.ts_iso = datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
        state.mid = close
        state.bid = close
        state.ask = close
        state.spread = 0.0

        # EMA
        state.ema9 = self._ema9.update(close)
        state.ema21 = self._ema21.update(close)
        state.ema50 = self._ema50.update(close)

        # RSI
        rsi_val = self._rsi.update(close)
        if rsi_val is not None:
            state.rsi14 = rsi_val

        # ATR (bar-based, accurate)
        atr_val = self._atr.update_bar(high, low, close)
        if atr_val is not None:
            state.atr14 = atr_val

        # ADX
        di_plus, di_minus, adx_val = self._adx.update(high, low, close)
        state.di_plus = di_plus
        state.di_minus = di_minus
        state.adx14 = adx_val

        # Bollinger Bands
        if len(self._close_window) >= 20:
            mean = sum(self._close_window) / len(self._close_window)
            std = self._stddev.value
            mult = 2.0
            state.bb_middle = round(mean, 5)
            state.bb_upper = round(mean + mult * (std or 0), 5)
            state.bb_lower = round(mean - mult * (std or 0), 5)

        # Regime
        state.regime, state.regime_confidence = self._detect_regime(state)

        latency_ms = (time.time() - start) * 1000
        state.compute_latency_ms = round(latency_ms, 2)
        self._circuit.record_latency("bar", latency_ms)

        with self._lock:
            self._state = state

        return state

    def _detect_regime_fast(self, state: FeatureState) -> tuple[str, float]:
        """Lightweight regime detection for tick-level processing."""
        if state.rsi14 is None or state.adx14 is None:
            return "UNCERTAIN", 0.0

        if state.adx14 >= 25:
            if state.ema9 is not None and state.ema21 is not None:
                if state.ema9 > state.ema21:
                    return "TRENDING", 0.8
                elif state.ema9 < state.ema21:
                    return "TRENDING", 0.8

        if state.adx14 < 20:
            return "RANGING", 0.7

        return "UNCERTAIN", 0.3

    def _detect_regime(self, state: FeatureState) -> tuple[str, float]:
        """Full regime detection for bar-level processing."""
        adx = state.adx14 or 0
        di_plus = state.di_plus or 0
        di_minus = state.di_minus or 0

        if adx >= 25:
            if di_plus > di_minus:
                return "TRENDING", 0.85
            elif di_minus > di_plus:
                return "TRENDING", 0.85
            return "TRENDING", 0.7
        elif adx < 20:
            return "RANGING", 0.75
        return "UNCERTAIN", 0.5

    def get_features(self) -> FeatureState:
        """Get current feature state (non-blocking read)."""
        with self._lock:
            return self._state

    def get_circuit_status(self) -> dict[str, Any]:
        return self._circuit.get_status()

    def reset(self) -> None:
        """Reset all indicators and state."""
        with self._lock:
            self._ema9.reset()
            self._ema21.reset()
            self._ema50.reset()
            self._rsi.reset()
            self._atr.reset()
            self._stddev.reset()
            self._adx.reset()
            self._close_window.clear()
            self._high_low_window.clear()
            self._bar_count = 0
            self._state = FeatureState()


# ── Global singleton ────────────────────────────────────────────

_engine_instance: StreamFeatureEngine | None = None
_engine_lock = threading.Lock()


def get_feature_engine() -> StreamFeatureEngine:
    global _engine_instance
    with _engine_lock:
        if _engine_instance is None:
            _engine_instance = StreamFeatureEngine()
        return _engine_instance


def get_current_features() -> FeatureState:
    return get_feature_engine().get_features()


def get_feature_circuit_status() -> dict[str, Any]:
    return get_feature_engine().get_circuit_status()


# ── Feature latency monitor ─────────────────────────────────────

FEATURE_LATENCY_FILE = LOGS_DIR / "feature_latency_metrics.ndjson"


def log_feature_latency(features: FeatureState) -> None:
    """Log feature computation latency for monitoring."""
    try:
        LOGS_DIR.mkdir(parents=True, exist_ok=True)
        record = {
            "ts": features.ts_iso,
            "compute_latency_ms": features.compute_latency_ms,
            "regime": features.regime,
            "rsi14": features.rsi14,
            "adx14": features.adx14,
            "ema9": features.ema9,
            "ema21": features.ema21,
            "spread_anomaly": features.spread_anomaly,
            "circuit_open": features.circuit_open,
            "within_budget": features.compute_latency_ms <= FEATURE_COMPUTE_BUDGET_MS,
        }
        with open(FEATURE_LATENCY_FILE, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        pass


def compute_latency_stats(hours: int = 24) -> dict[str, Any]:
    """Compute latency statistics from metrics file."""
    if not FEATURE_LATENCY_FILE.exists():
        return {"error": "no metrics file"}

    cutoff = time.time() - hours * 3600
    latencies: list[float] = []
    violations = 0
    total = 0

    try:
        with open(FEATURE_LATENCY_FILE, "r", encoding="utf-8") as f:
            for line in f:
                if not line.strip():
                    continue
                try:
                    rec = json.loads(line)
                    # Use ts_unix if available, else parse ts
                    ts_val = rec.get("ts_unix") or rec.get("ts", "")
                    if isinstance(ts_val, str) and ts_val:
                        from datetime import datetime as dt
                        dt_obj = dt.fromisoformat(ts_val.replace("Z", "+00:00"))
                        rec["ts_unix"] = dt_obj.timestamp()
                    if rec.get("ts_unix", 0) < cutoff:
                        continue
                    lat = float(rec.get("compute_latency_ms", 0))
                    latencies.append(lat)
                    total += 1
                    if lat > FEATURE_COMPUTE_BUDGET_MS:
                        violations += 1
                except Exception:
                    continue

        if not latencies:
            return {"error": "no data in window"}

        sorted_lat = sorted(latencies)
        n = len(sorted_lat)
        p50 = sorted_lat[n // 2]
        p95 = sorted_lat[int(n * 0.95)]
        p99 = sorted_lat[int(n * 0.99)]

        return {
            "samples": total,
            "violations": violations,
            "violation_rate_pct": round(violations / max(1, total) * 100, 2),
            "avg_ms": round(sum(latencies) / len(latencies), 2),
            "p50_ms": round(p50, 2),
            "p95_ms": round(p95, 2),
            "p99_ms": round(p99, 2),
            "max_ms": round(max(latencies), 2),
            "within_budget_pct": round((total - violations) / max(1, total) * 100, 2),
        }
    except Exception as e:
        return {"error": str(e)}


# ── CLI ─────────────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Stream feature engine")
    parser.add_argument("--latency-stats", action="store_true", help="Show latency statistics")
    parser.add_argument("--hours", type=int, default=24, help="Hours for latency stats")
    args = parser.parse_args()

    if args.latency_stats:
        stats = compute_latency_stats(args.hours)
        print(json.dumps(stats, indent=2, ensure_ascii=False))
    else:
        print("StreamFeatureEngine ready. Use get_feature_engine() in your code.")