"""tick_feed.py — Multi-source low-latency tick ingestion with cross-validation.

Architecture:
  1. Polls 3 authoritative sources in parallel (MT5, Swissquote, OANDA)
  2. Per-source HealthTracker with exponential decay reliability scoring
  3. Cross-validation: majority vote on bid/ask (+ outlier filter via IQR)
  4. Composite quote: best-bid / best-ask from highest-reliable source
  5. Anomaly detection: spread spike, price gap, stale quote → circuit breaker
  6. Shared state via threading.Lock for millisecond-safe reads
  7. Optional ZMQ pub for zero-copy inter-thread communication

Usage:
  from src.tick_feed import TickFeed, get_composite_quote, get_feed_status

  feed = TickFeed(max_sources=3, poll_interval_ms=50)
  feed.start()
  # Later in signal pipeline:
  quote = get_composite_quote()  # non-blocking read
"""

from __future__ import annotations

import gzip
import hashlib
import json
import math
import os
import random
import signal
import socket
import struct
import threading
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
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()
TICK_DIR = DATA_DIR / "ticks"

# ── Source configuration ─────────────────────────────────────────

SOURCES = {
    "mt5": {
        "enabled": os.environ.get("TICK_SOURCE_MT5", "true").lower() == "true",
        "poll_interval_ms": int(os.environ.get("TICK_POLL_MS_MT5", "50")),
        "priority": 1,
        "timeout_ms": 200,
    },
    "swissquote": {
        "enabled": os.environ.get("TICK_SOURCE_SWISSQUOTE", "true").lower() == "true",
        "poll_interval_ms": int(os.environ.get("TICK_POLL_MS_SQ", "100")),
        "priority": 2,
        "timeout_ms": 500,
        "url": "https://forex-data-feed.swissquote.com/public-quotes/bboquotes/instrument/XAU/USD",
    },
    "oanda": {
        "enabled": os.environ.get("TICK_SOURCE_OANDA", "false").lower() == "true",
        "poll_interval_ms": int(os.environ.get("TICK_POLL_MS_OANDA", "200")),
        "priority": 3,
        "timeout_ms": 1000,
        "account_id": os.environ.get("OANDA_ACCOUNT_ID", ""),
        "api_key": os.environ.get("OANDA_API_KEY", ""),
    },
}

# ── Reliability scoring ──────────────────────────────────────────

RELIABILITY_DECAY = 0.95  # per failed ping
RELIABILITY_BOOST = 0.02  # per successful ping
RELIABILITY_MAX = 1.0
RELIABILITY_MIN = 0.0
STALE_QUOTE_THRESHOLD_MS = 2000
SPREAD_SPIKE_MULTIPLIER = 3.0  # 3x normal spread = anomaly
PRICE_GAP_THRESHOLD_PCT = 0.05  # 0.05% price gap between sources


class SourceStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    STALE = "stale"
    ANOMALY = "anomaly"


# ── Data structures ──────────────────────────────────────────────

@dataclass
class TickQuote:
    """A single tick from one source."""
    source: str
    symbol: str = "XAUUSD"
    bid: float = 0.0
    ask: float = 0.0
    spread: float = 0.0
    ts: float = 0.0  # Unix timestamp
    ts_iso: str = ""
    latency_ms: float = 0.0  # round-trip latency
    reliability: float = 1.0  # 0..1
    status: SourceStatus = SourceStatus.HEALTHY
    raw: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.ts_iso:
            self.ts_iso = datetime.fromtimestamp(self.ts, tz=timezone.utc).isoformat()
        if self.spread == 0.0 and self.ask > self.bid:
            self.spread = round((self.ask - self.bid) * 100, 2)

    @property
    def mid(self) -> float:
        return (self.bid + self.ask) / 2 if self.ask > self.bid else self.bid

    @property
    def is_stale(self) -> bool:
        return (time.time() - self.ts) * 1000 > STALE_QUOTE_THRESHOLD_MS


@dataclass
class CompositeQuote:
    """Cross-validated composite quote."""
    symbol: str = "XAUUSD"
    bid: float = 0.0
    ask: float = 0.0
    spread: float = 0.0
    mid: float = 0.0
    ts: float = 0.0
    ts_iso: str = ""
    sources_used: list[str] = field(default_factory=list)
    primary_source: str = ""
    composite_method: str = "majority_vote"  # or "best_reliability"
    outlier_rejected: list[str] = field(default_factory=list)
    spread_anomaly: bool = False
    price_gap_detected: bool = False
    confidence: float = 1.0  # 0..1 based on cross-validation
    all_quotes: list[TickQuote] = field(default_factory=list)


