agent_panel: persona/synthesizer advisory consensus module

ai-hedge-fund-style panel: 4 DeepSeek personas (macro, technical, risk,
contrarian) review a finished SignalReport concurrently on a
panel-private thread pool, then a synthesizer chair issues
AGREE/NEUTRAL/OPPOSE + confidence + rationale. Advisory/display-only by
construction: consumers attach post-hoc via attach_panel; the fields
live on SignalReport with empty defaults and analyze_from_dataframes
never touches them (backtest parity, locked by test).

Spend control on top of the shared 500/day client cap: max_panels_per_day
UTC budget (default 20), exact per-bar result cache, and a reuse TTL
keyed (symbol, tf, direction) so a persisting signal re-serves its
verdict across bars instead of burning 5 requests per bar. <2 valid
takes skips the synthesizer; any failure degrades to no panel.

Claude-Session: https://claude.ai/code/session_01SisdfkxnXNL9iZsXxN3T15

agent_panel.py

@@ -0,0 +1,530 @@
+"""
+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.
+Reply with ONLY JSON:
+{"stance": "support"|"oppose"|"neutral", "confidence": <0-100>,
+ "note": "<max 25 words>"}
+"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": "hot CPI and hawkish repricing lift real yields against 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 — 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": "counter-trend sell against aligned higher timeframes"}
+""" + _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": "thin reward for extreme volatility into event risk"}
+""" + _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": "engine's own bearish evidence outweighs the entry trigger"}
+""" + _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.
+Reply with ONLY JSON:
+{"verdict": "AGREE"|"NEUTRAL"|"OPPOSE", "confidence": <0-100>,
+ "rationale": "<max 40 words>"}
+"""
+
+
+# ── 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

signal_aggregator.py

@@ -88,6 +88,15 @@ class SignalReport:
     news_description: str = ""
     news_summary: str = ""  # LLM market-context blurb (display-only)
 
+    # LLM agent-panel consensus (advisory/DISPLAY-ONLY). Populated
+    # post-hoc by consumers via agent_panel.attach_panel — NEVER inside
+    # analyze_from_dataframes (backtest parity) and never read by any
+    # gate or scoring path.
+    panel_verdict: str = ""        # "AGREE" / "NEUTRAL" / "OPPOSE" / ""
+    panel_confidence: float = 0.0
+    panel_rationale: str = ""
+    panel_takes: List[Dict] = field(default_factory=list)
+
     final_signal: SignalStrength = SignalStrength.NEUTRAL
     final_score: float = 0.0
     final_confidence: float = 0.0

test_all.py

@@ -6979,6 +6979,313 @@ def _():
             pass
 
 
+import agent_panel as ap
+from signal_aggregator import SignalReport, TradeSetup
+
+
+class _PanelStub:
+    """llm_client-singleton drop-in that routes replies by system prompt."""
+    available = True
+
+    def __init__(self, replies):
+        self.replies = replies      # system prompt -> reply dict | None
+        self.calls = []             # (system_prompt, payload) in call order
+
+    def chat_json(self, system_prompt, user_payload):
+        self.calls.append((system_prompt, user_payload))
+        reply = self.replies.get(system_prompt)
+        return reply(user_payload) if callable(reply) else reply
+
+
+_TAKE_OK = {"stance": "support", "confidence": 70, "note": "fine"}
+_SYNTH_OK = {"verdict": "AGREE", "confidence": 75, "rationale": "panel backs it"}
+_ALL_OK = {ap._MACRO_SYSTEM: _TAKE_OK, ap._TECH_SYSTEM: _TAKE_OK,
+           ap._RISK_SYSTEM: _TAKE_OK, ap._CONTRARIAN_SYSTEM: _TAKE_OK,
+           ap._SYNTH_SYSTEM: _SYNTH_OK}
+_PANEL_ON = {"enabled": True, "telegram": True, "dashboard": False,
+             "min_quality": "B", "max_panels_per_day": 20,
+             "reuse_ttl_seconds": 3600}
+
+
+def _panel_report(quality="A", signal=SignalStrength.BUY):
+    """Real report via the pipeline, then forced directional/quality."""
+    config = PlatformConfig()
+    config.primary_timeframe = Timeframe.H1
+    config.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
+    agg = SignalAggregator(config)
+    dfs = {Timeframe.H1: make_ohlcv(300, seed=42),
+           Timeframe.M15: make_ohlcv(300, seed=43),
+           Timeframe.H4: make_ohlcv(300, seed=44)}
+    report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
+    report.final_signal = signal
+    report.signal_quality = quality
+    return report
+
+
+@test("agent_panel: payload builders compact, filtered, JSON-safe")
+def _():
+    report = _panel_report()
+    now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
+    ctx = FundamentalContext(
+        news_headlines=[NewsSentiment(
+            headline=f"h{i}", source="rss", timestamp=now,
+            currencies_mentioned=["USD"], sentiment_score=0.1, keywords=[])
+            for i in range(15)],
+        events_today=[EconomicEvent(
+            time=now, currency="USD", impact="high", event=f"e{i}",
+            forecast="1.0", actual="2.0") for i in range(12)],
+    )
+    macro = ap._macro_payload(report, ctx)
+    assert len(macro["headlines"]) == 10, "headline cap"
+    assert len(macro["events"]) == 8, "event cap"
+    assert macro["events"][0]["released"] is True
+    assert ap._macro_payload(report, None)["headlines"] == []
+
+    tech = ap._technical_payload(report)
+    assert tech["indicators"], "real analysis must yield indicators"
+    hidden = [n for n in tech["indicators"]
+              if n.startswith("_")
+              or report.primary_analysis.indicators[n].weight == 0]
+    assert not hidden, f"private/weight-0 leaked: {hidden}"
+    assert tech["brief"]["direction"] == "BUY"
+    assert tech["brief"]["timeframe"] == report.primary_analysis.timeframe
+
+    assert "setup" not in ap._risk_payload(report), "no setup when None"
+    report.trade_setup = TradeSetup(
+        pair="XAU/USD", display_name="XAU/USD", direction="SELL",
+        entry_price=2000.0, stop_loss=2010.0, take_profit_1=1985.0,
+        take_profit_2=1970.0, risk_reward=1.5, risk_pips=100.0,
+        reward_pips=150.0, pip_value=0.1)
+    risk = ap._risk_payload(report)
+    assert risk["setup"]["risk_reward"] == 1.5
+    assert risk["brief"]["direction"] == "SELL", "setup direction wins"
+
+    contra = ap._contrarian_payload(report)
+    assert len(contra["bullish_reasons"]) <= 3
+
+    for payload in (macro, tech, risk, contra):
+        json.dumps(payload, sort_keys=True, allow_nan=False)
+
+
+@test("agent_panel: persona reply validation clamps and rejects")
+def _():
+    v = ap._validate_take
+    assert v("macro", {"stance": "moon", "confidence": 50, "note": "x"}) is None
+    assert v("macro", {"stance": "support", "confidence": "hot"}) is None
+    assert v("macro", {"stance": "support"}) is None, "missing confidence"
+    assert v("macro", None) is None
+    assert v("macro", "support") is None
+    take = v("risk", {"stance": "oppose", "confidence": 250, "note": "n" * 300})
+    assert take.confidence == 100.0 and len(take.note) == 160
+    assert v("risk", {"stance": "neutral", "confidence": -3, "note": ""}).confidence == 0.0
+
+
+@test("agent_panel: <2 valid takes → None without spending synthesizer")
+def _():
+    try:
+        stub = _PanelStub({ap._MACRO_SYSTEM: _TAKE_OK})   # others → None
+        lc._client = stub
+        assert ap.run_panel(_panel_report()) is None
+        assert ap._SYNTH_SYSTEM not in [s for s, _ in stub.calls], \
+            "synthesizer request must not be spent on a failed panel"
+        assert len(stub.calls) == 4
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: synthesizer failure or garbage verdict → None")
+def _():
+    try:
+        for synth in (None, {"verdict": "MAYBE", "confidence": 50},
+                      {"verdict": "AGREE", "confidence": "hot"}):
+            lc._client = _PanelStub({**_ALL_OK, ap._SYNTH_SYSTEM: synth})
+            assert ap.run_panel(_panel_report()) is None, f"synth={synth}"
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: full run yields verdict + 4 ordered takes")
+def _():
+    try:
+        stub = _PanelStub(dict(_ALL_OK))
+        stub.replies[ap._SYNTH_SYSTEM] = {
+            "verdict": "OPPOSE", "confidence": 72.4,
+            "rationale": "r" * 300}
+        lc._client = stub
+        v = ap.run_panel(_panel_report())
+        assert v.verdict == "OPPOSE" and v.confidence == 72.4
+        assert len(v.rationale) == 240, "rationale truncated"
+        assert [t.persona for t in v.takes] == \
+            ["macro", "technical", "risk", "contrarian"]
+        synth_payload = dict(stub.calls)[ap._SYNTH_SYSTEM]
+        assert synth_payload["direction"] == "BUY"
+        assert len(synth_payload["takes"]) == 4
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: NEUTRAL signal → no panel even forced")
+def _():
+    try:
+        stub = _PanelStub(dict(_ALL_OK))
+        lc._client = stub
+        report = _panel_report(signal=SignalStrength.NEUTRAL)
+        assert ap.run_panel(report) is None
+        assert ap.attach_panel(report, _PANEL_ON, force=True,
+                               context=FundamentalContext()) is False
+        assert stub.calls == [], "NEUTRAL must spend nothing"
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: exact + reuse-TTL cache re-attach with zero spend")
+def _():
+    try:
+        stub = _PanelStub(dict(_ALL_OK))
+        lc._client = stub
+        report = _panel_report()
+        assert ap.attach_panel(report, _PANEL_ON,
+                               context=FundamentalContext()) is True
+        assert report.panel_verdict == "AGREE"
+        assert len(stub.calls) == 5
+
+        # same bar again → exact-key hit
+        report2 = _panel_report()
+        assert ap.attach_panel(report2, _PANEL_ON,
+                               context=FundamentalContext()) is True
+        assert report2.panel_verdict == "AGREE" and len(stub.calls) == 5
+
+        # newer bar, same symbol/tf/direction → reuse-TTL hit
+        report3 = _panel_report()
+        report3.timestamp = "2099-01-01 01:00:00+00:00"
+        assert ap.attach_panel(report3, _PANEL_ON,
+                               context=FundamentalContext()) is True
+        assert len(stub.calls) == 5, "persisting signal must not re-spend"
+
+        # TTL 0 → the reuse layer is off; a new bar re-runs the panel
+        report4 = _panel_report()
+        report4.timestamp = "2099-01-01 02:00:00+00:00"
+        assert ap.attach_panel(report4, {**_PANEL_ON, "reuse_ttl_seconds": 0},
+                               context=FundamentalContext()) is True
+        assert len(stub.calls) == 10
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: daily budget exhausts and re-arms on UTC roll")
+def _():
+    try:
+        stub = _PanelStub(dict(_ALL_OK))
+        lc._client = stub
+        settings = {**_PANEL_ON, "max_panels_per_day": 1,
+                    "reuse_ttl_seconds": 0}
+        assert ap.attach_panel(_panel_report(), settings,
+                               context=FundamentalContext()) is True
+        assert len(stub.calls) == 5
+        report = _panel_report()
+        report.timestamp = "2099-01-01 01:00:00+00:00"
+        assert ap.attach_panel(report, settings,
+                               context=FundamentalContext()) is False
+        assert len(stub.calls) == 5, "budget-blocked attempt must spend nothing"
+        ap._panel_day = (_dt_m19.datetime.now(_dt_m19.timezone.utc).date()
+                         - _dt_m19.timedelta(days=1))
+        assert ap.attach_panel(report, settings,
+                               context=FundamentalContext()) is True
+        assert len(stub.calls) == 10
+        assert ap.panel_health()["used_today"] == 1
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: enabled/surface/min_quality gates; force bypasses")
+def _():
+    try:
+        stub = _PanelStub(dict(_ALL_OK))
+        lc._client = stub
+        report = _panel_report()
+        ctx = FundamentalContext()
+        assert ap.attach_panel(report, {}, context=ctx) is False, \
+            "absent config block must keep the feature dark"
+        assert ap.attach_panel(report, {**_PANEL_ON, "enabled": False},
+                               context=ctx) is False
+        assert ap.attach_panel(report, {**_PANEL_ON, "telegram": False},
+                               context=ctx) is False
+        assert ap.attach_panel(report, _PANEL_ON, surface="dashboard",
+                               context=ctx) is False, "dashboard defaults off"
+        assert ap.attach_panel(_panel_report(quality="C"), _PANEL_ON,
+                               context=ctx) is False, "C < min_quality B"
+        assert ap.attach_panel(_panel_report(quality="F"), _PANEL_ON,
+                               context=ctx) is False
+        assert stub.calls == [], "every gate must block before any spend"
+        assert ap.attach_panel(report, {**_PANEL_ON, "dashboard": True},
+                               surface="dashboard", context=ctx) is True
+        ap.reset_panel_state()
+        assert ap.attach_panel(_panel_report(quality="C"),
+                               {**_PANEL_ON, "enabled": False},
+                               force=True, context=ctx) is True, \
+            "force bypasses enabled/surface/quality"
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("backtest parity: analyze_from_dataframes never populates panel")
+def _():
+    try:
+        stub = _PanelStub(dict(_ALL_OK))
+        lc._client = stub
+        config = PlatformConfig()
+        config.primary_timeframe = Timeframe.H1
+        config.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
+        agg = SignalAggregator(config)
+        dfs = {Timeframe.H1: make_ohlcv(300, seed=42),
+               Timeframe.M15: make_ohlcv(300, seed=43),
+               Timeframe.H4: make_ohlcv(300, seed=44)}
+        for use_news in (True, False):
+            report = agg.analyze_from_dataframes(
+                "XAU/USD", dfs, context=FundamentalContext(),
+                use_news=use_news)
+            assert report.panel_verdict == ""
+            assert report.panel_confidence == 0.0
+            assert report.panel_rationale == ""
+            assert report.panel_takes == []
+        assert stub.calls == [], \
+            "the replayable core must never invoke the panel"
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: attach never raises (LLM explosion, garbage settings)")
+def _():
+    class _Bomb:
+        available = True
+
+        def chat_json(self, system_prompt, user_payload):
+            raise RuntimeError("boom")
+    try:
+        lc._client = _Bomb()
+        report = _panel_report()
+        assert ap.attach_panel(report, _PANEL_ON,
+                               context=FundamentalContext()) is False
+        assert report.panel_verdict == "" and report.panel_takes == []
+        logging.disable(logging.CRITICAL)   # silence the expected traceback
+        try:
+            assert ap.attach_panel(report, 42,
+                                   context=FundamentalContext()) is False
+        finally:
+            logging.disable(logging.NOTSET)
+    finally:
+        lc._client = None
+        ap.reset_panel_state()
+
+
 # ══════════════════════════════════════════════════════════════════════
 #  SUMMARY
 # ══════════════════════════════════════════════════════════════════════