""" 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). Verdict source (2026-09-03, owner decision): all four personas always run and are logged, but by default (``agent_panel.verdict_source = "macro"``) the verdict shown to consumers is the macro persona's own stance, and the synthesizer call is skipped (4 LLM calls per panel instead of 5); ``"chair"`` restores the synthesizer + OPPOSE-guard path below. 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 the complete reviewed evidence payload, ignoring only the bar timestamp. Changed price, setup, news, warnings, indicators, or candles always trigger a fresh review (subject to the daily budget); genuinely identical evidence can be re-served. Prompts are STATIC module constants (DeepSeek prefix-cache pricing — see news_engine.py); all variable content travels in the user payload. Payload enrichment (2026-08-15/23, owner-approved, advisory-only): the technical persona receives strategy/pattern context, raw indicator VALUES (with a units legend in the prompt), and ATR-NORMALIZED candles per timeframe (code does the arithmetic — raw prices are the model's weakest input; see backtest/research/llm_news_eval_2026-08-06.md). Sized by config ``candle_bars_primary`` (default 20) / ``candle_bars_confirm`` (default 10); ``candles_enabled: false`` reverts to the pre-enrichment payload. Prompts forbid outside/training-data knowledge and cap notes at 15 words / rationales at 30 (output tokens are the pricey side). Macro and contrarian receive only code-scored directional news; contrarian also receives the timestamped closed-market snapshot, complete setup, and visible indicators. Risk receives the stamped trailing policy and strategy cost assumption, never the candle range mislabeled as a spread. Verdict logging writes the verdict, creation time, cache provenance, payload hash, and a pending deterministic advisory outcome onto the scan's signals.db row via signal_log's rowid hand-off. No field is read by live behavior. """ import datetime as _dt import hashlib import json import logging import math 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 config import Timeframe 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 DEFAULT_CANDLE_BARS_PRIMARY = 20 DEFAULT_CANDLE_BARS_CONFIRM = 10 DEFAULT_VERDICT_SOURCE = "macro" # Maps a persona's stance straight onto the verdict vocabulary when the # macro take stands in for the chair. _STANCE_TO_VERDICT = {"support": "AGREE", "neutral": "NEUTRAL", "oppose": "OPPOSE"} # 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. Hard rules: use ONLY the data in the JSON you receive — do not invent numbers, events, or price levels. Your training data about gold, the markets, and current events is stale and OFF-LIMITS: never use outside knowledge; if a fact is not in the JSON, treat it as unknown. If the evidence your role needs is thin or missing, answer "neutral" with confidence below 30, unless your role's instructions say otherwise. Confidence scale (0-100): 80-100 strong one-sided evidence; 50-79 solid case with some doubt; 30-49 mixed evidence; below 30 little basis to judge. 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 directional "headlines", today's high-impact "events" (released: true means the number is out), and optional aggregate "sentiment" for gold. Neutral and unrelated headlines were removed by code. Each remaining headline has a code-computed "gold_influence" and score; trust that direction and do not reinterpret it from the wording alone. 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. Stance rules: - support: the released data and news flow clearly push gold in the proposed direction. - oppose: they clearly push gold the other way. - neutral: the picture is mixed or stale — or there are no meaningful headlines or events to judge (score that low confidence). """ + _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."} Example: brief direction BUY, headlines led by an escalating conflict and a soft jobs report -> {"stance": "support", "confidence": 75, "note": "Worries about war and a cooling economy usually send money into gold."} """ + _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 — the setup "strategy" and optional "setup_pattern", per-"indicators" signal/confidence/value, multi-timeframe "mtf" signals and scores, "key_levels" versus the current price, an "exhaustion" count of consecutive same-direction bars flagged by the engine, and optionally "candles". Indicator "value" units: rsi 0-100 (above 70 overbought, below 30 oversold); adx 0-100 (above 25 = strong trend); macd = histogram in dollars (momentum); roc = percent; bollinger = position 0-1 (0 = lower band, 1 = upper band, values may extend outside the bands); keltner = dollar distance from the mid-line; ema_trend = -1 to +1 alignment score; divergence and candlestick = pattern counts. RSI is interpreted in the supplied market regime, not as an automatic reversal at 70/30. "candles" is per timeframe: {"ref_close", "atr", "bars"} — bars are oldest-first, each [open, high, low, close] expressed in ATR units relative to ref_close: 0.0 = the latest close, +1.0 = one ATR above it, -2.0 = two ATR below it. Read the SHAPE (drift, swings, wicks); never quote these as prices. 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, an exhausted move being chased, or the candle path contradicting the indicator readings. For XAU_CP, the named completed reversal/continuation pattern is the setup trigger and may lead lagging indicators; indicator lag alone is not sufficient reason to oppose it. Judge whether the pattern and the remaining evidence are coherent together. Stance rules: - support: the indicators, higher timeframes, and regime mostly agree with the proposed direction and none of the conflicts above appear. - oppose: one or more of those conflicts is present and serious. - neutral: the evidence genuinely cuts both ways, or too much of it is missing to judge (score that low confidence). 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."} Example: brief direction BUY in regime trending_up with M15 and H1 both BUY and exhaustion 0 -> {"stance": "support", "confidence": 70, "note": "Every way of looking at the chart points up, so buying goes with the flow."} """ + _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" (strategy, optional setup pattern, entry, stop, advisory targets, risk, strategy cost assumption, and trailing policy), the "volatility" regime, engine "warnings", and calendar "event_risk". When trailing.enabled is true, TP1/TP2 are reference levels rather than hard exits: profit can run beyond them, and the stop starts ratcheting after arm_r using distance_r. Never evaluate that setup as a static TP1 trade. "round_trip_cost" is a strategy assumption, not a broker or data- provider quote. Judge ONE thing: is this setup's risk acceptable as constructed? Check stop distance versus volatility, cost as a fraction of risk, trailing geometry, and event landmines about to hit. Stance rules: - support: the trailing opportunity justifies the risk, the stop fits the current volatility, and no event landmine is imminent. - oppose: a stop/trailing policy mismatched to volatility, strategy cost eating a big share of the stop, an imminent high-impact event — or no setup at all (a missing setup means the signal graded too low for one; oppose unless the rest is exceptional). - neutral: acceptable but unremarkable — nothing dangerous, nothing attractive. Example: RR 1.4 with volatility extreme and a high-impact event due -> {"stance": "oppose", "confidence": 65, "note": "The stop and trailing plan do not suit wild price swings before major news."} Example: trailing enabled with normal volatility, low cost versus risk, and no events due -> {"stance": "support", "confidence": 70, "note": "The stop and trailing plan fit calm conditions, with low costs and no major news."} """ + _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 timestamped latest-closed-bar "market_snapshot", the complete trade "setup" and strategy context, visible primary-timeframe "indicators", filtered directional "news", the engine's bullish/bearish reasons, and its warnings. Neutral and unrelated news was removed by code; trust each headline's code-computed gold_influence instead of reinterpreting it. 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). The risk manager and contrarian have intentionally defensive roles; their objections are not automatic vetoes: "OPPOSE" requires that the macro analyst opposes, or that the technical analyst opposes while the macro analyst is not supportive. When only the defensive roles object, or the macro analyst supports the trade, the most the panel can say is "NEUTRAL". Do not count repeated versions of the same warning as independent evidence. Weigh role relevance, evidence quality, and confidence rather than averaging blindly. Base the verdict and rationale ONLY on the takes given — do not add outside analysis. Confidence scale (0-100): 80-100 the panel clearly leans one way; 50-79 a lean with real dissent; below 50 split or low conviction. 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_hash: str = "" source: str = "chair" # "macro" | "chair" — which path produced this # ── Payload builders (pure functions; llm_client json-dumps sorted) ── def _finite(x: Any) -> Optional[float]: """float(x) or None for non-numeric / non-finite (NaN/inf breaks the strict-JSON wire format the client sends).""" try: f = float(x) except (TypeError, ValueError): return None return f if math.isfinite(f) else None 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 _primary_indicator_payload(report: "SignalReport") -> Dict[str, Dict[str, Any]]: indicators: Dict[str, Dict[str, Any]] = {} analysis = report.primary_analysis if analysis is None: return indicators for name, ind in analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue entry: Dict[str, Any] = { "signal": ind.signal.name, "confidence": round(ind.confidence), } value = _finite(getattr(ind, "value", None)) if value is not None: entry["value"] = round(value, 2) indicators[name] = entry return indicators def _strategy_context(report: "SignalReport") -> Dict[str, Any]: setup = report.trade_setup code = getattr(setup, "strategy", None) if setup is not None else None meanings = { "XAU_ATR": "gold ATR setup produced by the scored technical signal", "XAU_PB": "trend-continuation pullback to EMA21 or Keltner mid-line", "XAU_CP": "completed classical chart-pattern entry", "GENERIC": "swing and ATR setup produced by the scored technical signal", } out: Dict[str, Any] = { "code": code or "NO_SETUP", "meaning": meanings.get(code, "no executable trade setup was produced"), } pattern = getattr(setup, "setup_pattern", None) if setup is not None else None if pattern: out["setup_pattern"] = str(pattern) return out def _trade_setup_payload(report: "SignalReport") -> Optional[Dict[str, Any]]: setup = report.trade_setup if setup is None: return None entry = _finite(getattr(setup, "entry_price", None)) stop = _finite(getattr(setup, "stop_loss", None)) risk = abs(entry - stop) if entry is not None and stop is not None else None cost = _finite(getattr(setup, "round_trip_cost", None)) arm = _finite(getattr(setup, "trail_arm_r", None)) distance = _finite(getattr(setup, "trail_dist_r", None)) out: Dict[str, Any] = { "direction": getattr(setup, "direction", None), "strategy": _strategy_context(report), "entry": round(entry, 2) if entry is not None else None, "stop_loss": round(stop, 2) if stop is not None else None, "take_profit_1": round(float(setup.take_profit_1), 2), "take_profit_2": round(float(setup.take_profit_2), 2), "risk_reward_to_advisory_tp1": round(float(setup.risk_reward), 2), "risk_price_distance": round(risk, 2) if risk is not None else None, "round_trip_cost": round(cost, 2) if cost is not None else None, "trailing": { "enabled": bool(arm is not None and arm > 0), "arm_r": round(arm, 2) if arm is not None else None, "distance_r": round(distance, 2) if distance is not None else None, "tp1_tp2_are_advisory": bool(arm is not None and arm > 0), }, } analysis = report.primary_analysis atr = _finite((analysis.key_levels or {}).get("atr")) \ if analysis is not None else None if atr is not None and atr > 0 and risk is not None: out["atr"] = round(atr, 2) out["stop_distance_atr"] = round(risk / atr, 2) if risk and cost is not None: out["cost_to_risk_pct"] = round(100.0 * cost / risk, 2) return out def _market_snapshot(report: "SignalReport") -> Dict[str, Any]: return { "price": round(report.current_price, 2), "bar_timestamp": report.timestamp, "basis": "latest_closed_candle", } def _headline_gold_score(headline: Any) -> Optional[float]: """Deterministic XAU/USD influence from an already-scored headline.""" by_currency = getattr(headline, "sentiment_by_currency", None) or {} xau = _finite(by_currency.get("XAU")) usd = _finite(by_currency.get("USD")) if xau is not None or usd is not None: score = (xau or 0.0) - (usd or 0.0) else: mentioned = set(getattr(headline, "currencies_mentioned", None) or []) raw = _finite(getattr(headline, "sentiment_score", None)) if raw is None: return None if "XAU" in mentioned and "USD" in mentioned: return None # pair direction is ambiguous without per-currency scores if "XAU" in mentioned: score = raw elif "USD" in mentioned: score = -raw else: return None return max(-1.0, min(1.0, score)) if math.isfinite(score) else None def _directional_news_payload( context: Optional["FundamentalContext"], limit: int = 10 ) -> List[Dict[str, Any]]: if context is None: return [] out: List[Dict[str, Any]] = [] for headline in context.news_headlines: score = _headline_gold_score(headline) if score is None or abs(score) < 0.1: continue timestamp = getattr(headline, "timestamp", None) out.append({ "headline": str(getattr(headline, "headline", "")), "published_at": (timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp or "")), "gold_influence": "bullish" if score > 0 else "bearish", "score": round(score, 3), }) if len(out) >= limit: break return out def _aggregate_sentiment_payload(report: "SignalReport") -> Optional[Dict[str, Any]]: score = _finite(report.news_sentiment) if score is None or abs(score) < 0.1: return None return { "gold_influence": "bullish" if score > 0 else "bearish", "score": round(score, 3), } def _macro_payload(report: "SignalReport", context: Optional["FundamentalContext"]) -> Dict[str, Any]: events: List[Dict[str, Any]] = [] if context is not None: events = [ {"event": e.event, "currency": e.currency, "time": e.time.isoformat(), "actual": e.actual, "forecast": e.forecast, "previous": e.previous, "surprise": e.surprise, "minutes_until": round(e.minutes_until, 1), "released": not e.is_upcoming} for e in context.events_today if e.is_high_impact ][:8] payload: Dict[str, Any] = { "brief": _brief(report), "headlines": _directional_news_payload(context), "events": events, } sentiment = _aggregate_sentiment_payload(report) if sentiment is not None: payload["sentiment"] = sentiment return payload def _technical_payload(report: "SignalReport", candles: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: key_levels: Dict[str, float] = {} analysis = report.primary_analysis if analysis is not None: key_levels = {k: round(v, 2) for k, v in analysis.key_levels.items()} payload: Dict[str, Any] = { "brief": _brief(report), "strategy": _strategy_context(report), "indicators": _primary_indicator_payload(report), "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, } if candles: payload["candles"] = candles return payload 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, } setup = _trade_setup_payload(report) if setup is not None: payload["setup"] = setup return payload def _contrarian_payload( report: "SignalReport", context: Optional["FundamentalContext"] = None ) -> Dict[str, Any]: return { "brief": _brief(report), "market_snapshot": _market_snapshot(report), "setup": _trade_setup_payload(report), "indicators": _primary_indicator_payload(report), "news": _directional_news_payload(context), "bullish_reasons": report.bullish_reasons[:3], "bearish_reasons": report.bearish_reasons[:3], "warnings": list(report.warnings), } # ── Candle context (ATR-normalized; technical persona only) ───────── # # Raw prices are the LLM's weakest input modality (measured in-repo: # numeric-comparison misreads, backtest/research/llm_news_eval_2026-08-06.md), # so code does the arithmetic and the model reads a normalized SHAPE: # each bar's [o,h,l,c] as offsets from the latest close in ATR units. # A panel-private provider singleton keeps the TTL cache warm across # attaches; any failure degrades to "no candles", never a failed panel. _TF_BY_VALUE = {tf.value: tf for tf in Timeframe} # Patchable for tests (mirrors the llm_client._client stub pattern). _dp: Any = None def _panel_provider() -> Any: global _dp if _dp is None: from config import PlatformConfig from providers import make_data_provider _dp = make_data_provider(PlatformConfig()) return _dp def _normalize_candles(df: Any, atr: float, bars: int) -> Dict[str, Any]: """Last `bars` closed candles as ATR-unit offsets from the final close: 0.0 = latest close, +1.0 = one ATR above it. Oldest-first.""" tail = df.tail(bars) ref = float(tail["Close"].iloc[-1]) out = [] for _, row in tail.iterrows(): out.append([round((float(row["Open"]) - ref) / atr, 1), round((float(row["High"]) - ref) / atr, 1), round((float(row["Low"]) - ref) / atr, 1), round((float(row["Close"]) - ref) / atr, 1)]) return {"ref_close": round(ref, 2), "atr": round(float(atr), 2), "bars": out} def _candle_context(report: "SignalReport", settings: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Best-effort {"": _normalize_candles(...)} per analysis TF; None = omit the key from the payload. NEVER raises.""" try: if not settings.get("candles_enabled", True): return None confirmations = list(report.confirmation_analyses or []) analyses = [report.primary_analysis] + confirmations counts = [int(settings.get("candle_bars_primary", DEFAULT_CANDLE_BARS_PRIMARY))] + [ int(settings.get("candle_bars_confirm", DEFAULT_CANDLE_BARS_CONFIRM)) ] * len(confirmations) dp = _panel_provider() out: Dict[str, Any] = {} for analysis, n in zip(analyses, counts): if analysis is None: continue atr = _finite(getattr( (analysis.indicators or {}).get("atr"), "value", None)) if not atr or atr <= 0: continue tf = _TF_BY_VALUE.get(analysis.timeframe) if tf is None: continue df = dp.fetch_ohlcv(report.symbol, tf) if df is None or len(df) < 2: continue out[analysis.timeframe] = _normalize_candles(df, atr, n) return out or None except Exception as e: logger.debug("agent panel: candle context unavailable: %s", e) return None # ── 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 _panel_jobs( report: "SignalReport", context: Optional["FundamentalContext"], candles: Optional[Dict[str, Any]], ) -> Tuple[Tuple[str, str, Dict[str, Any]], ...]: return ( ("macro", _MACRO_SYSTEM, _macro_payload(report, context)), ("technical", _TECH_SYSTEM, _technical_payload(report, candles)), ("risk", _RISK_SYSTEM, _risk_payload(report)), ("contrarian", _CONTRARIAN_SYSTEM, _contrarian_payload(report, context)), ) # NOTE (2026-09 audit): the TTL reuse key differs from the exact key # only by the contrarian's bar timestamp, and every other block (price, # ATR-normalized candles, indicator readings, drifting entry/stop) moves # each closed bar — so TTL_REUSE never fired live (30/30 attaches MISS). # That is the documented, test-locked decision ("changed evidence never # inherits a direction-only cached verdict"); spend is bounded by # max_panels_per_day instead. Relaxing the identity is an owner call. def _panel_payload_hash( jobs: Tuple[Tuple[str, str, Dict[str, Any]], ...], ignore_bar_timestamp: bool = False, ) -> str: client = get_llm_client() personas = [] for persona, system, payload in jobs: payload_copy = json.loads(json.dumps(payload, allow_nan=False)) if ignore_bar_timestamp and persona == "contrarian": snapshot = payload_copy.get("market_snapshot") if isinstance(snapshot, dict): snapshot.pop("bar_timestamp", None) personas.append({ "persona": persona, "system": system, "payload": payload_copy, }) envelope = { "contract": "agent-panel-input-v2", "model": getattr(client, "_model", client.__class__.__name__), "personas": personas, "synthesizer_system": _SYNTH_SYSTEM, } raw = json.dumps( envelope, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False, ).encode("utf-8") return "sha256:" + hashlib.sha256(raw).hexdigest() def run_panel(report: "SignalReport", context: Optional["FundamentalContext"] = None, candles: Optional[Dict[str, Any]] = None, jobs: Optional[Tuple[Tuple[str, str, Dict[str, Any]], ...]] = None, verdict_source: str = DEFAULT_VERDICT_SOURCE, ) -> Optional[PanelVerdict]: """Run 4 personas, then either take the macro persona's stance as the verdict (``verdict_source="macro"``, default — the synthesizer is not called) or weigh all four takes through the synthesizer chair (``"chair"``); None = no verdict (fail-safe). Requires ≥2 valid persona takes before spending a 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 = jobs or _panel_jobs(report, context, candles) payload_hash = _panel_payload_hash(jobs) 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 not takes: logger.debug("agent panel: 0/4 persona takes — no verdict") return None order = {"macro": 0, "technical": 1, "risk": 2, "contrarian": 3} takes.sort(key=lambda t: order.get(t.persona, 9)) if verdict_source == "macro": macro = next((t for t in takes if t.persona == "macro"), None) if macro is not None: return PanelVerdict( verdict=_STANCE_TO_VERDICT[macro.stance], confidence=macro.confidence, rationale=macro.note[:240], takes=takes, created_ts=time.time(), payload_hash=payload_hash, source="macro") # Macro take missing (timed out / invalid) — fall back to the chair. # Quorum guards the CHAIR only: it weighs several voices, so one lone # take is not worth a synthesizer request. The macro path above needs # only its own take and must never be gated by it — that gate used to # sit before the macro branch and silently dropped a good macro # verdict whenever the other three personas were slow. if len(takes) < 2: logger.debug("agent panel: %d/4 persona takes — skipping synthesizer", len(takes)) return None 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] verdict, confidence, rationale = _apply_oppose_guard( verdict, confidence, rationale, takes) return PanelVerdict(verdict=verdict, confidence=confidence, rationale=rationale, takes=takes, created_ts=time.time(), payload_hash=payload_hash, source="chair") # Deterministic backstop for the chair (ADVISORY-ONLY, no gate reads it). # # Evidence (2026-09 signal-log audit, 133 live verdicts, 107 with a # replayed outcome): the contrarian and risk personas opposed 117/133 # alerts each and the technical persona 88/133 — they object to the # trend extension the engine already discounted — so the chair said # OPPOSE 73% of the time while the OPPOSED trades made +30.1R (PF 2.6). # The macro persona was the only informative voice: macro=support # trades won 88% (+0.56R mean, n=69); macro=oppose won 25%; macro=neutral # 38%. With this rule the OPPOSE bucket becomes the weakest one # (n=29, +0.08R mean, 41% win) instead of the best. Small sample, one # bullish episode — re-score with analyze_signal_log --stored-candles # after more history before tightening further. # # Rule: OPPOSE stands only if the macro analyst opposes, or the technical # analyst opposes while the macro analyst is not supportive. Otherwise # the verdict is NEUTRAL (confidence capped below 50). def _apply_oppose_guard(verdict: str, confidence: float, rationale: str, takes: List[PersonaTake]): if verdict != "OPPOSE": return verdict, confidence, rationale stance = {t.persona: t.stance for t in takes} macro = stance.get("macro") technical = stance.get("technical") if macro == "oppose": return verdict, confidence, rationale if technical == "oppose" and macro != "support": return verdict, confidence, rationale note = ("The big-picture view does not object, so the panel stays " "neutral rather than against. ") return "NEUTRAL", min(float(confidence), 49.0), (note + rationale)[:240] # ── 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, str, str], PanelVerdict]" = OrderedDict() _CACHE_MAX = 32 # Reuse only when the full prompt + evidence payload is byte-equivalent. # Direction-only reuse served stale warnings/setups to materially changed # signals; the daily panel budget remains the hard spend bound. Keyed by # (reuse_hash, verdict_source) so a macro-mode verdict is never served to # a chair-mode attach (and vice versa). _last_by_payload: Dict[Tuple[str, str], PanelVerdict] = {} _inflight: Dict[Any, threading.Event] = {} # Per-process 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_payload.clear() for event in _inflight.values(): event.set() _inflight.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, cache_status: str ) -> 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] report.panel_created_at = _dt.datetime.fromtimestamp( v.created_ts, tz=_dt.timezone.utc).isoformat() report.panel_cache_status = cache_status report.panel_payload_hash = v.payload_hash report.panel_verdict_source = v.source def _log_panel_verdict(report: "SignalReport") -> None: """Persist the verdict onto the scan's signals.db row (fail-safe). The row is inserted inside analyze_pair — BEFORE the panel attaches post-hoc — so panel columns arrive via this update. A report that was never logged (backtest replay, logging disabled) is a silent no-op. """ try: from signal_log import log_panel_update log_panel_update(report) except Exception: logger.debug("agent panel: verdict logging failed", exc_info=True) 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 verdict_source = str(settings.get("verdict_source", DEFAULT_VERDICT_SOURCE)) direction = _direction(report) if direction is None: 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 candles = _candle_context(report, settings) jobs = _panel_jobs(report, context, candles) payload_hash = _panel_payload_hash(jobs) reuse_hash = _panel_payload_hash(jobs, ignore_bar_timestamp=True) analysis = report.primary_analysis tf = analysis.timeframe if analysis else "" exact_key = (report.symbol, tf, report.timestamp, direction, payload_hash, verdict_source) reuse_key = (reuse_hash, verdict_source) ttl = float(settings.get("reuse_ttl_seconds", DEFAULT_REUSE_TTL_SECONDS)) inflight_key = reuse_key if ttl > 0 else payload_hash hit = None cache_status = "" leader = False event: Optional[threading.Event] = None with _state_lock: hit = _cache.get(exact_key) if hit is not None: cache_status = "EXACT_HIT" elif ttl > 0: prior = _last_by_payload.get(reuse_key) if prior is not None and time.time() - prior.created_ts < ttl: hit = prior cache_status = "TTL_REUSE" if hit is None: event = _inflight.get(inflight_key) if event is None: event = threading.Event() _inflight[inflight_key] = event leader = True if hit is not None: _write_fields(report, hit, cache_status) _log_panel_verdict(report) return True if not leader: # A successful leader can spend one client timeout on personas # and another on synthesis; followers must cover both phases. client_timeout = getattr( get_llm_client(), "_timeout", DEFAULT_TIMEOUT) timeout = 2.0 * client_timeout + 12.0 if event is None or not event.wait(timeout=timeout): return False with _state_lock: hit = _cache.get(exact_key) cache_status = "EXACT_HIT" if hit is None and ttl > 0: hit = _last_by_payload.get(reuse_key) cache_status = "TTL_REUSE" if hit is None: return False _write_fields(report, hit, cache_status) _log_panel_verdict(report) return True try: cap = int(settings.get("max_panels_per_day", DEFAULT_MAX_PANELS_PER_DAY)) if not _under_panel_budget(cap): return False verdict = run_panel(report, context, candles=candles, jobs=jobs, verdict_source=verdict_source) if verdict is None: return False _write_fields(report, verdict, "MISS") with _state_lock: _cache[exact_key] = verdict while len(_cache) > _CACHE_MAX: _cache.popitem(last=False) _last_by_payload[reuse_key] = verdict _log_panel_verdict(report) return True finally: with _state_lock: current = _inflight.pop(inflight_key, None) if current is not None: current.set() except Exception: logger.exception("agent panel: attach failed — signal unaffected") return False