@dataclass
class SourceHealth:
    """Per-source health tracking."""
    name: str
    reliability: float = 1.0
    last_success: float = 0.0
    last_failure: float = 0.0
    consecutive_failures: int = 0
    total_polls: int = 0
    total_success: int = 0
    avg_latency_ms: float = 0.0
    status: SourceStatus = SourceStatus.HEALTHY
    last_quote: TickQuote | None = None
    error_count_24h: int = 0

    def record_success(self, latency_ms: float) -> None:
        self.last_success = time.time()
        self.consecutive_failures = 0
        self.total_polls += 1
        self.total_success += 1
        self.avg_latency_ms = self.avg_latency_ms * 0.9 + latency_ms * 0.1
        self.reliability = min(RELIABILITY_MAX, self.reliability + RELIABILITY_BOOST)
        self.status = SourceStatus.HEALTHY

    def record_failure(self, error: str = "") -> None:
        self.last_failure = time.time()
        self.consecutive_failures += 1
        self.total_polls += 1
        self.reliability = max(RELIABILITY_MIN, self.reliability * RELIABILITY_DECAY)
        if self.consecutive_failures >= 3:
            self.status = SourceStatus.DEGRADED
        if self.consecutive_failures >= 5:
            self.status = SourceStatus.STALE


# ── MT5 source ───────────────────────────────────────────────────

def _poll_mt5(timeout_ms: int = 200) -> TickQuote | None:
    """Poll MT5 via bridge file protocol (non-blocking)."""
    try:
        import config_manager
        MT5_FILES = config_manager.get_mt5_dir()
        if MT5_FILES is None:
            return None
        BRIDGE_FROM_EA = MT5_FILES / "gold_autotrade/bridge/from_ea"
        QUOTE_FILE = BRIDGE_FROM_EA / "quote_snapshot.json"

        if not QUOTE_FILE.exists():
            return None

        payload = json.loads(QUOTE_FILE.read_text(encoding="utf-8"))
        bid = float(payload.get("bid", 0) or 0)
        ask = float(payload.get("ask", 0) or 0)
        ts = payload.get("ts", time.time())
        if isinstance(ts, str):
            try:
                dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
                ts = dt.timestamp()
            except Exception:
                ts = time.time()

        if bid <= 0 or ask <= 0:
            return None

        spread = round((ask - bid) * 100, 2)
        return TickQuote(
            source="mt5",
            symbol="XAUUSD",
            bid=bid,
            ask=ask,
            spread=spread,
            ts=ts,
            ts_iso=datetime.fromtimestamp(ts, tz=timezone.utc).isoformat(),
        )
    except Exception:
        return None


# ── Swissquote source ───────────────────────────────────────────

def _poll_swissquote(timeout_ms: int = 500) -> TickQuote | None:
    """Poll Swissquote public forex feed."""
    try:
        import urllib.request

        url = "https://forex-data-feed.swissquote.com/public-quotes/bboquotes/instrument/XAU/USD"
        headers = {
            "User-Agent": "Mozilla/5.0",
            "Accept": "application/json",
        }
        req = urllib.request.Request(url, headers=headers)
        start = time.time()
        with urllib.request.urlopen(req, timeout=timeout_ms / 1000) as resp:
            latency_ms = (time.time() - start) * 1000
            data = json.loads(resp.read())

        if not data or not isinstance(data, list):
            return None

        entry = data[0]
        prices = entry.get("spreadProfilePrices", [])
        if not prices:
            return None

        p = prices[0]
        bid = float(p["bid"])
        ask = float(p["ask"])
        ts = time.time()
        spread = round((ask - bid) * 100, 2)

        return TickQuote(
            source="swissquote",
            symbol="XAUUSD",
            bid=bid,
            ask=ask,
            spread=spread,
            ts=ts,
            latency_ms=latency_ms,
        )
    except Exception:
        return None


# ── OANDA source ────────────────────────────────────────────────

