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 (2026-09-03, owner decision): all four personas always
run and are logged, but by default (``agent_panel.verdict_source =
"macro"``) the verdict shown to consumers is the macro persona's own
stance, and the synthesizer call is skipped (4 LLM calls per panel
instead of 5); ``"chair"`` restores the synthesizer + OPPOSE-guard path
below.

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 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 = "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.

_GOLD_DOCTRINE = """\
Gold doctrine: hawkish Fed / rate hikes / rising real yields / strong
dollar are gold-bearish; easing / rate cuts / safe-haven flows /
geopolitical stress / weak dollar are gold-bullish.
"""

_TAKE_CONTRACT = """\
Your review is advisory research on an already-generated signal; you are
one voice on a panel, not the decision maker.
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. If the
evidence your role needs is thin or missing, answer "neutral" with
confidence below 30, unless your role's instructions say otherwise.
Confidence scale (0-100): 80-100 strong one-sided evidence; 50-79 solid
case with some doubt; 30-49 mixed evidence; below 30 little basis to
judge.
Write the note 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:
{"stance": "support"|"oppose"|"neutral", "confidence": <0-100>,
 "note": "<max 15 words>"}
"stance" says whether YOUR analysis supports the proposed trade
direction, opposes it, or is genuinely mixed; "confidence" is how sure
you are of your stance; "note" is your single most important point.
"""

_MACRO_SYSTEM = """\
You are the macro analyst on a gold (XAU/USD) trade-review panel. You
receive JSON: a proposed trade brief ("brief": direction, timeframe,
quality, price, regime) plus the current fundamental picture — recent
directional "headlines", today's high-impact "events" (released: true
means the number is out), and optional aggregate "sentiment" for gold.
Neutral and unrelated headlines were removed by code. Each remaining
headline has a code-computed "gold_influence" and score; trust that
direction and do not reinterpret it from the wording alone.
Judge ONE thing: does the macro backdrop support the proposed direction?
Weigh released surprises over stale headlines; an imminent unreleased
high-impact event is a reason for lower confidence, not a direction.
Stance rules:
- support: the released data and news flow clearly push gold in the
  proposed direction.
- oppose: they clearly push gold the other way.
- neutral: the picture is mixed or stale — or there are no meaningful
  headlines or events to judge (score that low confidence).
""" + _GOLD_DOCTRINE + """\
Example: brief direction BUY, headlines dominated by hawkish Fed
repricing and a hot CPI print -> {"stance": "oppose", "confidence": 70,
"note": "Inflation is running hot, so interest rates should stay high — that usually pushes gold down."}
Example: brief direction BUY, headlines led by an escalating conflict
and a soft jobs report -> {"stance": "support", "confidence": 75,
"note": "Worries about war and a cooling economy usually send money into gold."}
""" + _TAKE_CONTRACT

_TECH_SYSTEM = """\
You are the technical analyst on a gold (XAU/USD) trade-review panel.
You receive JSON: a proposed trade brief ("brief") plus the technical
evidence — 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 of consecutive same-direction bars flagged by the engine, 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.
Judge ONE thing: is the technical picture internally coherent for the
proposed direction? Look for indicators fighting each other, the signal
fighting the regime (e.g. a BUY in trending_down), higher timeframes
disagreeing, price pressed into a key level, an exhausted move being
chased, or the candle path contradicting the indicator readings.
For XAU_CP, the named completed reversal/continuation pattern is the
setup trigger and may lead lagging indicators; indicator lag alone is
not sufficient reason to oppose it. Judge whether the pattern and the
remaining evidence are coherent together.
Stance rules:
- support: the indicators, higher timeframes, and regime mostly agree
  with the proposed direction and none of the conflicts above appear.
- oppose: one or more of those conflicts is present and serious.
- neutral: the evidence genuinely cuts both ways, or too much of it is
  missing to judge (score that low confidence).
Example: brief direction SELL in regime trending_up with H1 and H4 both
BUY -> {"stance": "oppose", "confidence": 80,
"note": "This bets on gold falling while the bigger picture still points up."}
Example: brief direction BUY in regime trending_up with M15 and H1 both
BUY and exhaustion 0 -> {"stance": "support", "confidence": 70,
"note": "Every way of looking at the chart points up, so buying goes with the flow."}
""" + _TAKE_CONTRACT

