from __future__ import annotations import json from dataclasses import asdict, dataclass, field from typing import Any import requests from crypto_spot_bot.config import Settings from crypto_spot_bot.models import Candle, Ticker, utc_now from crypto_spot_bot.storage import Storage REGIMES = {"uptrend", "downtrend", "range", "breakout", "breakdown", "panic", "unknown"} RISK_LEVELS = {"low", "medium", "high"} @dataclass(slots=True) class LlmAdvice: symbol: str enabled: bool model: str market_regime: str = "unknown" risk_level: str = "medium" confidence_adjustment: float = 0.0 block_entry: bool = False grid_suitable: bool = False reason_ru: str = "LLM Advisor не дал активной поправки." error: str = "" created_at: str = field(default_factory=lambda: utc_now().isoformat()) def as_dict(self) -> dict[str, Any]: return asdict(self) class LlmAdvisor: def __init__(self, settings: Settings, storage: Storage): self.settings = settings self.storage = storage self._cache: dict[str, LlmAdvice] = {} def advice_for( self, *, symbol: str, candles: list[Candle], ticker: Ticker | None, pattern: dict[str, Any], learning: dict[str, Any], open_positions_for_symbol: int, account: dict[str, float], ) -> LlmAdvice: if not self.settings.llm_advisor_enabled: return LlmAdvice( symbol=symbol, enabled=False, model=self.settings.ollama_model, reason_ru="LLM Advisor выключен.", ) cached = self._cache.get(symbol) if cached and _age_seconds(cached.created_at) < self.settings.llm_advisor_min_interval_seconds: return cached context = _build_context(symbol, candles, ticker, pattern, learning, open_positions_for_symbol, account) prompt = _prompt(context, self.settings.llm_advisor_max_adjustment) response_text = "" error = "" try: response = requests.post( f"{self.settings.ollama_base_url}/api/generate", json={ "model": self.settings.ollama_model, "prompt": prompt, "stream": False, "options": {"temperature": 0.1}, }, timeout=self.settings.llm_advisor_timeout_seconds, ) response.raise_for_status() payload = response.json() response_text = str(payload.get("response", "")) advice = self._parse(symbol, response_text) except Exception as exc: error = str(exc) advice = LlmAdvice( symbol=symbol, enabled=True, model=self.settings.ollama_model, reason_ru="LLM Advisor временно недоступен; используется нейтральная поправка.", error=error, ) self._cache[symbol] = advice self.storage.insert_llm_advice( symbol=symbol, model=self.settings.ollama_model, prompt_json=context, response_text=response_text, advice_json=advice.as_dict(), error=error or advice.error, ) return advice def snapshot(self) -> dict[str, Any]: return { "enabled": self.settings.llm_advisor_enabled, "base_url": self.settings.ollama_base_url, "model": self.settings.ollama_model, "min_interval_seconds": self.settings.llm_advisor_min_interval_seconds, "max_adjustment": self.settings.llm_advisor_max_adjustment, "items": [advice.as_dict() for advice in self._cache.values()], } def _parse(self, symbol: str, response_text: str) -> LlmAdvice: data = _extract_json(response_text) regime = str(data.get("market_regime", "unknown")).strip().lower() risk = str(data.get("risk_level", "medium")).strip().lower() adjustment = _clamp_float( data.get("confidence_adjustment", 0.0), -self.settings.llm_advisor_max_adjustment, self.settings.llm_advisor_max_adjustment, ) return LlmAdvice( symbol=symbol, enabled=True, model=self.settings.ollama_model, market_regime=regime if regime in REGIMES else "unknown", risk_level=risk if risk in RISK_LEVELS else "medium", confidence_adjustment=adjustment, block_entry=bool(data.get("block_entry", False)), grid_suitable=bool(data.get("grid_suitable", False)), reason_ru=str(data.get("reason_ru", "LLM Advisor не объяснил вывод."))[:240], ) def _build_context( symbol: str, candles: list[Candle], ticker: Ticker | None, pattern: dict[str, Any], learning: dict[str, Any], open_positions_for_symbol: int, account: dict[str, float], ) -> dict[str, Any]: latest = candles[-1] if candles else None return { "mode": "paper_demo_only", "symbol": symbol, "objective": "reduce avoidable losing spot-long entries; do not promise profit", "market": { "last_price": ticker.last_price if ticker else None, "spread_percent": ticker.spread_percent if ticker else None, "turnover_24h": ticker.turnover_24h if ticker else None, "change_24h": ticker.change_24h if ticker else None, "close": latest.close if latest else None, "rsi_14": latest.rsi_14 if latest else None, "ema_20": latest.ema_20 if latest else None, "ema_50": latest.ema_50 if latest else None, "ema_200": latest.ema_200 if latest else None, "atr_14": latest.atr_14 if latest else None, "volume": latest.volume if latest else None, "volume_ma_20": latest.volume_ma_20 if latest else None, }, "pattern": pattern, "learning": learning, "risk_state": { "equity": account.get("equity"), "cash": account.get("cash"), "exposure": account.get("exposure"), "drawdown": account.get("drawdown"), "open_positions_for_symbol": open_positions_for_symbol, }, "allowed_output": { "market_regime": sorted(REGIMES), "risk_level": sorted(RISK_LEVELS), "confidence_adjustment": "number within configured bounds", "block_entry": "boolean; can only block buy, never force buy", "grid_suitable": "boolean", "reason_ru": "short Russian explanation", }, } def _prompt(context: dict[str, Any], max_adjustment: float) -> str: return ( "Ты LLM Advisor для paper-only crypto spot LONG бота. " "Ты не открываешь сделки и не обещаешь прибыль. " "Верни только валидный JSON без markdown. " f"confidence_adjustment должен быть от {-max_adjustment:.4f} до {max_adjustment:.4f}. " "Если рынок падающий, шаблон отрицательный или обучение убыточное, используй отрицательную поправку или block_entry=true. " "Если боковик и риск умеренный, можешь отметить grid_suitable=true. " "JSON keys: market_regime, risk_level, confidence_adjustment, block_entry, grid_suitable, reason_ru. " f"Context: {json.dumps(context, ensure_ascii=False, separators=(',', ':'))}" ) def _extract_json(text: str) -> dict[str, Any]: stripped = text.strip() if stripped.startswith("```"): stripped = stripped.strip("`").strip() if stripped.lower().startswith("json"): stripped = stripped[4:].strip() try: data = json.loads(stripped) except json.JSONDecodeError: start = stripped.find("{") end = stripped.rfind("}") if start < 0 or end <= start: raise data = json.loads(stripped[start : end + 1]) if not isinstance(data, dict): raise ValueError("LLM response JSON is not an object") return data def _clamp_float(value: Any, low: float, high: float) -> float: try: parsed = float(value) except (TypeError, ValueError): parsed = 0.0 return round(max(low, min(high, parsed)), 4) def _age_seconds(created_at: str) -> float: from datetime import datetime return (utc_now() - datetime.fromisoformat(created_at)).total_seconds()