""" ForexBot2 — LLM agent panel (advisory consensus) ================================================= ai-hedge-fund-style persona panel for XAU/USD signals: four DeepSeek personas (macro, technical, risk, contrarian) each review a finished SignalReport from their own angle, then a synthesizer chair weighs the takes and issues a consensus verdict (AGREE / NEUTRAL / OPPOSE with confidence and a short rationale). Two invariants, both load-bearing (CLAUDE.md rules 1 and 4): - ADVISORY / DISPLAY-ONLY. The verdict is written onto SignalReport fields that no gate and no scoring path ever reads. Any failure — LLM down, cap exhausted, garbage reply — degrades to "no panel", never to a changed or blocked signal. - NEVER runs inside ``analyze_from_dataframes``. The backtester replays that function; consumers (telegram_bot, main --panel, dashboard_publisher) attach the panel POST-HOC via ``attach_panel``, so replayed books can never see a panel field. Backtest parity holds by construction. Cost model: every full panel spends 5 requests (4 personas + 1 synthesizer) against the shared DeepSeek daily cap (llm_client). Two layers bound the spend on top of that cap: - ``max_panels_per_day`` (config, default 20) — a UTC-day budget of panel *attempts*, mirroring the client's request cap. - A reuse TTL (default 3600 s) keyed on (symbol, timeframe, direction): a persisting signal re-serves its cached verdict across bars instead of re-spending. Without this, one sticky M5 dashboard signal would burn 5 calls per 5-minute bar — 1,440/day against the 500 cap. Prompts are STATIC module constants (DeepSeek prefix-cache pricing — see news_engine.py); all variable content travels in the user payload. """ import datetime as _dt import logging import threading import time from collections import OrderedDict from concurrent.futures import ThreadPoolExecutor, wait as _fwait from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple from app_config import get_app_config from llm_client import DEFAULT_TIMEOUT, get_llm_client if TYPE_CHECKING: # type hints only — keep the coupling one-way from news_engine import FundamentalContext from signal_aggregator import SignalReport logger = logging.getLogger(__name__) DEFAULT_MIN_QUALITY = "B" DEFAULT_MAX_PANELS_PER_DAY = 20 DEFAULT_REUSE_TTL_SECONDS = 3600.0 # Quality ranks for the min_quality gate. Deliberately a local copy — # telegram_bot's ranking is private and dashboard_publisher must not # import telegram_bot (hard independence rule). _QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1} # ── DeepSeek prompts ───────────────────────────────────────────────── # STATIC constants on purpose: DeepSeek prices a repeated byte-identical # prompt prefix at the cache-hit rate (~50x cheaper), so nothing here may # be interpolated or timestamped. All variable content travels in the # user payload. Panel output is display-only — it never gates signals. _GOLD_DOCTRINE = """\ Gold doctrine: hawkish Fed / rate hikes / rising real yields / strong dollar are gold-bearish; easing / rate cuts / safe-haven flows / geopolitical stress / weak dollar are gold-bullish. """ _TAKE_CONTRACT = """\ Your review is advisory research on an already-generated signal; you are one voice on a panel, not the decision maker. Write the note in plain everyday English for someone who has never traded: short simple sentences, no trading jargon, no acronyms, no abbreviations. Say "profit versus loss", not "RR"; "the 4-hour chart", not "H4"; describe what an indicator MEANS ("the price has climbed unusually fast"), never its name or number. Reply with ONLY JSON: {"stance": "support"|"oppose"|"neutral", "confidence": <0-100>, "note": ""} "stance" says whether YOUR analysis supports the proposed trade direction, opposes it, or is genuinely mixed; "confidence" is how sure you are of your stance; "note" is your single most important point. """ _MACRO_SYSTEM = """\ You are the macro analyst on a gold (XAU/USD) trade-review panel. You receive JSON: a proposed trade brief ("brief": direction, timeframe, quality, price, regime) plus the current fundamental picture — recent "headlines", today's high-impact "events" (released: true means the number is out), a news "sentiment" score for gold, and an optional "news_summary" blurb. Judge ONE thing: does the macro backdrop support the proposed direction? Weigh released surprises over stale headlines; an imminent unreleased high-impact event is a reason for lower confidence, not a direction. """ + _GOLD_DOCTRINE + """\ Example: brief direction BUY, headlines dominated by hawkish Fed repricing and a hot CPI print -> {"stance": "oppose", "confidence": 70, "note": "Inflation is running hot, so interest rates should stay high — that usually pushes gold down."} """ + _TAKE_CONTRACT _TECH_SYSTEM = """\ You are the technical analyst on a gold (XAU/USD) trade-review panel. You receive JSON: a proposed trade brief ("brief") plus the technical evidence — per-"indicators" signal/confidence, multi-timeframe "mtf" signals and scores, "key_levels" versus the current price, and an "exhaustion" count of consecutive same-direction bars flagged by the engine. Judge ONE thing: is the technical picture internally coherent for the proposed direction? Look for indicators fighting each other, the signal fighting the regime (e.g. a BUY in trending_down), higher timeframes disagreeing, price pressed into a key level, or an exhausted move being chased. Example: brief direction SELL in regime trending_up with H1 and H4 both BUY -> {"stance": "oppose", "confidence": 80, "note": "This bets on gold falling while the bigger picture still points up."} """ + _TAKE_CONTRACT _RISK_SYSTEM = """\ You are the risk manager on a gold (XAU/USD) trade-review panel. You receive JSON: a proposed trade brief ("brief") plus the risk facts — the trade "setup" (entry, stop_loss, take_profit_1/2, risk_reward, risk_dollars), the "volatility" regime, engine "warnings", the calendar "event_risk" level, and the assumed "spread". Judge ONE thing: is this setup's risk acceptable as constructed? Check reward-to-risk adequacy, stop distance versus the volatility regime (a tight stop in extreme vol dies to noise; a huge stop in low vol is oversized), spread as a fraction of the stop, and event landmines about to hit. A missing setup means the signal graded too low for one — oppose unless the rest is exceptional. Example: RR 1.4 with volatility extreme and a high-impact event due -> {"stance": "oppose", "confidence": 65, "note": "The possible profit is small for such wild price swings, and big economic news is due soon."} """ + _TAKE_CONTRACT _CONTRARIAN_SYSTEM = """\ You are the contrarian on a gold (XAU/USD) trade-review panel — the designated skeptic. You receive JSON: a proposed trade brief ("brief"), the engine's own "bullish_reasons" and "bearish_reasons", its "warnings", and an optional "news_summary". Your job: construct the strongest honest case AGAINST the proposed trade, then score how damaging that case is. Use "oppose" only when the counter-case is genuinely strong, "neutral" when it is real but modest, and "support" when even the skeptic finds nothing that bites. """ + _GOLD_DOCTRINE + """\ Example: a BUY whose own engine lists three bearish reasons and a counter-trend warning -> {"stance": "oppose", "confidence": 60, "note": "The system's own list of negatives looks stronger than its reason to buy."} """ + _TAKE_CONTRACT _SYNTH_SYSTEM = """\ You are the chair of a gold (XAU/USD) trade-review panel. You receive JSON: the proposed trade "direction" and the panel's "takes" — one per analyst ("persona", "stance", "confidence", "note"). Weigh the takes into ONE consensus on whether to act on the signal: "AGREE" (the panel backs the trade), "OPPOSE" (the panel is against it), or "NEUTRAL" (genuinely split or low conviction). Weigh a high- confidence oppose from any single analyst seriously — one strong veto can outweigh two lukewarm supports. Do not average blindly. Write the rationale in plain everyday English for someone who has never traded: short simple sentences, no trading jargon, no acronyms, no abbreviations. Example rationale: "Three of four analysts are against this: the possible profit is too small and the price has already climbed unusually fast. Only the news backdrop supports buying." Reply with ONLY JSON: {"verdict": "AGREE"|"NEUTRAL"|"OPPOSE", "confidence": <0-100>, "rationale": ""} """ # ── Result types ───────────────────────────────────────────────────── @dataclass class PersonaTake: persona: str # "macro" | "technical" | "risk" | "contrarian" stance: str # "support" | "oppose" | "neutral" confidence: float # 0-100 note: str @dataclass class PanelVerdict: verdict: str # "AGREE" | "NEUTRAL" | "OPPOSE" confidence: float # 0-100 rationale: str takes: List[PersonaTake] = field(default_factory=list) created_ts: float = 0.0 # ── Payload builders (pure functions; llm_client json-dumps sorted) ── def _direction(report: "SignalReport") -> Optional[str]: """BUY/SELL the panel should review; None = nothing to review.""" if report.trade_setup is not None: return report.trade_setup.direction if report.final_signal.value > 0: return "BUY" if report.final_signal.value < 0: return "SELL" return None def _brief(report: "SignalReport") -> Dict[str, Any]: analysis = report.primary_analysis return { "pair": report.display_name, "timeframe": analysis.timeframe if analysis else "", "direction": _direction(report), "quality": report.signal_quality, "score": round(report.final_score, 1), "confidence": round(report.final_confidence), "price": round(report.current_price, 2), "market_regime": report.market_regime, "volatility_regime": report.volatility_regime, "mtf_agreement": report.mtf_agreement, } def _macro_payload(report: "SignalReport", context: Optional["FundamentalContext"]) -> Dict[str, Any]: headlines: List[str] = [] events: List[Dict[str, Any]] = [] if context is not None: headlines = [h.headline for h in context.news_headlines[:10]] events = [ {"event": e.event, "currency": e.currency, "actual": e.actual, "forecast": e.forecast, "released": not e.is_upcoming} for e in context.events_today if e.is_high_impact ][:8] return { "brief": _brief(report), "headlines": headlines, "events": events, "sentiment": round(report.news_sentiment, 3), "news_summary": report.news_summary, } def _technical_payload(report: "SignalReport") -> Dict[str, Any]: indicators: Dict[str, Dict[str, Any]] = {} key_levels: Dict[str, float] = {} analysis = report.primary_analysis if analysis is not None: # Same visibility filter as main.report_to_dict: private series # ride as _-prefixed pseudo-indicators, weight-0 = context-only. indicators = { name: {"signal": ind.signal.name, "confidence": round(ind.confidence)} for name, ind in analysis.indicators.items() if not name.startswith("_") and ind.weight != 0 } key_levels = {k: round(v, 2) for k, v in analysis.key_levels.items()} return { "brief": _brief(report), "indicators": indicators, "mtf": {tf: {"signal": sig.name, "score": round(report.mtf_scores.get(tf, 0.0), 1)} for tf, sig in report.mtf_signals.items()}, "key_levels": key_levels, "exhaustion": report.exhaustion_count, } def _risk_payload(report: "SignalReport") -> Dict[str, Any]: payload: Dict[str, Any] = { "brief": _brief(report), "volatility": report.volatility_regime, "warnings": list(report.warnings), "event_risk": report.event_risk, "event_description": report.event_description, "spread": round(report.spread_estimate, 2), } ts = report.trade_setup if ts is not None: payload["setup"] = { "direction": ts.direction, "entry": round(ts.entry_price, 2), "stop_loss": round(ts.stop_loss, 2), "take_profit_1": round(ts.take_profit_1, 2), "take_profit_2": round(ts.take_profit_2, 2), "risk_reward": round(ts.risk_reward, 2), "risk_dollars": round(ts.risk_pips * ts.pip_value, 2), } return payload def _contrarian_payload(report: "SignalReport") -> Dict[str, Any]: return { "brief": _brief(report), "bullish_reasons": report.bullish_reasons[:3], "bearish_reasons": report.bearish_reasons[:3], "warnings": list(report.warnings), "news_summary": report.news_summary, } # ── Orchestration ──────────────────────────────────────────────────── # Panel-private pool: the 4 persona calls run concurrently WITHOUT # occupying telegram_bot's shared scan executor (which has 4 workers — # borrowing them would stall every other scan for the panel's duration). _pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="agent-panel") _VALID_STANCES = ("support", "oppose", "neutral") _VALID_VERDICTS = ("AGREE", "NEUTRAL", "OPPOSE") def _clamp_confidence(raw: Any) -> Optional[float]: try: return max(0.0, min(100.0, float(raw))) except (TypeError, ValueError): return None def _validate_take(persona: str, reply: Any) -> Optional[PersonaTake]: if not isinstance(reply, dict): return None stance = reply.get("stance") if stance not in _VALID_STANCES: return None confidence = _clamp_confidence(reply.get("confidence")) if confidence is None: return None note = str(reply.get("note", "")).strip()[:160] return PersonaTake(persona=persona, stance=stance, confidence=confidence, note=note) def _ask_persona(persona: str, system: str, payload: Dict[str, Any]) -> Optional[PersonaTake]: # Client resolved at call time: tests install stubs via lc._client. reply = get_llm_client().chat_json(system, payload) return _validate_take(persona, reply) def run_panel(report: "SignalReport", context: Optional["FundamentalContext"] = None, ) -> Optional[PanelVerdict]: """Run 4 personas + synthesizer; None = no verdict (fail-safe). Requires ≥2 valid persona takes before spending the synthesizer request; the synthesizer's reply is validated strictly and there is deliberately NO programmatic fallback verdict — a panel that could not really convene reports nothing rather than something invented. """ direction = _direction(report) if direction is None: return None jobs = ( ("macro", _MACRO_SYSTEM, _macro_payload(report, context)), ("technical", _TECH_SYSTEM, _technical_payload(report)), ("risk", _RISK_SYSTEM, _risk_payload(report)), ("contrarian", _CONTRARIAN_SYSTEM, _contrarian_payload(report)), ) futures = [_pool.submit(_ask_persona, name, system, payload) for name, system, payload in jobs] # The HTTP timeout inside the client bounds each call; the margin # covers pool queueing. Stragglers past the wait are abandoned (their # threads finish harmlessly) and simply don't vote. client_timeout = getattr(get_llm_client(), "_timeout", DEFAULT_TIMEOUT) done, _ = _fwait(futures, timeout=client_timeout + 8.0) takes = [f.result() for f in done if not f.exception()] takes = [t for t in takes if t is not None] if len(takes) < 2: logger.debug("agent panel: %d/4 persona takes — skipping synthesizer", len(takes)) return None order = {"macro": 0, "technical": 1, "risk": 2, "contrarian": 3} takes.sort(key=lambda t: order.get(t.persona, 9)) reply = get_llm_client().chat_json(_SYNTH_SYSTEM, { "direction": direction, "takes": [{"persona": t.persona, "stance": t.stance, "confidence": t.confidence, "note": t.note} for t in takes], }) if not isinstance(reply, dict): return None verdict = reply.get("verdict") confidence = _clamp_confidence(reply.get("confidence")) if verdict not in _VALID_VERDICTS or confidence is None: return None rationale = str(reply.get("rationale", "")).strip()[:240] return PanelVerdict(verdict=verdict, confidence=confidence, rationale=rationale, takes=takes, created_ts=time.time()) # ── Cache, budget, attach ──────────────────────────────────────────── _state_lock = threading.Lock() # Exact result per reviewed bar: repeated attaches on the same signal # (re-scan, multi-consumer) are free. _cache: "OrderedDict[Tuple[str, str, str, str], PanelVerdict]" = OrderedDict() _CACHE_MAX = 32 # Reuse layer: a persisting signal (same symbol/tf/direction, newer bar) # re-serves its last verdict for reuse_ttl_seconds instead of re-spending # 5 requests per bar (see module docstring for the M5 dashboard math). _last_by_dir: Dict[Tuple[str, str, str], PanelVerdict] = {} # UTC-day budget of panel attempts (mirrors llm_client._under_cap). _panel_day: Optional[_dt.date] = None _panel_count: int = 0 _warned_budget = False def reset_panel_state() -> None: """Drop caches and counters (tests / config reloads).""" global _panel_day, _panel_count, _warned_budget with _state_lock: _cache.clear() _last_by_dir.clear() _panel_day = None _panel_count = 0 _warned_budget = False def panel_health() -> Dict[str, Any]: """Read-only counters for status displays.""" settings = get_app_config().agent_panel_settings cap = int(settings.get("max_panels_per_day", DEFAULT_MAX_PANELS_PER_DAY)) today = _dt.datetime.now(_dt.timezone.utc).date() with _state_lock: used = _panel_count if _panel_day == today else 0 return {"used_today": used, "cap": cap} def _under_panel_budget(cap: int) -> bool: """Count one panel attempt against the UTC-day budget.""" global _panel_day, _panel_count, _warned_budget today = _dt.datetime.now(_dt.timezone.utc).date() with _state_lock: if _panel_day != today: _panel_day = today _panel_count = 0 _warned_budget = False if _panel_count >= cap: if not _warned_budget: _warned_budget = True logger.warning( "agent panel: daily budget (%d) exhausted — no more " "panels until UTC midnight", cap) return False _panel_count += 1 return True def _write_fields(report: "SignalReport", v: PanelVerdict) -> None: report.panel_verdict = v.verdict report.panel_confidence = float(v.confidence) report.panel_rationale = v.rationale report.panel_takes = [ {"persona": t.persona, "stance": t.stance, "confidence": t.confidence, "note": t.note} for t in v.takes] def attach_panel(report: "SignalReport", settings: Optional[Dict[str, Any]] = None, surface: str = "telegram", force: bool = False, context: Optional["FundamentalContext"] = None) -> bool: """Best-effort panel attach; True iff panel fields were populated. Advisory feature contract: NEVER raises — an exploding panel must not cost an alert send or a dashboard publish. ``force`` (CLI --panel) bypasses the enabled/surface/quality toggles but never the daily budget or the shared LLM cap. """ try: if settings is None: settings = get_app_config().agent_panel_settings if not force: if not settings.get("enabled", False): return False surface_on = { "telegram": settings.get("telegram", True), "dashboard": settings.get("dashboard", False), }.get(surface, True) if not surface_on: return False min_quality = settings.get("min_quality", DEFAULT_MIN_QUALITY) if (_QUALITY_RANK.get(report.signal_quality, -1) < _QUALITY_RANK.get(min_quality, 2)): return False direction = _direction(report) if direction is None: return False analysis = report.primary_analysis tf = analysis.timeframe if analysis else "" exact_key = (report.symbol, tf, report.timestamp, direction) dir_key = (report.symbol, tf, direction) ttl = float(settings.get("reuse_ttl_seconds", DEFAULT_REUSE_TTL_SECONDS)) with _state_lock: hit = _cache.get(exact_key) if hit is None: prior = _last_by_dir.get(dir_key) if prior is not None and time.time() - prior.created_ts < ttl: hit = prior if hit is not None: _write_fields(report, hit) return True cap = int(settings.get("max_panels_per_day", DEFAULT_MAX_PANELS_PER_DAY)) if not _under_panel_budget(cap): return False if context is None: # Best-effort; the news engine's class-level 600 s caches make # this nearly free when warm in both bot and dashboard. try: from config import PlatformConfig from news_engine import NewsEngine context = NewsEngine( PlatformConfig().news).get_fundamental_context() except Exception as e: logger.debug("agent panel: news context unavailable: %s", e) context = None verdict = run_panel(report, context) if verdict is None: return False _write_fields(report, verdict) with _state_lock: _cache[exact_key] = verdict while len(_cache) > _CACHE_MAX: _cache.popitem(last=False) _last_by_dir[dir_key] = verdict return True except Exception: logger.exception("agent panel: attach failed — signal unaffected") return False