ForexBot2/agent_panel.py

"""
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.1-2026-09-26"   # v2.1: beginner wording rules in _TAKE_CONTRACT

#: 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. Never use these words:
"priced in", "bullish", "bearish", "hawkish", "dovish", "macro",
"sentiment", "momentum", "volatility", "resistance", "support level",
"breakout", "pullback", "basis points", "yields". Say instead: "the
price already moved on that news", "news that is good for gold",
"news that is bad for gold", "higher interest rates", "the price
fell fast", "the price bounced". Each sentence must make sense to a
reader who does not know what gold or interest rates usually do.
Reply with ONLY JSON:
{"p_win": <0-100>, "reason_for": "<max 12 words>",
 "reason_against": "<max 12 words>", "note": "<max 15 words>"}
"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 came before the drop; the price already moved on it.",
"note": "The price already moved on the bad news; 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": "<max 30 words>"}
"""


# ── 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 {"<tf>": _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:
            # The macro persona's own p_win is the verdict's probability
            # (2026-09-26 signals.db review: every v2 row logged under
            # verdict_source="macro" had panel_probability NULL — the
            # verdict carried none, _write_fields wrote 0.0 and the logger
            # dropped the falsy value — so the 100/300-outcome evaluation
            # would have had no live probabilities to score).
            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",
                probability=(round(float(macro.p_win), 1)
                             if macro.p_win is not None else None)))
        # 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