Harwinder Sekhon
b8f903ad0e
agent_panel.verdict_source defaults to "macro": the macro take's stance is the verdict, the synthesizer call is skipped (4 LLM calls per panel instead of 5); "chair" restores the four-voice consensus + OPPOSE guard. Verdict source is cached per mode and logged (panel_verdict_source). 406 tests. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FC4s6VCURFte4dub4xKxYL
1054 lines
45 KiB
Python
1054 lines
45 KiB
Python
"""
|
|
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 5 requests (4 personas + 1
|
|
synthesizer) against the shared DeepSeek daily cap (llm_client). Two
|
|
layers bound the spend on top of that cap:
|
|
|
|
- ``max_panels_per_day`` (config, default 20) — a UTC-day budget of
|
|
panel *attempts*, mirroring the client's request cap.
|
|
- A reuse TTL (default 3600 s) keyed on 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 len(takes) < 2:
|
|
logger.debug("agent panel: %d/4 persona takes — skipping synthesizer",
|
|
len(takes))
|
|
return None
|
|
order = {"macro": 0, "technical": 1, "risk": 2, "contrarian": 3}
|
|
takes.sort(key=lambda t: order.get(t.persona, 9))
|
|
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.
|
|
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
|