"""The AI brain: a second opinion on trades, web research, post-mortems and plain-language conclusions. Hard limits (enforced in code, not by asking nicely): - the AI can only APPROVE, shrink (never enlarge) or skip a trade the rules already proposed - it cannot place or close trades, move stops, or change any risk limit - everything it says is validated and clamped (schema.py) - in 'shadow' mode it is recorded and measured but has NO effect - if the AI is down or slow, trading carries on exactly as without it (configurable) """ import logging import threading import time from dataclasses import dataclass, field import pandas as pd from . import prompts, schema from .llm import LLMError, Router, parse_json from .research import Research log = logging.getLogger("sailar.brain") @dataclass class Verdict: ok: bool = False decision: str = "APPROVE" size_mult: float = 1.0 # the multiplier that is actually APPLIED confidence: int = 0 event_risk: str = "none" reasons: list = field(default_factory=list) risks: list = field(default_factory=list) summary: str = "" provider: str = "" model: str = "" latency_ms: int = 0 log_id: int = None applied_skip: bool = False apply_reason: str = "" error: str = "" class Brain: def __init__(self, cfg, journal, env): self.cfg, self.j = cfg, journal self.c = cfg["brain"] self.router = Router(self.c, env) self.research = Research(self.router, journal, self.c, cfg["symbol"], env) self.router.calls_today = journal.brain_calls_today() self.router.day = time.strftime("%Y-%m-%d", time.gmtime()) self._rt = None self.last_research_error = "" # ------------------------------------------------------------------ state @property def mode(self): return self.c.get("mode", "shadow") @property def enabled(self): return self.mode in ("shadow", "active") and self.router.available() def _log(self, kind, now, version, reply=None, ok=True, applied="", summary="", response=None, trade_id=None, error=""): return self.j.log_brain(now, version, kind, reply.provider if reply else "", reply.model if reply else "", reply.latency_ms if reply else 0, ok, applied, summary, response, trade_id, error) # ------------------------------------------------------------------ 1. second opinion on a proposed trade def review(self, ctx, now, version=None): vd = Verdict() research = self.research.latest() if research: ctx["research"] = {k: research.get(k) for k in ("events", "headlines", "mood", "bias", "summary", "age_min")} black = self.research.blackout(now, self.c.get("blackout_before_min", 30), self.c.get("blackout_after_min", 15)) \ if (self.c.get("news_blackout", True) and research) else None reply = None try: reply = self.router.chat(prompts.SYS_REVIEW, prompts.review_user(ctx), json_mode=True, timeout=self.c.get("review_timeout", 8), max_tokens=1000, prefer="groq") v = schema.verdict(parse_json(reply.text)) vd.ok, vd.decision, vd.confidence = True, v["decision"], v["confidence"] vd.event_risk, vd.reasons, vd.risks, vd.summary = v["event_risk"], v["reasons"], v["risks"], v["summary"] vd.provider, vd.model, vd.latency_ms = reply.provider, reply.model, reply.latency_ms vd._mult = v["size_multiplier"] except LLMError as e: vd.error = str(e)[:240] vd._mult = 1.0 log.warning("AI review failed: %s", vd.error) # ---- decide what actually happens (only in ACTIVE mode) would = "" if black: would = f"skip: high-impact {black['currency']} news '{black['title']}' at {black['time_utc'][11:16]} UTC" elif vd.ok and vd.decision == "REJECT" and vd.confidence >= self.c.get("reject_min_confidence", 65): would = "skip: AI rejected the trade" elif vd.ok and vd.decision == "REDUCE": would = f"reduce size to {vd._mult:.2f}x" elif not vd.ok and self.c.get("on_error", "proceed") == "skip": would = "skip: AI unavailable and on_error is 'skip'" if self.mode == "active" and would: if would.startswith("skip"): vd.applied_skip, vd.apply_reason = True, would[6:] else: vd.size_mult, vd.apply_reason = max(0.25, min(1.0, vd._mult)), would vd.log_id = self._log("review", now, version, reply, vd.ok, ("APPLIED: " if self.mode == "active" else "SHADOW: ") + would if would else "", vd.summary, {"decision": vd.decision, "confidence": vd.confidence, "event_risk": vd.event_risk, "reasons": vd.reasons, "risks": vd.risks, "size_multiplier": getattr(vd, "_mult", 1.0), "blackout": black}, error=vd.error) return vd # ------------------------------------------------------------------ 2. research, kept fresh in the background def tick(self, now): """Called every loop. Starts a background refresh when the brief is old. Never blocks trading.""" if not (self.enabled and self.c.get("research", True)): return if self._rt is not None and self._rt.is_alive(): return if time.time() - getattr(self, "_last_try", 0) < 120 or not self.research.due(now): return self._last_try = time.time() def work(): try: self.research.refresh(now) self.last_research_error = "" except Exception as e: self.last_research_error = str(e)[:240] log.warning("research failed: %s", self.last_research_error) self._rt = threading.Thread(target=work, daemon=True) self._rt.start() # ------------------------------------------------------------------ 3. learn from each closed trade def postmortem_async(self, trade, version, now): if not (self.enabled and self.c.get("postmortems", True)): return data = {k: trade.get(k) for k in ("side", "entry", "exit", "sl", "tp", "pnl", "risk_amount", "exit_reason", "bars_held", "confidence")} data["result_r"] = round(trade["pnl"] / trade["risk_amount"], 2) if trade.get("risk_amount") else None data["factors_at_entry"] = trade.get("factors") def work(): try: reply = self.router.chat(prompts.SYS_POST, "TRADE (JSON):\n" + __import__("json").dumps(data, default=str), json_mode=True, timeout=20, max_tokens=400, prefer="groq") pm = schema.postmortem(parse_json(reply.text)) self._log("postmortem", now, version, reply, True, pm["category"], pm["lesson"], pm, trade_id=trade.get("id")) except Exception as e: log.warning("post-mortem failed: %s", e) threading.Thread(target=work, daemon=True).start() # ------------------------------------------------------------------ 4. plain-language conclusions def _ask(self, system, payload, key, prefer="gemini"): reply = self.router.chat(system, __import__("json").dumps(payload, default=str), json_mode=True, timeout=30, max_tokens=700, temperature=0.3, prefer=prefer) obj = parse_json(reply.text) return str(obj.get(key, ""))[:900], reply def failure_conclusion(self, version, stats, params, notes): if not self.enabled: return None try: lessons = [r["summary"] for r in self.j.brain_logs("postmortem", 12, version)] text, reply = self._ask(prompts.SYS_CONCLUDE, {"stats": stats, "params": params, "rule_based_findings": notes, "trade_lessons": lessons}, "conclusion") self._log("conclusion", pd.Timestamp.now(tz="UTC"), version, reply, True, "", text) return text except Exception as e: log.warning("failure conclusion failed: %s", e) return None def daily_note(self, version, stats, windows): if not self.enabled: return None try: text, reply = self._ask(prompts.SYS_DAILY, {"all_time": stats, "recent_windows": windows, "research": (self.research.latest() or {}).get("summary")}, "note") self._log("daily", pd.Timestamp.now(tz="UTC"), version, reply, True, "", text) return text except Exception as e: log.warning("daily note failed: %s", e) return None def advise(self, stats, params, backtest): reply = self.router.chat(prompts.SYS_ADVISE, __import__("json").dumps({"stats": stats, "params": params, "backtest": backtest}, default=str), json_mode=True, timeout=30, max_tokens=800, temperature=0.3, prefer="gemini") return schema.advice(parse_json(reply.text)) # ------------------------------------------------------------------ is the brain actually helping? def value(self): groups = {} for row in self.j.brain_trade_outcomes(): g = groups.setdefault(row["decision"], {"n": 0, "wins": 0, "r": 0.0, "net": 0.0}) g["n"] += 1 g["wins"] += 1 if row["pnl"] > 0 else 0 g["net"] += row["pnl"] g["r"] += row["pnl"] / row["risk_amount"] if row["risk_amount"] else 0.0 out = {} for k, g in groups.items(): out[k] = {"trades": g["n"], "win_rate": round(g["wins"] / g["n"], 3), "avg_r": round(g["r"] / g["n"], 3), "net": round(g["net"], 2)} rej = out.get("REJECT", {"trades": 0, "net": 0.0}) out["_summary"] = {"reviewed_trades": sum(g["n"] for g in groups.values()), "money_saved_if_rejects_were_skipped": round(-rej["net"], 2), "enough_data": sum(g["n"] for g in groups.values()) >= 30} return out def snapshot(self): now = pd.Timestamp.now(tz="UTC") reviews = self.j.brain_logs("review", 8) pms = self.j.brain_logs("postmortem", 40) cats = {} for p in pms: cats[p["applied"]] = cats.get(p["applied"], 0) + 1 r = self.research.latest() return { "mode": self.mode, "enabled": self.enabled, "providers": self.router.health(), "calls_today": self.router.calls_today, "budget": self.c.get("max_calls_per_day", 300), "last_error": self.router.last_error or self.last_research_error, "research": r, "research_error": self.last_research_error, "reviews": reviews, "postmortem_categories": cats, "postmortems": pms[:6], "conclusions": self.j.brain_logs("conclusion", 3), "value": self.value(), } def make_brain(cfg, journal, env=None): import os return Brain(cfg, journal, env if env is not None else os.environ)