""" 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. All four personas always run and are logged. Since 2026-09-17 (``PANEL_PROMPT_VERSION`` v2, ``verdict_source = "mean"``, the default) every persona returns a PROBABILITY that the trade closes with a profit under its trailing exit (``p_win``, 0-100), anchored to the validated base rate for that setup class, and the verdict is the equal-weight mean of those probabilities (quorum 3 of 4; AGREE at 60 or above, OPPOSE at 40 or below, NEUTRAL between). No LLM aggregates. The two earlier sources remain for comparison: ``"macro"`` (2026-09-03 owner decision: the macro persona's stance verbatim, 4 calls) and ``"chair"`` (a fifth synthesizer call plus the OPPOSE guard). Why: the 2026-09-17 live review (backtest/research/signal_log_review_2026-09-17.md, §7.1) found the macro-only verdict opposed target-reaching trades as often as it agreed (its only input was headlines, a lagging echo of price) and three personas voted the same way on nearly every setup; the chair had already proved anti-informative on 2026-09-03. Persona weights stay equal until about 300 graded outcomes exist (owner rule). Persona set (2026-09-17, owner decision): only the MACRO persona runs by default (``agent_panel.personas = ["macro"]``, one LLM call per panel). Evidence: on the 27 v1 outcomes the technical persona said "support" on 79% of setups (it re-reads the indicators the signal was built from), risk said "oppose" on 79%, contrarian 65% — none carried information — while macro was the only voice with any ordering (rank correlation with R +0.36; the 2026-09-03 88%/25% win-rate split). Under the v2 prompts all four sat within two points of the base rate. Removing the three saves three quarters of the tokens, shrinks every logged payload and removes contradictory notes from the alert. The full panel stays available (``personas`` may list any of macro / technical / risk / contrarian) for research replays; the code-computed context block now carries the geometry and event-timing facts the risk persona used to judge. Every payload now carries a code-computed ``context`` block (price path already travelled in ATR, distance from the 24 h extreme, candle size, stop/target in ATR, the setup class's base rate from the two-year book, event timing against the expected hold) so the model judges facts instead of narrative; ``PANEL_PROMPT_VERSION`` is logged with every verdict and is part of the payload hash so prompt eras never mix. 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 4 requests (the personas; 5 with verdict_source="chair", which adds the 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 pathlib import Path 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 = "mean" VERDICT_SOURCES = ("mean", "macro", "chair") #: Logged with every verdict and folded into the payload hash. Bump on #: ANY prompt or context-schema change so the scorecard can split eras #: (analyze_signal_log splits by panel_prompt_version). PANEL_PROMPT_VERSION = "v2-2026-09-17" #: Symmetric probability bands (percent). A persona's stance and the #: mean verdict are DERIVED from p_win with the same thresholds, so #: "support" and "oppose" are equally hard to earn. P_SUPPORT = 60.0 P_OPPOSE = 40.0 #: Personas that must have answered before a mean verdict is issued #: (capped at the number of personas configured). MEAN_QUORUM = 3 #: Every persona the module knows, in display order. ALL_PERSONAS = ("macro", "technical", "risk", "contrarian") #: Personas that run live unless ``agent_panel.personas`` says otherwise. DEFAULT_PERSONAS = ("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. # # v2 (2026-09-17). Design rules, each traceable to the live review: # - every persona returns p_win, a probability anchored to the setup # class's validated base rate, plus one reason for and one against # (calibration is measurable; constant voting is visible); # - no "news X always moves gold Y" doctrine — the review showed the # macro persona was a lagging echo of price; it now has to judge # whether the news is NEW relative to the move; # - indicator agreement is declared uninformative (the signal was built # from those indicators), so the technical persona cannot rubber-stamp; # - volatility regime and stretch tags are declared non-reasons (book # evidence), so the risk and contrarian personas cannot oppose by # default. _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. Every payload carries a "context" block computed by code: - base_rate: how trades of this exact kind (timeframe, setup type, grade, direction) fared in the validated two-year test — win_rate (percent that closed with a profit), mean_r, tp1_reach_rate, median_bars_held, and n (how many trades the number rests on; below 30 is a weak anchor). This is your STARTING POINT for p_win. - prior_move_4h_atr / prior_move_24h_atr: how far price already moved IN the trade's direction over the last 4 / 24 hours, in ATR units (one ATR = one typical bar's range). move_from_24h_start_atr is the distance from where the current move began (move_started_at); off_24h_extreme_atr is how far price sits short of the 24-hour high (for a buy) or low (for a sell); hours_since_24h_extreme says how fresh that extreme is. - max_bar_range_last3_atr / signal_bar_body_atr: candle size around the signal in ATR units (2 or more is a big candle). Note on both of the above: in the validated two-year test, entries made after a long prior move or right after a big candle were NOT worse on average (the system is a trend follower; it enters after a move has started). Report these facts, but do not lower p_win for them alone. - stop_atr / tp1_r / tp1_atr / cost_to_risk_pct: the stop and first target in ATR and in R (R = the amount risked), and cost as a percent of the risk. - next_high_impact_event_minutes / expected_hold_minutes / event_inside_expected_hold: whether a scheduled major release falls inside the time this kind of trade usually takes. - penalty_tags / exhaustion_count: the engine's stretch flags. In the validated test these flags did NOT identify losing trades; treat them as description, never as a reason by themselves. Estimate p_win: the probability (0-100) that THIS trade closes with a profit under its trailing exit. Start from base_rate.win_rate. Move away from it only for a concrete fact in the JSON, by an amount you can defend; a move of more than 20 points must name that fact in the note. When the evidence your role needs is thin or missing, stay close to the base rate. Give exactly one reason_for (the strongest fact in favour) and one reason_against (the strongest fact against), each max 12 words. Write everything 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: {"p_win": <0-100>, "reason_for": "", "reason_against": "", "note": ""} "note" is your single most important point about this trade. """ _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), the code-computed "context" block, recent directional "headlines" (each with a code-computed "gold_influence", score and published_at time), today's high-impact "events" (released: true means the number is out), optional aggregate "sentiment", and "candles" — the recent price path per timeframe in ATR units relative to the latest close (0.0 = the latest close, +1.0 = one ATR above it), oldest first. Neutral and unrelated headlines were removed by code. Trust each headline's gold_influence direction; do not reinterpret it from the wording. Judge ONE thing: is the news flow NEW information, or is it already in the price? Compare each headline's published_at with context.move_started_at and read the candles. A headline older than the start of the current move, or one whose direction matches a move that has already run several ATR, is already reflected in the price and is NOT a reason to expect more of the same. Only a released surprise, or a headline newer than the move start, counts as new information. An unreleased high-impact event inside expected_hold_minutes is a reason to stay close to the base rate, not a direction. Do not apply any fixed rule of the form "this kind of news always moves gold this way"; the payload's gold_influence already did that scoring. Example: brief direction SELL, price already fell 5 ATR in 4 hours, every bearish headline published before move_started_at, base_rate.win_rate 58 -> {"p_win": 55, "reason_for": "Recent news still leans against gold.", "reason_against": "That news is older than the drop, so it is priced in.", "note": "The bad news is already in the price; expect the usual odds."} Example: brief direction BUY, a high-impact release came out 20 minutes ago with a surprise the code marked bullish for gold, base_rate.win_rate 60 -> {"p_win": 68, "reason_for": "A fresh surprise in the data favours gold.", "reason_against": "Big moves right after a release can reverse.", "note": "New data just came out and it helps the buy."} Example: brief direction BUY, a high-impact release came out 15 minutes ago with a surprise the code marked bearish for gold, base_rate.win_rate 60 -> {"p_win": 47, "reason_for": "Before the news, the price was climbing.", "reason_against": "Fresh data just came out against gold.", "note": "New data works against this buy; the odds drop below usual."} """ + _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"), the code-computed "context" block, 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, 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. Important: the signal was BUILT from these indicators, so most of them agreeing with it is expected and carries almost no information. Do not raise p_win because the indicators agree. What does carry information: a higher timeframe pointing the other way ("mtf" disagreeing with the trade), price pressed against a key level that stands in the trade's way, and a candle path that contradicts the indicator readings. For XAU_CP, the named completed pattern is the trigger and may lead lagging indicators; indicator lag alone is not a reason against it. A move that has already run far, or a big candle just before the signal, is NOT a reason by itself: in the validated test such entries were not worse on average, because this system enters after moves have started. Describe them if you must; do not lower p_win for them alone. Judge ONE thing: does the chart make this entry better or worse than the average trade of its kind? Move p_win down when a higher timeframe points against the trade or a key level blocks the path; move it up when the higher timeframes agree and nothing blocks the path. Example: brief direction SELL, the 1-hour and 4-hour charts both point up and a support level sits just below the price, base_rate.win_rate 55 -> {"p_win": 42, "reason_for": "The short-term slide is still going.", "reason_against": "Bigger charts point up and support sits just below.", "note": "This sells while the bigger picture still points up."} Example: brief direction BUY, higher timeframes agree, no level in the way, price moved 1 ATR since the move began, base_rate.win_rate 60 -> {"p_win": 65, "reason_for": "Every chart points up and nothing blocks the path.", "reason_against": "Nothing specific stands out against it.", "note": "A clean move with the bigger charts on its side."} """ + _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"), the code-computed "context" block, 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 particular setup's risk different from the average trade of its kind? Compare, do not just describe: - stop size: context.stop_atr against the usual 2 to 4 ATR for this strategy; a stop far wider than usual puts the target far away too. - cost: context.cost_to_risk_pct above 10 is expensive. - event timing: context.event_inside_expected_hold true means a major release lands while the trade is usually still open. Things that are NOT reasons by themselves: a high or extreme volatility regime, exhaustion or stretch tags, and generic warnings — the validated test shows trades with those labels were not worse on average. A missing setup means the signal graded too low for one; then stay near the base rate and lower p_win only if the rest of the evidence is weak. Example: stop 6 ATR, cost 3 percent of risk, a high-impact release in 40 minutes with expected_hold_minutes 180, base_rate.win_rate 58 -> {"p_win": 46, "reason_for": "Costs are small next to the risk.", "reason_against": "Major news lands while the trade is still open.", "note": "Scheduled news is due before this trade would normally finish."} Example: stop 2 ATR, cost 3 percent, no event inside the hold, volatility extreme, base_rate.win_rate 58 -> {"p_win": 58, "reason_for": "Stop and costs are normal for this kind of trade.", "reason_against": "Prices are swinging hard, though that alone is not a problem.", "note": "Nothing about this setup's risk stands out from the usual."} """ + _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 code-computed "context" block, 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, its warnings, and "candles" (the price path in ATR units relative to the latest close, oldest first). Neutral and unrelated news was removed by code; trust each headline's code-computed gold_influence instead of reinterpreting it. Your job: build the strongest honest case AGAINST the trade, then weigh it against the base rate. The engine's own list of negatives is not evidence by itself: those warnings were already priced into the grade, and the validated test shows flagged trades were not worse on average. A counter-case is strong only when it rests on something the engine did not already discount: a scheduled high-impact release inside the expected hold, news newer than the move that points the other way, a higher timeframe pointing the other way, or a stop far wider than usual for this kind of trade. A move that has already run far, or a big candle, is NOT such a thing: the validated test shows entries after them were not worse on average, so they never carry the counter-case alone. If the skeptic finds nothing that bites, say so and stay at the base rate or above it. Example: brief direction BUY, a headline published after the move began is marked bearish for gold, and the 4-hour chart points down, base_rate.win_rate 55 -> {"p_win": 44, "reason_for": "The shorter charts still point up.", "reason_against": "Fresh news works against gold and the bigger chart points down.", "note": "New facts, not old warnings, argue against this buy."} Example: brief direction SELL, no news inside the hold, higher charts agree, stop of the usual size, base_rate.win_rate 55 -> {"p_win": 57, "reason_for": "Every chart agrees and nothing new argues against it.", "reason_against": "The engine's warnings, which the test says are not a problem.", "note": "Even the skeptic finds little that bites here."} """ + _TAKE_CONTRACT # The chair is kept ONLY for verdict_source="chair" comparisons. It was # shown anti-informative on 2026-09-03 and is not the default. _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", "p_win", "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" (derived from p_win) confidence: float # 0-100 (v2: distance of p_win from 50, doubled) note: str # v2: probability (0-100) the trade closes with a profit under its # trailing exit, and the single strongest fact each way. A v1-format # reply (stance + confidence) gets a pseudo-probability so both eras # score on the same scale (support -> 50 + conf/2, oppose -> 50 - # conf/2, neutral -> 50). p_win: Optional[float] = None reason_for: str = "" reason_against: str = "" @dataclass class PanelVerdict: verdict: str # "AGREE" | "NEUTRAL" | "OPPOSE" confidence: float # 0-100 (for source="mean": the probability itself) rationale: str takes: List[PersonaTake] = field(default_factory=list) created_ts: float = 0.0 payload_hash: str = "" source: str = "chair" # "mean" | "macro" | "chair" — which path produced this probability: Optional[float] = None # mean p_win (source="mean"), else None prompt_version: str = PANEL_PROMPT_VERSION # Exactly what each persona saw (persona -> payload) and the shared # code-computed context, kept so scripts/panel_replay.py can re-run a # new prompt version on the SAME evidence and score it against the # realised outcome. Logged as panel_payloads_json / panel_context_json. payloads: Dict[str, Any] = field(default_factory=dict) context: Dict[str, Any] = field(default_factory=dict) # ── 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"], candles: Optional[Dict[str, Any]] = None, ctx: Optional[Dict[str, Any]] = None) -> 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 if candles: payload["candles"] = candles if ctx: payload["context"] = ctx return payload def _technical_payload(report: "SignalReport", candles: Optional[Dict[str, Any]] = None, ctx: 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 if ctx: payload["context"] = ctx return payload def _risk_payload(report: "SignalReport", ctx: Optional[Dict[str, Any]] = None) -> 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 if ctx: payload["context"] = ctx return payload def _contrarian_payload( report: "SignalReport", context: Optional["FundamentalContext"] = None, candles: Optional[Dict[str, Any]] = None, ctx: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: payload: Dict[str, Any] = { "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), } if candles: payload["candles"] = candles if ctx: payload["context"] = ctx return payload # ── 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 # ── Decision context (code-computed facts every persona receives) ─── # # The 2026-09-17 review found the verdict persona judged from headlines # alone (a lagging echo of price) while the facts that mattered — how # far the move had already run, whether a big candle just printed, the # stop in ATR, the class base rate, and event timing against the hold — # were never in any payload. Code computes them here; the model reads # numbers, not narrative. Bounded lookbacks (<= 24 h of primary bars, # capped at 500) keep the block identical on a full history and on the # live fetch tail (same rule as position_sizing). NEVER raises. _TF_SECONDS = {"5min": 300, "15min": 900, "1h": 3600, "4h": 14400, "1d": 86400} _CONTEXT_MAX_BARS = 500 _BIG_BAR_ATR = 2.0 _BASE_RATES_FILE = Path(__file__).resolve().with_name("agent_panel_base_rates.json") _BASE_RATE_MIN_N = 30 _base_rates_cache: Optional[Dict[str, Any]] = None _base_rates_lock = threading.Lock() def _load_base_rates() -> Dict[str, Any]: """agent_panel_base_rates.json (built by scripts/build_panel_base_rates.py from the MD5-verified reference books), loaded once; {} when absent.""" global _base_rates_cache with _base_rates_lock: if _base_rates_cache is None: try: with open(_BASE_RATES_FILE, "r", encoding="utf-8") as fh: data = json.load(fh) _base_rates_cache = data if isinstance(data, dict) else {} except Exception as e: logger.debug("agent panel: base rates unavailable: %s", e) _base_rates_cache = {} return _base_rates_cache def reset_base_rates_for_tests(path: Optional[Any] = None) -> None: global _base_rates_cache, _BASE_RATES_FILE with _base_rates_lock: _base_rates_cache = None if path is not None: _BASE_RATES_FILE = Path(path) def _base_rate(timeframe: str, strategy: Optional[str], grade: str, direction: Optional[str]) -> Optional[Dict[str, Any]]: """Validated-book outcome statistics for a setup class, falling back from (tf, strategy, grade, direction) to (tf, direction) to (tf) when the exact class has fewer than _BASE_RATE_MIN_N trades.""" table = _load_base_rates().get("classes") or {} if not table: return None candidates = ( ("class", f"{timeframe}|{strategy or 'NO_SETUP'}|{grade}|{direction}", _BASE_RATE_MIN_N), ("timeframe_direction", f"{timeframe}|*|*|{direction}", 1), ("timeframe", f"{timeframe}|*|*|*", 1), ) for level, key, min_n in candidates: row = table.get(key) if not isinstance(row, dict): continue n = int(_finite(row.get("n")) or 0) if n < min_n: continue out: Dict[str, Any] = {"level": level, "n": n} for k in ("win_rate", "mean_r", "tp1_reach_rate", "median_bars_held"): v = _finite(row.get(k)) if v is not None: out[k] = round(v, 2) if k != "median_bars_held" else int(v) book = row.get("book") if book: out["book"] = str(book) return out return None def _primary_frame(report: "SignalReport", frames: Any) -> Any: """The primary-timeframe bars: the live wrapper's hand-off when present, else the panel provider's cached fetch. None on failure.""" analysis = report.primary_analysis tf = _TF_BY_VALUE.get(analysis.timeframe) if analysis else None if tf is None: return None df = None if isinstance(frames, dict): df = frames.get(tf) if df is None: try: df = _panel_provider().fetch_ohlcv(report.symbol, tf) except Exception as e: logger.debug("agent panel: primary frame unavailable: %s", e) return None return df if df is not None and len(df) >= 3 else None def _price_context(report: "SignalReport", frames: Any = None, context: Optional["FundamentalContext"] = None ) -> Dict[str, Any]: """The shared "context" block (see module docstring). {} on failure.""" out: Dict[str, Any] = {} try: direction = _direction(report) if direction is None: return out d = 1.0 if direction == "BUY" else -1.0 analysis = report.primary_analysis tf_value = analysis.timeframe if analysis else "" atr = _finite((analysis.key_levels or {}).get("atr")) if analysis else None setup = report.trade_setup strategy = getattr(setup, "strategy", None) if setup is not None else None # Geometry in ATR and R. if setup is not None and atr and atr > 0: entry = _finite(getattr(setup, "entry_price", None)) stop = _finite(getattr(setup, "stop_loss", None)) tp1 = _finite(getattr(setup, "take_profit_1", None)) cost = _finite(getattr(setup, "round_trip_cost", None)) if entry is not None and stop is not None and abs(entry - stop) > 0: risk = abs(entry - stop) out["stop_atr"] = round(risk / atr, 2) if tp1 is not None: out["tp1_atr"] = round(abs(tp1 - entry) / atr, 2) out["tp1_r"] = round(abs(tp1 - entry) / risk, 2) if cost is not None: out["cost_to_risk_pct"] = round(100.0 * cost / risk, 2) # Base rate from the validated books. rate = _base_rate(tf_value, strategy, report.signal_quality, direction) if rate is not None: out["base_rate"] = rate # Price path already travelled. secs = _TF_SECONDS.get(tf_value) df = _primary_frame(report, frames) if df is not None and secs and atr and atr > 0: bars_4h = max(1, 4 * 3600 // secs) bars_24h = max(1, 24 * 3600 // secs) tail = df.iloc[-min(len(df), min(_CONTEXT_MAX_BARS, bars_24h + 1)):] close = tail["Close"].astype(float).to_numpy() high = tail["High"].astype(float).to_numpy() low = tail["Low"].astype(float).to_numpy() opn = tail["Open"].astype(float).to_numpy() idx = tail.index px = float(close[-1]) n = len(close) if n > bars_4h: out["prior_move_4h_atr"] = round(d * (px - close[-1 - bars_4h]) / atr, 2) lag24 = min(bars_24h, n - 1) out["prior_move_24h_atr"] = round(d * (px - close[-1 - lag24]) / atr, 2) if lag24 < bars_24h: out["prior_move_24h_bars_available"] = int(lag24) window = slice(max(0, n - bars_24h), n) hi_i = int(high[window].argmax()) + window.start lo_i = int(low[window].argmin()) + window.start if d > 0: ext_i, start_i = hi_i, lo_i out["off_24h_extreme_atr"] = round((high[ext_i] - px) / atr, 2) out["move_from_24h_start_atr"] = round((px - low[start_i]) / atr, 2) else: ext_i, start_i = lo_i, hi_i out["off_24h_extreme_atr"] = round((px - low[ext_i]) / atr, 2) out["move_from_24h_start_atr"] = round((high[start_i] - px) / atr, 2) try: out["hours_since_24h_extreme"] = round( (idx[-1] - idx[ext_i]).total_seconds() / 3600.0, 1) started = idx[start_i] out["move_started_at"] = (started.isoformat() if hasattr(started, "isoformat") else str(started)) except Exception: pass last3 = slice(max(0, n - 3), n) out["max_bar_range_last3_atr"] = round( float((high[last3] - low[last3]).max()) / atr, 2) out["signal_bar_body_atr"] = round(d * (px - float(opn[-1])) / atr, 2) out["big_candle_last3"] = bool(out["max_bar_range_last3_atr"] >= _BIG_BAR_ATR) out["bar_minutes"] = int(secs // 60) # Event timing against the expected hold. hold_bars = (out.get("base_rate") or {}).get("median_bars_held") if hold_bars and secs: out["expected_hold_minutes"] = int(hold_bars * secs // 60) if context is not None: minutes = [] for ev in list(getattr(context, "events_today", None) or []) + \ list(getattr(context, "events_upcoming_1h", None) or []): try: if ev.is_high_impact and ev.is_upcoming: minutes.append(float(ev.minutes_until)) except Exception: continue if minutes: nxt = round(min(minutes), 1) out["next_high_impact_event_minutes"] = nxt if "expected_hold_minutes" in out: out["event_inside_expected_hold"] = bool( nxt <= out["expected_hold_minutes"]) # Engine stretch flags, with the book's verdict on them. tags = sorted(str(k) for k in (report.penalty_tags or {}).keys()) out["penalty_tags"] = tags out["exhaustion_count"] = int(report.exhaustion_count or 0) out["penalty_note"] = ( "In the two-year test these tags did not identify losing trades " "(tagged M5 trades did better than average, M15 about the same).") except Exception as e: logger.debug("agent panel: price context failed: %s", e, exc_info=True) # JSON-safe: drop non-finite floats. return {k: v for k, v in out.items() if not (isinstance(v, float) and not math.isfinite(v))} # ── 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 derive_stance(p_win: float) -> str: """Symmetric bands: support at P_SUPPORT or above, oppose at P_OPPOSE or below, neutral between.""" if p_win >= P_SUPPORT: return "support" if p_win <= P_OPPOSE: return "oppose" return "neutral" def pseudo_probability(stance: str, confidence: float) -> float: """v1 stance + confidence -> a probability on the v2 scale, so both prompt eras score with the same Brier arithmetic.""" if stance == "support": return 50.0 + confidence / 2.0 if stance == "oppose": return 50.0 - confidence / 2.0 return 50.0 def _validate_take(persona: str, reply: Any) -> Optional[PersonaTake]: """Accept a v2 reply (p_win + reasons) or a v1 reply (stance + confidence); anything else means the persona does not vote.""" if not isinstance(reply, dict): return None p_win = _clamp_confidence(reply.get("p_win")) if p_win is not None: stance = derive_stance(p_win) confidence = round(abs(p_win - 50.0) * 2.0, 1) else: stance = reply.get("stance") if stance not in _VALID_STANCES: return None confidence = _clamp_confidence(reply.get("confidence")) if confidence is None: return None p_win = pseudo_probability(stance, confidence) note = str(reply.get("note", "")).strip()[:160] return PersonaTake( persona=persona, stance=stance, confidence=confidence, note=note, p_win=round(float(p_win), 1), reason_for=str(reply.get("reason_for", "")).strip()[:160], reason_against=str(reply.get("reason_against", "")).strip()[:160]) _PERSONA_ORDER = {"macro": 0, "technical": 1, "risk": 2, "contrarian": 3} def mean_verdict(takes: List[PersonaTake], payload_hash: str = "", quorum: int = MEAN_QUORUM) -> Optional[PanelVerdict]: """Equal-weight mean of the personas' p_win (the default verdict since v2). None below the quorum — nothing is invented. The rationale is the note of the persona furthest from 50 (ties: macro first).""" valid = [t for t in takes if t.p_win is not None] if len(valid) < quorum: return None p = sum(t.p_win for t in valid) / len(valid) verdict = ("AGREE" if p >= P_SUPPORT else "OPPOSE" if p <= P_OPPOSE else "NEUTRAL") lead = max(valid, key=lambda t: (abs(t.p_win - 50.0), -_PERSONA_ORDER.get(t.persona, 9))) return PanelVerdict( verdict=verdict, confidence=round(p, 1), rationale=lead.note[:240], takes=list(takes), created_ts=time.time(), payload_hash=payload_hash, source="mean", probability=round(p, 1)) 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 normalize_personas(personas: Any) -> Tuple[str, ...]: """Validated, de-duplicated persona tuple in display order; unknown names are dropped; an empty result falls back to DEFAULT_PERSONAS.""" if personas is None: return DEFAULT_PERSONAS if isinstance(personas, str): personas = [personas] try: wanted = {str(p).strip().lower() for p in personas} except TypeError: return DEFAULT_PERSONAS out = tuple(p for p in ALL_PERSONAS if p in wanted) return out or DEFAULT_PERSONAS def _panel_jobs( report: "SignalReport", context: Optional["FundamentalContext"], candles: Optional[Dict[str, Any]], ctx: Optional[Dict[str, Any]] = None, personas: Tuple[str, ...] = ALL_PERSONAS, ) -> Tuple[Tuple[str, str, Dict[str, Any]], ...]: builders = { "macro": lambda: ("macro", _MACRO_SYSTEM, _macro_payload(report, context, candles, ctx)), "technical": lambda: ("technical", _TECH_SYSTEM, _technical_payload(report, candles, ctx)), "risk": lambda: ("risk", _RISK_SYSTEM, _risk_payload(report, ctx)), "contrarian": lambda: ("contrarian", _CONTRARIAN_SYSTEM, _contrarian_payload(report, context, candles, ctx)), } return tuple(builders[p]() for p in ALL_PERSONAS if p in personas) # 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-v3", "prompt_version": PANEL_PROMPT_VERSION, "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, ctx: Optional[Dict[str, Any]] = None, personas: Tuple[str, ...] = ALL_PERSONAS, ) -> Optional[PanelVerdict]: """Run the configured personas (direct callers default to all four; ``attach_panel`` passes ``agent_panel.personas``, default macro only), then aggregate by ``verdict_source``: ``"mean"`` (default) = equal- weight mean of the personas' p_win, quorum min(3, personas), no LLM aggregator; ``"macro"`` = the macro persona's stance verbatim (the synthesizer is not called); ``"chair"`` = all takes through the synthesizer plus the OPPOSE guard (needs at least two personas). 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 personas = normalize_personas(personas) if jobs is None: if ctx is None: ctx = _price_context(report, getattr(report, "_frames", None), context) jobs = _panel_jobs(report, context, candles, ctx, personas) payload_hash = _panel_payload_hash(jobs) payloads = {name: payload for name, _system, payload in 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 takes.sort(key=lambda t: _PERSONA_ORDER.get(t.persona, 9)) def _finish(v: Optional[PanelVerdict]) -> Optional[PanelVerdict]: if v is not None: v.payloads = payloads v.context = dict(ctx or {}) v.prompt_version = PANEL_PROMPT_VERSION return v if verdict_source == "mean": quorum = min(MEAN_QUORUM, len(jobs)) v = mean_verdict(takes, payload_hash, quorum=quorum) if v is None: logger.debug("agent panel: %d/%d persona takes — below the mean " "quorum of %d, no verdict", len(takes), len(jobs), quorum) return _finish(v) if verdict_source == "macro": macro = next((t for t in takes if t.persona == "macro"), None) if macro is not None: return _finish(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, "p_win": t.p_win, "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 _finish(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, "p_win": t.p_win, "reason_for": t.reason_for, "reason_against": t.reason_against} 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 report.panel_probability = (float(v.probability) if v.probability is not None else 0.0) report.panel_prompt_version = v.prompt_version report.panel_payloads = dict(v.payloads or {}) report.panel_context = dict(v.context or {}) 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 personas = normalize_personas(settings.get("personas", DEFAULT_PERSONAS)) candles = _candle_context(report, settings) ctx = _price_context(report, getattr(report, "_frames", None), context) jobs = _panel_jobs(report, context, candles, ctx, personas) 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 "" # payload_hash already reflects the persona set (fewer jobs, fewer # envelope entries), so a macro-only verdict is never served to a # full-panel attach or vice versa. 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, ctx=ctx, personas=personas) 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