def _poll_oanda(timeout_ms: int = 1000) -> TickQuote | None:
    """Poll OANDA API for XAUUSD pricing."""
    try:
        account_id = os.environ.get("OANDA_ACCOUNT_ID", "")
        api_key = os.environ.get("OANDA_API_KEY", "")
        if not account_id or not api_key:
            return None

        import urllib.request

        url = f"https://api-fxpractice.oanda.com/v3/accounts/{account_id}/pricing?instruments=XAU_USD"
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }
        req = urllib.request.Request(url, headers=headers)
        start = time.time()
        with urllib.request.urlopen(req, timeout=timeout_ms / 1000) as resp:
            latency_ms = (time.time() - start) * 1000
            data = json.loads(resp.read())

        prices = data.get("prices", [])
        if not prices:
            return None

        p = prices[0]
        bids = p.get("bids", [])
        asks = p.get("asks", [])
        if not bids or not asks:
            return None

        bid = float(bids[0]["market"])
        ask = float(asks[0]["market"])
        ts_str = p.get("time", "")
        ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")).timestamp() if ts_str else time.time()
        spread = round((ask - bid) * 100, 2)

        return TickQuote(
            source="oanda",
            symbol="XAUUSD",
            bid=bid,
            ask=ask,
            spread=spread,
            ts=ts,
            latency_ms=latency_ms,
        )
    except Exception:
        return None


# ── Cross-validation & composite ────────────────────────────────

