"""Intraday watcher (Milestone 4).

Pulls candidates from a watchlist provider, evaluates each with the pure
strategy engine, drives a per-symbol state machine, and emits canonical
events into the event store:

    discovered -> watching -> ready | blocked   (re-evaluated each tick)

Signals are debounced: a symbol emits signal_ready at most once per session
unless it falls back to blocked first.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime, date
from typing import Callable, Protocol

import pandas as pd

from storage.event_schema import (
    CriteriaEvaluatedEvent,
    EventMode,
    SignalBlockedEvent,
    SignalReadyEvent,
    SymbolDiscoveredEvent,
    SymbolStateChangedEvent,
)
from storage.event_store import EventStore
from strategy.evaluation.setup_evaluator import evaluate_setup
from strategy.population import PopulationConfig, population_reject

logger = logging.getLogger(__name__)


@dataclass
class WatchCandidate:
    """One symbol the watcher should evaluate."""

    symbol: str
    last_price: float | None = None
    previous_close: float | None = None
    avg_daily_volume: float | None = None
    source: str = "research"
    # spike-robust baseline for squeeze verification (see _population_reject)
    median_daily_volume: float | None = None


class WatchlistProvider(Protocol):
    """Source of candidates and their session bars."""

    def get_candidates(self, session_date: date) -> list[WatchCandidate]: ...

    def get_bars(self, symbol: str, session_date: date) -> pd.DataFrame: ...


@dataclass
class WatcherConfig:
    session_id: str = ""
    mode: EventMode = EventMode.PAPER
    max_symbols: int = 25
    min_bars: int = 10
    ready_score_pct: float = 60.0
    min_quality: float = 0.30
    # Shared tradeable-population gate, applied BEFORE the per-tick evaluation.
    # None = off (the historical behaviour every existing caller and test relies
    # on). The live loop passes PopulationConfig.from_env() so the watcher stops
    # scoring names the entry path can never accept: on 2026-09-10 the watcher
    # wrote 191,458 criteria receipts across 57 symbols and 154,260 of them
    # (80.6%) were for gaps outside the tradeable band — large caps like NOK, F
    # and SOFI that sit inside the $2-20 price filter but can never pass the gap
    # floor. Those two event types were 3.1 GB of a 7.6 GB events table.
    population: "PopulationConfig | None" = None


@dataclass
class WatcherTickResult:
    session_date: str
    evaluated: int = 0
    ready: list[str] = field(default_factory=list)
    blocked: list[str] = field(default_factory=list)
    discovered: list[str] = field(default_factory=list)
    skipped: int = 0          # rejected by the population gate before evaluation
    errors: list[str] = field(default_factory=list)


class Watcher:
    """Event-emitting intraday watcher."""

    def __init__(
        self,
        store: EventStore,
        provider: WatchlistProvider,
        config: WatcherConfig | None = None,
        catalyst_score_provider: Callable[[str], float | None] | None = None,
        spread_provider: Callable[[str], float | None] | None = None,
    ):
        self.store = store
        self.provider = provider
        self.config = config or WatcherConfig()
        # optional symbol -> 0..1 catalyst score (or None). Injected so the pure
        # evaluator stays DB-free; None provider == today's behavior exactly.
        self.catalyst_score_provider = catalyst_score_provider
        # optional symbol -> decision-time bid/ask spread fraction (or None).
        # Observe-only: it is logged onto the evaluation/ready events but does NOT
        # gate. None provider == today's behavior exactly (spread_pct logs as None,
        # matching the always-null minute_bars.spread_pct column).
        self.spread_provider = spread_provider
        if not self.config.session_id:
            self.config.session_id = f"session-{date.today().isoformat()}"
        # symbol -> current state for this session
        self._states: dict[str, str] = {}

    # -- state helpers -------------------------------------------------

    def _transition(self, symbol: str, new_state: str, reason: str | None) -> None:
        previous = self._states.get(symbol)
        if previous == new_state:
            return
        self._states[symbol] = new_state
        self.store.emit(
            SymbolStateChangedEvent(
                timestamp=datetime.now(),
                mode=self.config.mode,
                correlation_id=self.config.session_id,
                message=f"{symbol}: {previous or 'new'} -> {new_state}"
                + (f" ({reason})" if reason else ""),
                symbol=symbol,
                previous_state=previous,
                new_state=new_state,
                state_reason=reason,
                payload={"symbol": symbol, "previous_state": previous,
                         "new_state": new_state, "state_reason": reason},
            )
        )

    def _population_reject(self, candidate: WatchCandidate, bars) -> str | None:
        """Reject reason if this name is outside the tradeable population, else None.

        Uses the STATIC OPEN gap (session open vs previous close) — the same
        open-knowable quantity the live entry gate uses, and fixed for the day —
        never the live price, which would make the verdict wander intraday.
        Fail-open throughout: anything unreadable returns None so a data gap can
        never silently bench a symbol.
        """
        try:
            # The SESSION open is the first REGULAR-HOURS bar's open. bars[0] is a
            # PREMARKET bar before 09:30, which produces a fabricated gap: FTFT was
            # benched 2026-09-14 as "gap -12.5%" when its true static open gap was
            # +14.6%, and it never recovered because the frame keeps its premarket
            # bars all session and _transition only fires on change.
            if "is_regular_hours" in bars:
                rth = bars[bars["is_regular_hours"] == True]  # noqa: E712 — pandas mask
                if len(rth) == 0:
                    return None      # session has not opened — gap unknowable, fail open
                open_px = float(rth["open"].iloc[0])
            else:
                # no session flag: the frame is already regular-hours only (tests,
                # replay). Using bars[0] is correct there.
                open_px = float(bars["open"].iloc[0])
        except Exception:  # noqa: BLE001
            return None
        prev_close = candidate.previous_close
        gap = None
        if prev_close:
            try:
                gap = (open_px - float(prev_close)) / float(prev_close)
            except (TypeError, ValueError, ZeroDivisionError):
                gap = None
        volume = rvol = None
        try:
            volume = float(bars["volume"].sum())
            # squeeze verification takes the stronger of the mean- and
            # median-baseline readings (same units): a prior spike day in the
            # 20d window must not bench a real squeeze (KITT 2026-09-25).
            readings = [volume / float(b) for b in (candidate.avg_daily_volume,
                                                    candidate.median_daily_volume)
                        if b and float(b) > 0]
            rvol = max(readings) if readings else None
        except Exception:  # noqa: BLE001
            volume = rvol = None
        return population_reject(
            self.config.population,
            gap_pct=gap, price=open_px, volume=volume, rvol=rvol, as_of="open",
        )

    def _discover(self, candidate: WatchCandidate) -> None:
        if candidate.symbol in self._states:
            return
        self.store.emit(
            SymbolDiscoveredEvent(
                timestamp=datetime.now(),
                mode=self.config.mode,
                correlation_id=self.config.session_id,
                message=f"Discovered {candidate.symbol} via {candidate.source}",
                symbol=candidate.symbol,
                symbol_data={
                    "last_price": candidate.last_price,
                    "previous_close": candidate.previous_close,
                    "avg_daily_volume": candidate.avg_daily_volume,
                    "source": candidate.source,
                },
                payload={"symbol": candidate.symbol, "source": candidate.source},
            )
        )
        self._transition(candidate.symbol, "watching", "added to watchlist")

    # -- main loop body ------------------------------------------------

    def tick(self, session_date: date | None = None) -> WatcherTickResult:
        """One evaluation pass over all candidates."""
        session_date = session_date or date.today()
        result = WatcherTickResult(session_date=session_date.isoformat())

        try:
            candidates = self.provider.get_candidates(session_date)
        except Exception as exc:  # provider failure should not kill the loop
            logger.exception("watchlist provider failed")
            result.errors.append(f"provider: {exc}")
            return result

        for candidate in candidates[: self.config.max_symbols]:
            symbol = candidate.symbol
            try:
                if symbol not in self._states:
                    self._discover(candidate)
                    result.discovered.append(symbol)

                bars = self.provider.get_bars(symbol, session_date)
                if bars is None or len(bars) < self.config.min_bars:
                    self._transition(
                        symbol, "watching",
                        f"waiting for data ({0 if bars is None else len(bars)} bars)",
                    )
                    continue

                # POPULATION PRE-FILTER — skip the expensive evaluation (and its
                # two fat receipts) for a name the entry path could never accept.
                # Recomputed every tick rather than cached: a >150% gapper starts
                # "squeeze unverified" on thin volume and is admitted later once
                # rvol confirms, so a cached rejection would bench it for the
                # session. _transition only emits on CHANGE, so a rejected symbol
                # costs exactly one event and then goes silent.
                if self.config.population is not None:
                    reason = self._population_reject(candidate, bars)
                    if reason is not None:
                        self._transition(symbol, "out_of_population", reason)
                        result.skipped += 1
                        continue

                # Evaluate the setup as of the latest bar's timestamp, not the
                # wall clock. Minute bars are stored UTC-naive, but the entry
                # cutoff reasons in US/Eastern wall-clock, so convert. This
                # keeps replay and backfilled data honest (a 09:40 ET bar is
                # judged against the 09:40 cutoff regardless of run time).
                eval_time = None
                try:
                    from datetime import timezone as _tz
                    from zoneinfo import ZoneInfo as _ZI

                    last_ts = pd.Timestamp(bars["timestamp"].iloc[-1]).to_pydatetime()
                    if last_ts.tzinfo is None:
                        last_ts = last_ts.replace(tzinfo=_tz.utc)
                    eval_time = last_ts.astimezone(_ZI("America/New_York")).replace(
                        tzinfo=None
                    )
                except Exception:  # noqa: BLE001 - fall back to now()
                    eval_time = None

                # optional LLM catalyst score (None unless the blend is enabled
                # AND a provider is injected) — never raises into the loop.
                cat_score = None
                if self.catalyst_score_provider is not None:
                    try:
                        cat_score = self.catalyst_score_provider(symbol)
                    except Exception:  # noqa: BLE001
                        cat_score = None

                evaluation = evaluate_setup(
                    bars,
                    previous_close=candidate.previous_close,
                    avg_daily_volume=candidate.avg_daily_volume,
                    ready_score_pct=self.config.ready_score_pct,
                    evaluation_time=eval_time,
                    min_bars=self.config.min_bars,
                    catalyst_score=cat_score,
                )
                result.evaluated += 1

                # Decision-time bid/ask spread (observe-only). Guarded like the
                # catalyst provider so a quote blip never kills the loop; None
                # when no provider is injected or the quote is missing/bad.
                spread_pct = None
                if self.spread_provider is not None:
                    try:
                        spread_pct = self.spread_provider(symbol)
                    except Exception:  # noqa: BLE001
                        spread_pct = None

                self.store.emit(
                    CriteriaEvaluatedEvent(
                        timestamp=datetime.now(),
                        mode=self.config.mode,
                        correlation_id=self.config.session_id,
                        message=(
                            f"{symbol} criteria {evaluation.criteria_passed}/"
                            f"{evaluation.criteria_total} "
                            f"({evaluation.success_score_pct:.0f}%)"
                        ),
                        symbol=symbol,
                        criteria_results={
                            "passed": evaluation.criteria_names_passed,
                            "failed": evaluation.criteria_names_failed,
                            "detail": evaluation.criteria_detail,
                            "gap_pct": evaluation.gap_pct,
                            "relative_volume": evaluation.relative_volume,
                            "spread_pct": spread_pct,
                            "status": evaluation.status,
                            "reason": evaluation.reason,
                        },
                        total_criteria=evaluation.criteria_total,
                        passed_criteria=evaluation.criteria_passed,
                        success_score_pct=evaluation.success_score_pct,
                        payload={
                            "symbol": symbol,
                            "success_score_pct": evaluation.success_score_pct,
                            "status": evaluation.status,
                            "reason": evaluation.reason,
                            "gap_pct": evaluation.gap_pct,
                            "relative_volume": evaluation.relative_volume,
                            "spread_pct": spread_pct,
                        },
                    )
                )

                if evaluation.status == "ready" and evaluation.setups:
                    result.ready.append(symbol)
                    if self._states.get(symbol) != "ready":
                        setup = evaluation.setups[0]
                        self._transition(symbol, "ready", setup["setup_type"])
                        signal_data = {**setup, "spread_pct": spread_pct}
                        self.store.emit(
                            SignalReadyEvent(
                                timestamp=datetime.now(),
                                mode=self.config.mode,
                                correlation_id=self.config.session_id,
                                message=(
                                    f"{symbol} READY {setup['setup_type']} "
                                    f"entry {setup['entry_price']} stop "
                                    f"{setup['stop_loss_price']}"
                                ),
                                symbol=symbol,
                                signal_type=setup["setup_type"],
                                confidence=setup["confidence"],
                                signal_data=signal_data,
                                payload={
                                    "symbol": symbol,
                                    "signal_type": setup["setup_type"],
                                    "confidence": setup["confidence"],
                                    "signal_data": signal_data,
                                },
                            )
                        )
                else:
                    status = "blocked" if evaluation.status != "late" else "late"
                    result.blocked.append(symbol)
                    if self._states.get(symbol) == "ready":
                        # fell out of ready — allow re-signal later
                        self._transition(symbol, status, evaluation.reason)
                    elif self._states.get(symbol) != status:
                        self._transition(symbol, status, evaluation.reason)
                    self.store.emit(
                        SignalBlockedEvent(
                            timestamp=datetime.now(),
                            mode=self.config.mode,
                            correlation_id=self.config.session_id,
                            message=f"{symbol} blocked: {evaluation.reason}",
                            symbol=symbol,
                            blocking_reason=evaluation.reason or "unknown",
                            unmet_criteria=evaluation.criteria_names_failed,
                            payload={
                                "symbol": symbol,
                                "blocking_reason": evaluation.reason,
                                "unmet_criteria": evaluation.criteria_names_failed,
                            },
                        )
                    )
            except Exception as exc:
                logger.exception("watcher tick failed for %s", symbol)
                result.errors.append(f"{symbol}: {exc}")

        return result