Centaur_Quant_Architecture/Python/models/analyzer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
models/analyzer.py — MarketAnalyzer | LLM AI Evaluation Engine (Track A)

Converts a raw SDP Setup_Detected envelope (symbol, setup_type, prices,
algorithmic confidence, historical_context swings) into a strict quant
prompt, calls an external LLM through a modular backend (OpenAI or
Google GenAI SDK), and enforces the structured JSON contract:

    {"score": <float 0-100>, "reason": "<string, max 10 words>"}

Anti-Veto safety (charter):
    Any timeout (>1.5 s hard budget), API error, missing SDK, or malformed
    reply returns the neutral FALLBACK_SCORE = 40.0. Because 40.0 < 50,
    the MT5 executor drops the trade to quarter-risk and still executes —
    the AI advises, it never vetoes.

Provider selection (environment driven):
    PROVIDER=auto|openai|google   (default auto; falls back across providers)
    MODEL | OPENAI_MODEL | GEMINI_MODEL   (model overrides)
    OPENAI_API_KEY | GEMINI_API_KEY       (credentials)
    OPENAI_BASE_URL                  (optional custom OpenAI-compatible gateway,
                                      e.g. https://api4.inferdeck.net/v1 + MODEL="MQL5 Lite")
"""

import json
import logging
import os
import queue
import re
import threading
import time
from typing import Any, Dict, Optional, Tuple

logger = logging.getLogger(__name__)

# --------------------------------------------------------------------------
# Constants
# --------------------------------------------------------------------------
DEFAULT_TIMEOUT_S = 1.5      # strict wall-clock budget for the LLM call
FALLBACK_SCORE = 40.0        # neutral score -> triggers MT5 quarter-risk path
MAX_REASON_WORDS = 10        # the "reason" contract: max 10 words

# --------------------------------------------------------------------------
# Quant system prompt (single source of truth for LLM behaviour)
# --------------------------------------------------------------------------
SYSTEM_PROMPT = (
    "You are a senior Smart Money Concepts (SMC) quant analyst evaluating a single "
    "Order Block + Fair Value Gap (OB/FVG) trade setup.\n\n"
    "Evaluate BOTH dimensions:\n"
    "1. MOMENTUM — is the displacement candle that created the FVG strong and "
    "recent? A larger imbalance relative to context means stronger momentum.\n"
    "2. TREND ALIGNMENT — is the setup in the direction of the prevailing swing "
    "structure (higher highs / higher lows for buys, the reverse for sells)?\n\n"
    "Reply with EXACTLY ONE JSON object and nothing else. No markdown, no prose, "
    "no code fences:\n"
    '{"score": <float 0-100>, "reason": "<string, max 10 words>"}\n\n'
    "Rules:\n"
    "- score = probability (0-100) the setup resolves in its intended direction "
    "after a retracement into the unmitigated zone.\n"
    "- reason must be 10 words or fewer.\n"
    "- If evidence is ambiguous, score toward the middle (40-60)."
)


# --------------------------------------------------------------------------
# Modular SDK backends (lazy imports keep this module importable everywhere)
# --------------------------------------------------------------------------
class _BaseBackend:
    """Common interface every SDK adapter implements."""

    name = "base"

    def __init__(self, model: str, timeout: float) -> None:
        self._model = model
        self._timeout = timeout

    def generate(self, system_prompt: str, user_prompt: str) -> str:
        raise NotImplementedError


class _OpenAIBackend(_BaseBackend):
    """OpenAI chat.completions adapter (openai SDK)."""

    name = "openai"

    def __init__(self, model: Optional[str], timeout: float) -> None:
        super().__init__(model or os.environ.get("OPENAI_MODEL", "gpt-4o-mini"), timeout)
        from openai import OpenAI  # lazy: module stays importable without the SDK
        self._client = OpenAI(
            api_key=os.environ.get("OPENAI_API_KEY"),
            base_url=os.environ.get("OPENAI_BASE_URL") or None,
            timeout=self._timeout,
        )

    def generate(self, system_prompt: str, user_prompt: str) -> str:
        resp = self._client.chat.completions.create(
            model=self._model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0.0,
            max_tokens=120,
        )
        return resp.choices[0].message.content or ""


class _GoogleGenAIBackend(_BaseBackend):
    """Google GenAI adapter (google-genai SDK)."""

    name = "google"

    def __init__(self, model: Optional[str], timeout: float) -> None:
        super().__init__(model or os.environ.get("GEMINI_MODEL", "gemini-3.1-flash-lite"), timeout)
        from google import genai  # lazy import
        from google.genai import types
        self._types = types
        self._client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))

    def generate(self, system_prompt: str, user_prompt: str) -> str:
        config = self._types.GenerateContentConfig(
            system_instruction=system_prompt,
            temperature=0.0,
            max_output_tokens=120,
        )
        try:
            resp = self._client.models.generate_content(
                model=self._model, contents=user_prompt, config=config,
                timeout=self._timeout,
            )
        except TypeError:
            # older SDK versions do not accept the timeout kwarg
            resp = self._client.models.generate_content(
                model=self._model, contents=user_prompt, config=config,
            )
        return getattr(resp, "text", "") or ""


_PROVIDERS: Dict[str, Any] = {
    "openai": _OpenAIBackend,
    "google": _GoogleGenAIBackend,
}


def _build_backend(provider: str, model: Optional[str],
                   timeout: float) -> Optional[_BaseBackend]:
    """Instantiate the requested provider; gracefully try the other on failure."""
    if provider in _PROVIDERS:
        candidates = [provider] + [p for p in _PROVIDERS if p != provider]
    else:
        candidates = list(_PROVIDERS)   # provider="auto" or unknown -> try all

    for name in candidates:
        try:
            backend = _PROVIDERS[name](model, timeout)
            logger.info("MarketAnalyzer: provider '%s' ready (model=%s).",
                        name, backend._model)
            return backend
        except ImportError as exc:
            logger.warning("MarketAnalyzer: provider '%s' unavailable (%s); trying next.",
                           name, exc)
        except Exception as exc:  # noqa: BLE001 - bad key/init must not crash the router
            logger.warning("MarketAnalyzer: provider '%s' init failed: %s", name, exc)

    logger.warning("MarketAnalyzer: no LLM backend available — fallback-only mode (score %.1f).",
                   FALLBACK_SCORE)
    return None


def _run_with_deadline(fn, timeout: float, *args, **kwargs) -> Tuple[str, Any]:
    """Run fn in a daemon thread under a HARD wall-clock budget.

    Returns ("ok", result) | ("error", exception) | ("timeout", None).
    The worker thread is daemonized, so a hung SDK call can never block
    process shutdown.
    """
    result_q: "queue.Queue[Tuple[str, Any]]" = queue.Queue(maxsize=1)

    def _worker() -> None:
        try:
            result_q.put(("ok", fn(*args, **kwargs)))
        except Exception as exc:  # noqa: BLE001 - must never leak from the worker
            result_q.put(("error", exc))

    threading.Thread(target=_worker, daemon=True).start()

    try:
        return result_q.get(timeout=timeout)
    except queue.Empty:
        return ("timeout", None)


# --------------------------------------------------------------------------
# MarketAnalyzer — the AI evaluation engine consumed by the router
# --------------------------------------------------------------------------
class MarketAnalyzer:
    """
    Core AI evaluation engine for Setup_Detected envelopes.

    Usage:
        analyzer = MarketAnalyzer()               # provider/model from env
        result   = analyzer.analyze(setup_payload)  # SDP Setup_Detected dict
        score    = result["score"]                # 0-100, or 40.0 fallback
    """

    def __init__(self,
                 provider: str = "auto",
                 model: Optional[str] = None,
                 timeout: float = DEFAULT_TIMEOUT_S,
                 retries: int = 1,
                 fallback_score: float = FALLBACK_SCORE) -> None:
        if provider in ("", "auto"):
            provider = os.environ.get("PROVIDER", "auto")
        self._timeout = max(0.1, float(timeout))
        self._fallback_score = max(0.0, min(100.0, float(fallback_score)))
        self._retries = max(0, int(retries))
        self._backend = _build_backend(provider, model, self._timeout)

    # ------------------------------------------------------------------
    # Prompt engineering
    # ------------------------------------------------------------------
    def build_prompt(self, payload: Dict[str, Any]) -> Tuple[str, str]:
        """
        Translate the raw Setup_Detected envelope into (system, user) prompts.

        The user prompt embeds every quant fact the LLM needs: symbol,
        timeframe, setup type, prices, structural confidence, and the
        historical_context swing array (compact, oldest -> newest).
        """
        p = payload.get("payload") or {}
        hist = payload.get("historical_context") or []

        swings = []
        for h in hist[-12:]:   # cap context size; newest swings matter most
            if isinstance(h, dict):
                swings.append("H=%s L=%s @ %s" % (
                    h.get("swing_high", "?"),
                    h.get("swing_low", "?"),
                    h.get("time", "?"),
                ))
        ctx = "; ".join(swings) if swings else "none"

        user_prompt = (
            "SDP Setup_Detected envelope:\n"
            "Symbol: %s | Timeframe: %s | Setup: %s\n"
            "Entry: %s | SL: %s | TP: %s | R:R: %s\n"
            "Algorithmic confidence: %s\n"
            "Historical context (swing highs/lows, oldest -> newest):\n%s\n\n"
            "Evaluate OB/FVG momentum and trend alignment. "
            "Reply with the strict JSON object only."
        ) % (
            payload.get("symbol", "?"),
            payload.get("timeframe", "?"),
            p.get("setup_type", "?"),
            p.get("entry", "?"),
            p.get("sl", "?"),
            p.get("tp", "?"),
            p.get("risk_reward", "?"),
            payload.get("algorithmic_confidence_score", "?"),
            ctx,
        )
        return SYSTEM_PROMPT, user_prompt

    # ------------------------------------------------------------------
    # Public evaluation entry point
    # ------------------------------------------------------------------
    @staticmethod
    def _is_transient(exc: Exception) -> bool:
        """True bila error transien (rate limit 429 / 5xx / koneksi) — layak retry."""
        code = getattr(exc, "status_code", None) or getattr(exc, "code", None)
        if code in (429, 500, 502, 503, 504):
            return True
        msg = str(exc).lower()
        if any(k in msg for k in ("rate limit", "429", "too many requests",
                                  "connection", "timeout", "temporarily unavailable")):
            return True
        return isinstance(exc, (ConnectionError, TimeoutError))

    def _retry_or_fallback(self, system_prompt: str, user_prompt: str,
                           last_error: Exception, started: float) -> Dict[str, Any]:
        """Retry transien dengan backoff eksponensial; habis -> fallback 40.0 (Anti-Veto)."""
        for attempt in range(1, self._retries + 1):
            if not self._is_transient(last_error):
                break
            backoff = min(2.0 * attempt, 8.0)
            logger.warning("MarketAnalyzer: error transien %r — retry %d/%d dalam %.1fs",
                           last_error, attempt, self._retries, backoff)
            time.sleep(backoff)
            status, value = _run_with_deadline(
                self._backend.generate, self._timeout, system_prompt, user_prompt
            )
            if status == "ok":
                parsed = self._parse_json_reply(str(value))
                if parsed is not None:
                    score, reason = parsed
                    return {
                        "score": score, "reason": reason,
                        "used_fallback": False,
                        "provider": self._backend.name,
                        "latency_ms": int((time.perf_counter() - started) * 1000),
                    }
                return self._fallback("malformed/unparseable LLM reply", started)
            if status == "timeout":
                return self._fallback("LLM timeout > %.1fs" % self._timeout, started)
            last_error = value
        return self._fallback("LLM error: %s" % last_error, started)

    def analyze(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """
        Evaluate one setup; always returns a dict with 'score' (0-100).

        Never raises: every failure path collapses to the neutral fallback
        score so the MT5 Anti-Veto pipeline keeps running.
        """
        started = time.perf_counter()

        if self._backend is None:
            return self._fallback("no LLM backend available", started)

        system_prompt, user_prompt = self.build_prompt(payload)

        status, value = _run_with_deadline(
            self._backend.generate, self._timeout, system_prompt, user_prompt
        )

        if status == "timeout":
            return self._fallback("LLM timeout > %.1fs" % self._timeout, started)
        if status == "error":
            return self._retry_or_fallback(system_prompt, user_prompt, value, started)

        parsed = self._parse_json_reply(str(value))
        if parsed is None:
            return self._fallback("malformed/unparseable LLM reply", started)

        score, reason = parsed
        return {
            "score": score,
            "reason": reason,
            "used_fallback": False,
            "provider": self._backend.name,
            "latency_ms": int((time.perf_counter() - started) * 1000),
        }

    def is_ready(self) -> bool:
        """True when a real LLM backend is configured and importable."""
        return self._backend is not None

    # ------------------------------------------------------------------
    # Internals
    # ------------------------------------------------------------------
    @staticmethod
    def _parse_json_reply(raw: str) -> Optional[Tuple[float, str]]:
        """
        Extract and validate the structured JSON reply.

        Tolerates markdown fences and surrounding prose; rejects replies
        that violate the schema (missing/invalid score, score out of range).
        """
        cleaned = raw.strip()
        if cleaned.startswith("```"):
            cleaned = re.sub(r"^```[a-zA-Z0-9_-]*\s*", "", cleaned)
            cleaned = re.sub(r"\s*```$", "", cleaned)

        start = cleaned.find("{")
        end = cleaned.rfind("}")
        if start == -1 or end <= start:
            return None

        try:
            obj = json.loads(cleaned[start:end + 1])
        except json.JSONDecodeError:
            return None

        if not isinstance(obj, dict):
            return None

        try:
            score = float(obj.get("score"))
        except (TypeError, ValueError):
            return None
        if not (0.0 <= score <= 100.0):
            return None

        reason = str(obj.get("reason", "")).strip()
        words = reason.split()
        if len(words) > MAX_REASON_WORDS:
            reason = " ".join(words[:MAX_REASON_WORDS])   # enforce the 10-word cap

        return round(score, 2), reason

    def _fallback(self, why: str, started: float) -> Dict[str, Any]:
        """Neutral fallback result — triggers MT5 quarter-risk execution."""
        logger.warning("MarketAnalyzer: fallback score %.1f used — %s",
                       self._fallback_score, why)
        return {
            "score": self._fallback_score,
            "reason": "neutral fallback",
            "used_fallback": True,
            "fallback_reason": why,
            "provider": self._backend.name if self._backend is not None else "none",
            "latency_ms": int((time.perf_counter() - started) * 1000),
        }