_RISK_SYSTEM = """\
You are the risk manager on a gold (XAU/USD) trade-review panel. You
receive JSON: a proposed trade brief ("brief") plus the risk facts —
the trade "setup" (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 setup's risk acceptable as constructed? Check
stop distance versus volatility, cost as a fraction of risk, trailing
geometry, and event landmines about to hit.
Stance rules:
- support: the trailing opportunity justifies the risk, the stop fits
  the current volatility, and no event landmine is imminent.
- oppose: a stop/trailing policy mismatched to volatility, strategy cost
  eating a big share of the stop, an imminent
  high-impact event — or no setup at all (a missing setup means the
  signal graded too low for one; oppose unless the rest is
  exceptional).
- neutral: acceptable but unremarkable — nothing dangerous, nothing
  attractive.
Example: RR 1.4 with volatility extreme and a high-impact event due ->
{"stance": "oppose", "confidence": 65,
"note": "The stop and trailing plan do not suit wild price swings before major news."}
Example: trailing enabled with normal volatility, low cost versus risk,
and no events due ->
{"stance": "support", "confidence": 70,
"note": "The stop and trailing plan fit calm conditions, with low costs and no major news."}
""" + _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 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, and
its warnings. Neutral and unrelated news was removed by code; trust each
headline's code-computed gold_influence instead of reinterpreting it.
Your job: construct the strongest honest case AGAINST the proposed
trade, then score how damaging that case is. Use "oppose" only when the
counter-case is genuinely strong, "neutral" when it is real but modest,
and "support" when even the skeptic finds nothing that bites.
""" + _GOLD_DOCTRINE + """\
Example: a BUY whose own engine lists three bearish reasons and a
counter-trend warning -> {"stance": "oppose", "confidence": 60,
"note": "The system's own list of negatives looks stronger than its reason to buy."}
""" + _TAKE_CONTRACT

_SYNTH_SYSTEM = """\
You are the chair of a gold (XAU/USD) trade-review panel. You receive
JSON: the proposed trade "direction" and the panel's "takes" — one per
analyst ("persona", "stance", "confidence", "note").
Weigh the takes into ONE consensus on whether to act on the signal:
"AGREE" (the panel backs the trade), "OPPOSE" (the panel is against it),
or "NEUTRAL" (genuinely split or low conviction). 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"
    confidence: float   # 0-100
    note: str


@dataclass
class PanelVerdict:
    verdict: str        # "AGREE" | "NEUTRAL" | "OPPOSE"
    confidence: float   # 0-100
    rationale: str
    takes: List[PersonaTake] = field(default_factory=list)
    created_ts: float = 0.0
    payload_hash: str = ""
    source: str = "chair"   # "macro" | "chair" — which path produced this


# ── 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"]) -> 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
    return payload


def _technical_payload(report: "SignalReport",
                       candles: 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
    return payload


def _risk_payload(report: "SignalReport") -> Dict[str, Any]:
    payload: Dict[str, Any] = {
        "brief": _brief(report),
        "volatility": report.volatility_regime,
        "warnings": list(report.warnings),
        "event_risk": report.event_risk,
        "event_description": report.event_description,
    }
    setup = _trade_setup_payload(report)
    if setup is not None:
        payload["setup"] = setup
    return payload


def _contrarian_payload(
    report: "SignalReport", context: Optional["FundamentalContext"] = None
) -> Dict[str, Any]:
    return {
        "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),
    }


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


# ── Orchestration ────────────────────────────────────────────────────

# Panel-private pool: the 4 persona calls run concurrently WITHOUT
# occupying telegram_bot's shared scan executor (which has 4 workers —
# borrowing them would stall every other scan for the panel's duration).
_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="agent-panel")

_VALID_STANCES = ("support", "oppose", "neutral")
_VALID_VERDICTS = ("AGREE", "NEUTRAL", "OPPOSE")


def _clamp_confidence(raw: Any) -> Optional[float]:
    try:
        return max(0.0, min(100.0, float(raw)))
    except (TypeError, ValueError):
        return None


def _validate_take(persona: str, reply: Any) -> Optional[PersonaTake]:
    if not isinstance(reply, dict):
        return None
    stance = reply.get("stance")
    if stance not in _VALID_STANCES:
        return None
    confidence = _clamp_confidence(reply.get("confidence"))
    if confidence is None:
        return None
    note = str(reply.get("note", "")).strip()[:160]
    return PersonaTake(persona=persona, stance=stance,
                       confidence=confidence, note=note)


def _ask_persona(persona: str, system: str,
                  payload: Dict[str, Any]) -> Optional[PersonaTake]:
    # Client resolved at call time: tests install stubs via lc._client.
    reply = get_llm_client().chat_json(system, payload)
    return _validate_take(persona, reply)


def _panel_jobs(
    report: "SignalReport",
    context: Optional["FundamentalContext"],
    candles: Optional[Dict[str, Any]],
) -> Tuple[Tuple[str, str, Dict[str, Any]], ...]:
    return (
        ("macro", _MACRO_SYSTEM, _macro_payload(report, context)),
        ("technical", _TECH_SYSTEM, _technical_payload(report, candles)),
        ("risk", _RISK_SYSTEM, _risk_payload(report)),
        ("contrarian", _CONTRARIAN_SYSTEM,
         _contrarian_payload(report, context)),
    )


# 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-v2",
        "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,
              ) -> Optional[PanelVerdict]:
    """Run 4 personas, then either take the macro persona's stance as the
    verdict (``verdict_source="macro"``, default — the synthesizer is not
    called) or weigh all four takes through the synthesizer chair
    (``"chair"``); 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
    jobs = jobs or _panel_jobs(report, context, candles)
    payload_hash = _panel_payload_hash(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
    order = {"macro": 0, "technical": 1, "risk": 2, "contrarian": 3}
    takes.sort(key=lambda t: order.get(t.persona, 9))
    if verdict_source == "macro":
        macro = next((t for t in takes if t.persona == "macro"), None)
        if macro is not None:
            return 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, "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 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} 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


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

        candles = _candle_context(report, settings)
        jobs = _panel_jobs(report, context, candles)
        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 ""
        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)
            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