From 7be2231eb32af9ea0043afe41c86b9f1322fc630 Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Fri, 7 Aug 2026 22:31:50 -0400 Subject: [PATCH] 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 | 530 +++++++++++++++++++++++++++++++++++++++++++ signal_aggregator.py | 9 + test_all.py | 307 +++++++++++++++++++++++++ 3 files changed, 846 insertions(+) create mode 100644 agent_panel.py diff --git a/agent_panel.py b/agent_panel.py new file mode 100644 index 0000000..3fc13cd --- /dev/null +++ b/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": ""} +"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": ""} +""" + + +# ── 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 diff --git a/signal_aggregator.py b/signal_aggregator.py index e21a676..50a00c1 100755 --- a/signal_aggregator.py +++ b/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 diff --git a/test_all.py b/test_all.py index c1b9a44..361c17d 100644 --- a/test_all.py +++ b/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 # ══════════════════════════════════════════════════════════════════════