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.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SisdfkxnXNL9iZsXxN3T15

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
 # ══════════════════════════════════════════════════════════════════════