MidasBot/ai_analyzer/claude_client.py

"""Client Claude headless via le CLI `claude` (authentifié par l'abonnement).

On n'utilise PAS le SDK `anthropic` ni de clé API. On invoque `claude -p` en
sous-processus avec sortie structurée (`--json-schema`), ce qui consomme l'abonnement
Claude Code de l'utilisateur (pas de facturation au token).
"""
from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path
from typing import Optional

from market_data import PairSnapshot
from models import MarketBias, MarketBiasBatch

_HERE = Path(__file__).resolve().parent
_SYSTEM_PROMPT_FILE = _HERE / "system_prompt.md"


class ClaudeError(RuntimeError):
    pass


class ClaudeClient:
    def __init__(
        self,
        model: str = "claude-sonnet-4-6",
        max_turns: int = 6,
        mcp_config: Optional[str] = None,
        allowed_tools: Optional[str] = None,
        timeout_s: int = 180,
    ) -> None:
        self.model = model
        self.max_turns = max_turns
        # .mcp.json à la racine projet, chargé seulement si présent.
        self.mcp_config = mcp_config
        # Ex. "mcp__tradingview" pour autoriser uniquement les outils du serveur MCP.
        self.allowed_tools = allowed_tools
        self.timeout_s = timeout_s
        self._guard_no_api_key()

    @staticmethod
    def _guard_no_api_key() -> None:
        # ANTHROPIC_API_KEY ferait basculer Claude sur la facturation au token.
        if os.environ.get("ANTHROPIC_API_KEY"):
            raise ClaudeError(
                "ANTHROPIC_API_KEY est défini : on veut l'abonnement, pas l'API. "
                "Retire cette variable (utilise CLAUDE_CODE_OAUTH_TOKEN)."
            )

    def _build_prompt(self, snapshots: list[PairSnapshot]) -> str:
        payload = {
            "instruction": "Produis un biais directionnel pour chaque paire ci-dessous.",
            "market_snapshot": [s.to_dict() for s in snapshots],
            "pairs": [s.pair for s in snapshots],
        }
        return (
            "Voici l'instantané de marché (JSON). Si le MCP TradingView est disponible, "
            "tu peux l'utiliser pour confirmer (RSI/MACD/Bollinger).\n\n"
            + json.dumps(payload, ensure_ascii=False)
        )

    def get_biases(self, snapshots: list[PairSnapshot]) -> MarketBiasBatch:
        if not snapshots:
            return MarketBiasBatch(biases=[])

        system_prompt = (
            _SYSTEM_PROMPT_FILE.read_text(encoding="utf-8")
            if _SYSTEM_PROMPT_FILE.exists()
            else ""
        )

        cmd = [
            "claude",
            "-p",
            self._build_prompt(snapshots),
            "--model",
            self.model,
            "--output-format",
            "json",
            "--json-schema",
            json.dumps(MarketBiasBatch.json_schema()),
            "--max-turns",
            str(self.max_turns),
        ]
        if system_prompt:
            cmd += ["--append-system-prompt", system_prompt]
        if self.mcp_config:
            cmd += ["--mcp-config", self.mcp_config, "--strict-mcp-config"]
        if self.allowed_tools:
            cmd += ["--allowedTools", self.allowed_tools]
        else:
            # Pas de MCP : on coupe les outils intégrés (analyse pure sur l'instantané).
            cmd += [
                "--disallowedTools",
                "Bash,Read,Write,Edit,WebSearch,WebFetch,Glob,Grep,Task",
            ]

        try:
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.timeout_s,
                env=os.environ.copy(),
            )
        except subprocess.TimeoutExpired as exc:
            raise ClaudeError(f"claude -p a dépassé {self.timeout_s}s") from exc

        if proc.returncode != 0:
            raise ClaudeError(
                f"claude -p code {proc.returncode}: {proc.stderr.strip()[:500]}"
            )

        return self._parse(proc.stdout)

    @staticmethod
    def _parse(stdout: str) -> MarketBiasBatch:
        try:
            envelope = json.loads(stdout)
        except json.JSONDecodeError as exc:
            raise ClaudeError(f"Sortie claude non-JSON: {stdout[:300]}") from exc

        if envelope.get("is_error"):
            raise ClaudeError(
                f"claude erreur ({envelope.get('subtype')}): "
                f"{envelope.get('errors') or envelope.get('result')}"
            )

        # Préférence : sortie structurée validée par le schéma.
        structured = envelope.get("structured_output")
        if isinstance(structured, dict):
            return MarketBiasBatch.model_validate(structured)

        # Repli : extraire un objet JSON du champ `result`.
        result = envelope.get("result", "")
        obj = _extract_json_object(result)
        if obj is not None:
            return MarketBiasBatch.model_validate(obj)

        raise ClaudeError(f"Pas de sortie structurée exploitable: {result[:300]}")


def _extract_json_object(text: str) -> Optional[dict]:
    start = text.find("{")
    end = text.rfind("}")
    if start == -1 or end == -1 or end <= start:
        return None
    try:
        return json.loads(text[start : end + 1])
    except json.JSONDecodeError:
        return None