def _compute_iqr(values: list[float]) -> tuple[float, float, float]:
    """Compute IQR-based outlier detection. Returns (q1, q3, iqr)."""
    if len(values) < 4:
        return 0, 0, 0
    s = sorted(values)
    q1 = s[len(s) // 4]
    q3 = s[3 * len(s) // 4]
    iqr = q3 - q1
    return q1, q3, iqr


def _is_outlier(value: float, q1: float, q3: float, iqr: float, multiplier: float = 1.5) -> bool:
    if iqr == 0:
        return False
    lower = q1 - multiplier * iqr
    upper = q3 + multiplier * iqr
    return value < lower or value > upper


def _is_spread_anomaly(quotes: list[TickQuote], normal_spread: float = 50.0) -> bool:
    """Detect if any quote has a spread spike."""
    if not quotes:
        return False
    for q in quotes:
        if q.spread > normal_spread * SPREAD_SPIKE_MULTIPLIER:
            return True
    return False


def _is_price_gap(quotes: list[TickQuote]) -> bool:
    """Detect if bid/ask prices diverge significantly across sources."""
    if len(quotes) < 2:
        return False
    mids = [q.mid for q in quotes if q.mid > 0]
    if len(mids) < 2:
        return False
    ref_mid = sum(mids) / len(mids)
    for m in mids:
        gap_pct = abs(m - ref_mid) / ref_mid * 100
        if gap_pct > PRICE_GAP_THRESHOLD_PCT:
            return True
    return False


def compute_composite(quotes: list[TickQuote], method: str = "majority_vote") -> CompositeQuote:
    """Cross-validate multiple quotes and produce a composite."""
    if not quotes:
        return CompositeQuote()

    valid = [q for q in quotes if q.bid > 0 and q.ask > 0 and not q.is_stale]
    if not valid:
        # Fall back to latest available even if stale
        valid = [q for q in quotes if q.bid > 0 and q.ask > 0]

    if not valid:
        return CompositeQuote()

    # Sort by reliability * priority
    def score(q: TickQuote) -> float:
        p = next((SOURCES[s]["priority"] for s in SOURCES if s == q.source), 99)
        return q.reliability / max(1, p)

    valid.sort(key=score, reverse=True)

    # Outlier detection via IQR on mid prices
    mids = [q.mid for q in valid]
    q1, q3, iqr = _compute_iqr(mids)

    filtered: list[TickQuote] = []
    rejected: list[str] = []
    for q in valid:
        if _is_outlier(q.mid, q1, q3, iqr):
            rejected.append(q.source)
        else:
            filtered.append(q)

    if len(filtered) == 0:
        # No consensus — use most reliable source
        filtered = [valid[0]]
        rejected = [q.source for q in valid[1:]]

    # Composite: weighted average by reliability
    total_weight = sum(q.reliability for q in filtered)
    composite_bid = sum(q.bid * q.reliability for q in filtered) / total_weight
    composite_ask = sum(q.ask * q.reliability for q in filtered) / total_weight
    composite_spread = composite_ask - composite_bid

    # Confidence based on agreement level
    mids_filtered = [q.mid for q in filtered]
    mid_std = (sum((m - sum(mids_filtered) / len(mids_filtered)) ** 2 for m in mids_filtered) / len(mids_filtered)) ** 0.5
    mid_ref = sum(mids_filtered) / len(mids_filtered)
    agreement = 1.0 - min(1.0, mid_std / max(1, mid_ref)) if mid_ref else 0.5
    confidence = round(agreement * (len(filtered) / max(1, len(quotes))), 3)

    return CompositeQuote(
        symbol="XAUUSD",
        bid=round(composite_bid, 3),
        ask=round(composite_ask, 3),
        spread=round(composite_spread, 2),
        mid=round((composite_bid + composite_ask) / 2, 3),
        ts=time.time(),
        ts_iso=datetime.now(timezone.utc).isoformat(),
        sources_used=[q.source for q in filtered],
        primary_source=filtered[0].source,
        composite_method=method,
        outlier_rejected=rejected,
        spread_anomaly=_is_spread_anomaly(quotes, normal_spread=50.0),
        price_gap_detected=_is_price_gap(quotes),
        confidence=confidence,
        all_quotes=quotes,
    )


# ── Tick logger ──────────────────────────────────────────────────

TICK_LOG_FILE = TICK_DIR / "tick_stream.ndjson"
TICK_METRICS_FILE = LOGS_DIR / "tick_feed_metrics.ndjson"


def _append_tick(quote: CompositeQuote) -> None:
    """Log composite tick to rotating NDJSON file."""
    try:
        TICK_DIR.mkdir(parents=True, exist_ok=True)
        record = {
            "ts": quote.ts_iso,
            "ts_unix": quote.ts,
            "symbol": quote.symbol,
            "bid": quote.bid,
            "ask": quote.ask,
            "spread": quote.spread,
            "mid": quote.mid,
            "primary_source": quote.primary_source,
            "sources_used": quote.sources_used,
            "confidence": quote.confidence,
            "spread_anomaly": quote.spread_anomaly,
            "price_gap": quote.price_gap_detected,
            "outlier_rejected": quote.outlier_rejected,
        }
        with open(TICK_LOG_FILE, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        pass


def _append_tick_metric(source: str, latency_ms: float, success: bool, status: str) -> None:
    """Log tick fetch metrics."""
    try:
        LOGS_DIR.mkdir(parents=True, exist_ok=True)
        record = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "source": source,
            "latency_ms": round(latency_ms, 2),
            "success": success,
            "status": status,
        }
        with open(TICK_METRICS_FILE, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        pass


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

class AnomalyCircuit:
    """Fast circuit breaker for tick feed anomalies."""

    def __init__(self, trigger_threshold: int = 3, recovery_timeout_sec: int = 30) -> None:
        self._lock = threading.Lock()
        self.trigger_threshold = trigger_threshold
        self.recovery_timeout_sec = recovery_timeout_sec
        self.anomaly_count = 0
        self.last_anomaly_ts = 0.0
        self.is_open = False
        self.suppressed_sources: set[str] = set()

    def record_anomaly(self, source: str) -> None:
        with self._lock:
            self.anomaly_count += 1
            self.last_anomaly_ts = time.time()
            self.suppressed_sources.add(source)
            if self.anomaly_count >= self.trigger_threshold:
                self.is_open = True

    def record_recovery(self, source: str) -> None:
        with self._lock:
            if source in self.suppressed_sources:
                self.suppressed_sources.discard(source)
                self.anomaly_count = max(0, self.anomaly_count - 1)
            if self.is_open and time.time() - self.last_anomaly_ts > self.recovery_timeout_sec:
                if self.anomaly_count < self.trigger_threshold:
                    self.is_open = False

    def is_source_suppressed(self, source: str) -> bool:
        with self._lock:
            return source in self.suppressed_sources

    def check_recovery(self) -> None:
        with self._lock:
            if self.is_open and time.time() - self.last_anomaly_ts > self.recovery_timeout_sec:
                self.is_open = False


# ── Main TickFeed engine ────────────────────────────────────────

class TickFeed:
    """Multi-source tick feed with cross-validation and anomaly detection."""

    def __init__(
        self,
        poll_interval_ms: int = 50,
        max_sources: int = 3,
        log_ticks: bool = True,
    ) -> None:
        self.poll_interval_ms = poll_interval_ms
        self.max_sources = max_sources
        self.log_ticks = log_ticks

        self._health: dict[str, SourceHealth] = {
            name: SourceHealth(name=name) for name in SOURCES
        }
        self._circuit = AnomalyCircuit()
        self._composite = CompositeQuote()
        self._lock = threading.Lock()
        self._running = False
        self._thread: threading.Thread | None = None
        self._last_composite_ts = 0.0

        # Optional ZMQ pub socket for zero-copy inter-thread
        self._zmq_ctx = None
        self._zmq_socket = None
        if os.environ.get("TICK_ZMQ_ENABLED", "false").lower() == "true":
            try:
                import zmq
                self._zmq_ctx = zmq.Context()
                self._zmq_socket = self._zmq_ctx.socket(zmq.PUB)
                addr = os.environ.get("TICK_ZMQ_ADDR", "ipc:///tmp/tick_feed.ipc")
                self._zmq_socket.bind(addr)
            except Exception:
                self._zmq_ctx = None
                self._zmq_socket = None

        # Rolling price history for spread normalization
        self._spread_history = deque(maxlen=100)

    def _poll_all_sources(self) -> list[TickQuote]:
        """Poll all enabled sources in parallel via threads."""
        results: list[TickQuote] = []

        def poll_mt5() -> None:
            if not SOURCES["mt5"]["enabled"] or self._circuit.is_source_suppressed("mt5"):
                return
            start = time.time()
            q = _poll_mt5(timeout_ms=SOURCES["mt5"]["timeout_ms"])
            if q:
                q.latency_ms = (time.time() - start) * 1000
                results.append(q)
                h = self._health["mt5"]
                h.record_success(q.latency_ms)
                h.last_quote = q
                _append_tick_metric("mt5", q.latency_ms, True, "ok")
            else:
                self._health["mt5"].record_failure()
                _append_tick_metric("mt5", 0, False, "timeout")

        def poll_swissquote() -> None:
            if not SOURCES["swissquote"]["enabled"] or self._circuit.is_source_suppressed("swissquote"):
                return
            start = time.time()
            q = _poll_swissquote(timeout_ms=SOURCES["swissquote"]["timeout_ms"])
            if q:
                q.latency_ms = (time.time() - start) * 1000
                results.append(q)
                h = self._health["swissquote"]
                h.record_success(q.latency_ms)
                h.last_quote = q
                _append_tick_metric("swissquote", q.latency_ms, True, "ok")
            else:
                self._health["swissquote"].record_failure()
                _append_tick_metric("swissquote", 0, False, "timeout")

        def poll_oanda() -> None:
            if not SOURCES["oanda"]["enabled"] or self._circuit.is_source_suppressed("oanda"):
                return
            start = time.time()
            q = _poll_oanda(timeout_ms=SOURCES["oanda"]["timeout_ms"])
            if q:
                q.latency_ms = (time.time() - start) * 1000
                results.append(q)
                h = self._health["oanda"]
                h.record_success(q.latency_ms)
                h.last_quote = q
                _append_tick_metric("oanda", q.latency_ms, True, "ok")
            else:
                self._health["oanda"].record_failure()
                _append_tick_metric("oanda", 0, False, "timeout")

        threads = [
            threading.Thread(target=poll_mt5, daemon=True),
            threading.Thread(target=poll_swissquote, daemon=True),
            threading.Thread(target=poll_oanda, daemon=True),
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=1.0)

        return results

    def _process_cycle(self) -> None:
        """One polling cycle."""
        quotes = self._poll_all_sources()

        if not quotes:
            return

        composite = compute_composite(quotes)

        # Update spread history for normalization
        if composite.spread > 0:
            self._spread_history.append(composite.spread)
            avg_spread = sum(self._spread_history) / len(self._spread_history)
            if composite.spread > avg_spread * SPREAD_SPIKE_MULTIPLIER:
                composite.spread_anomaly = True
                self._circuit.record_anomaly(composite.primary_source)

        if composite.price_gap_detected:
            self._circuit.record_anomaly("cross_source_gap")

        # Recovery check
        self._circuit.check_recovery()
        for h in self._health.values():
            if h.consecutive_failures == 0:
                self._circuit.record_recovery(h.name)

        with self._lock:
            self._composite = composite
            self._last_composite_ts = time.time()

        if self.log_ticks:
            _append_tick(composite)

        # ZMQ publish
        if self._zmq_socket:
            try:
                msg = json.dumps({
                    "bid": composite.bid,
                    "ask": composite.ask,
                    "mid": composite.mid,
                    "ts": composite.ts,
                    "source": composite.primary_source,
                    "confidence": composite.confidence,
                }, ensure_ascii=False)
                self._zmq_socket.send_string(msg, flags=zmq.NOBLOCK)
            except Exception:
                pass

    def _run_loop(self) -> None:
        """Background polling loop."""
        while self._running:
            cycle_start = time.time()
            self._process_cycle()
            elapsed_ms = (time.time() - cycle_start) * 1000
            sleep_ms = self.poll_interval_ms - elapsed_ms
            if sleep_ms > 0:
                time.sleep(sleep_ms / 1000)

    def start(self) -> None:
        """Start the background polling thread."""
        if self._running:
            return
        self._running = True
        self._thread = threading.Thread(target=self._run_loop, daemon=True, name="TickFeed")
        self._thread.start()

    def stop(self) -> None:
        """Stop the background polling thread."""
        self._running = False
        if self._thread:
            self._thread.join(timeout=2.0)
            self._thread = None
        if self._zmq_socket:
            try:
                self._zmq_socket.close()
            except Exception:
                pass
        if self._zmq_ctx:
            try:
                self._zmq_ctx.term()
            except Exception:
                pass

    def get_quote(self) -> CompositeQuote:
        """Get latest composite quote (non-blocking read)."""
        with self._lock:
            return self._composite

    def get_feed_status(self) -> dict[str, Any]:
        """Get feed health and status."""
        with self._lock:
            composite = self._composite
        health = {name: {
            "reliability": round(h.reliability, 4),
            "status": h.status.value,
            "avg_latency_ms": round(h.avg_latency_ms, 2),
            "consecutive_failures": h.consecutive_failures,
            "total_polls": h.total_polls,
            "total_success": h.total_success,
            "last_success_ts": h.last_success,
            "is_suppressed": self._circuit.is_source_suppressed(name),
        } for name, h in self._health.items()}

        # Compute normal spread from history
        normal_spread = 50.0
        if self._spread_history:
            normal_spread = sum(self._spread_history) / len(self._spread_history)

        return {
            "ts": datetime.now(timezone.utc).isoformat(),
            "poll_interval_ms": self.poll_interval_ms,
            "is_running": self._running,
            "circuit_open": self._circuit.is_open,
            "suppressed_sources": list(self._circuit.suppressed_sources),
            "normal_spread": round(normal_spread, 2),
            "composite": {
                "bid": composite.bid,
                "ask": composite.ask,
                "mid": composite.mid,
                "spread": composite.spread,
                "spread_anomaly": composite.spread_anomaly,
                "primary_source": composite.primary_source,
                "sources_used": composite.sources_used,
                "confidence": composite.confidence,
                "ts_iso": composite.ts_iso,
            },
            "sources": health,
        }


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

_feed_instance: TickFeed | None = None
_feed_lock = threading.Lock()


def get_tick_feed(poll_interval_ms: int = 50) -> TickFeed:
    """Get or create the global tick feed instance."""
    global _feed_instance
    with _feed_lock:
        if _feed_instance is None:
            _feed_instance = TickFeed(poll_interval_ms=poll_interval_ms)
            _feed_instance.start()
        return _feed_instance


def get_composite_quote() -> CompositeQuote:
    """Get the latest composite quote (reads from singleton)."""
    global _feed_instance
    if _feed_instance is None:
        return CompositeQuote()
    return _feed_instance.get_quote()


def get_feed_status() -> dict[str, Any]:
    """Get tick feed status."""
    global _feed_instance
    if _feed_instance is None:
        return {"is_running": False}
    return _feed_instance.get_feed_status()


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

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Multi-source tick feed")
    parser.add_argument("--poll-ms", type=int, default=50, help="Poll interval in ms")
    parser.add_argument("--status", action="store_true", help="Show feed status")
    parser.add_argument("--test-sources", action="store_true", help="Test each source")
    args = parser.parse_args()

    if args.status:
        status = get_feed_status()
        print(json.dumps(status, indent=2, ensure_ascii=False))
    elif args.test_sources:
        print("Testing MT5...")
        q = _poll_mt5()
        print(f"  MT5: {q}")
        print("Testing Swissquote...")
        q = _poll_swissquote()
        print(f"  Swissquote: {q}")
        print("Testing OANDA...")
        q = _poll_oanda()
        print(f"  OANDA: {q}")
    else:
        feed = TickFeed(poll_interval_ms=args.poll_ms)
        feed.start()
        print(f"Tick feed started at {args.poll_ms}ms interval. Press Ctrl+C to stop.")
        try:
            while True:
                time.sleep(1)
                q = feed.get_quote()
                if q.bid > 0:
                    print(f"[{q.ts_iso[11:19]}] {q.symbol} bid={q.bid} ask={q.ask} "
                          f"spread={q.spread} primary={q.primary_source} conf={q.confidence}")
        except KeyboardInterrupt:
            feed.stop()
            print("\nStopped.")