726 lines
31 KiB
Python
726 lines
31 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).
|
|
|
|
Two invariants, both load-bearing (CLAUDE.md rules 1 and 4):
|
|
|
|
- ADVISORY / DISPLAY-ONLY. The verdict is written onto SignalReport
|
|
fields that no gate and no scoring path ever reads. Any failure —
|
|
LLM down, cap exhausted, garbage reply — degrades to "no panel",
|
|
never to a changed or blocked signal.
|
|
- NEVER runs inside ``analyze_from_dataframes``. The backtester replays
|
|
that function; consumers (telegram_bot, main --panel,
|
|
dashboard_publisher) attach the panel POST-HOC via ``attach_panel``,
|
|
so replayed books can never see a panel field. Backtest parity holds
|
|
by construction.
|
|
|
|
Cost model: every full panel spends 5 requests (4 personas + 1
|
|
synthesizer) against the shared DeepSeek daily cap (llm_client). Two
|
|
layers bound the spend on top of that cap:
|
|
|
|
- ``max_panels_per_day`` (config, default 20) — a UTC-day budget of
|
|
panel *attempts*, mirroring the client's request cap.
|
|
- A reuse TTL (default 3600 s) keyed on (symbol, timeframe, direction):
|
|
a persisting signal re-serves its cached verdict across bars instead
|
|
of re-spending. Without this, one sticky M5 dashboard signal would
|
|
burn 5 calls per 5-minute bar — 1,440/day against the 500 cap.
|
|
|
|
Prompts are STATIC module constants (DeepSeek prefix-cache pricing —
|
|
see news_engine.py); all variable content travels in the user payload.
|
|
|
|
Payload enrichment (2026-08-15, owner-approved, advisory-only): the
|
|
technical persona also receives raw indicator VALUES (with a units
|
|
legend in the prompt) and ATR-NORMALIZED recent 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).
|
|
|
|
Verdict logging (2026-08-15): after a successful attach, the verdict is
|
|
written back onto the scan's signals.db row via signal_log's rowid
|
|
hand-off — building the "does OPPOSE predict losers?" dataset without
|
|
changing any live behavior.
|
|
"""
|
|
|
|
import datetime as _dt
|
|
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
|
|
|
|
# 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
|
|
"headlines", today's high-impact "events" (released: true means the
|
|
number is out), a news "sentiment" score for gold, and an optional
|
|
"news_summary" blurb.
|
|
Judge ONE thing: does the macro backdrop support the proposed direction?
|
|
Weigh released surprises over stale headlines; an imminent unreleased
|
|
high-impact event is a reason for lower confidence, not a direction.
|
|
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 — 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); keltner = dollar distance from the mid-line;
|
|
ema_trend = 0-1 alignment score; divergence and candlestick = pattern
|
|
counts.
|
|
"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.
|
|
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" (entry, stop_loss, take_profit_1/2, risk_reward,
|
|
risk_dollars), the "volatility" regime, engine "warnings", the calendar
|
|
"event_risk" level, and the assumed "spread".
|
|
Judge ONE thing: is this setup's risk acceptable as constructed? Check
|
|
reward-to-risk adequacy, stop distance versus the volatility regime
|
|
(a tight stop in extreme vol dies to noise; a huge stop in low vol is
|
|
oversized), spread as a fraction of the stop, and event landmines about
|
|
to hit.
|
|
Stance rules:
|
|
- support: the likely reward clearly justifies the risk, the stop fits
|
|
the current volatility, and no event landmine is imminent.
|
|
- oppose: reward too thin for the risk, a stop mismatched to the
|
|
volatility, spread 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 possible profit is small for such wild price swings, and big economic news is due soon."}
|
|
Example: RR 2.4 with normal volatility and no events due ->
|
|
{"stance": "support", "confidence": 70,
|
|
"note": "The possible profit is more than double the possible loss, and conditions look calm."}
|
|
""" + _TAKE_CONTRACT
|
|
|
|
_CONTRARIAN_SYSTEM = """\
|
|
You are the contrarian on a gold (XAU/USD) trade-review panel — the
|
|
designated skeptic. You receive JSON: a proposed trade brief ("brief"),
|
|
the engine's own "bullish_reasons" and "bearish_reasons", its
|
|
"warnings", and an optional "news_summary".
|
|
Your job: construct the strongest honest case AGAINST the proposed
|
|
trade, then score how damaging that case is. Use "oppose" only when the
|
|
counter-case is genuinely strong, "neutral" when it is real but modest,
|
|
and "support" when even the skeptic finds nothing that bites.
|
|
""" + _GOLD_DOCTRINE + """\
|
|
Example: a BUY whose own engine lists three bearish reasons and a
|
|
counter-trend warning -> {"stance": "oppose", "confidence": 60,
|
|
"note": "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). Weigh a high-
|
|
confidence oppose from any single analyst seriously — one strong veto
|
|
can outweigh two lukewarm supports. Do not average 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 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 _macro_payload(report: "SignalReport",
|
|
context: Optional["FundamentalContext"]) -> Dict[str, Any]:
|
|
headlines: List[str] = []
|
|
events: List[Dict[str, Any]] = []
|
|
if context is not None:
|
|
headlines = [h.headline for h in context.news_headlines[:10]]
|
|
events = [
|
|
{"event": e.event, "currency": e.currency,
|
|
"actual": e.actual, "forecast": e.forecast,
|
|
"released": not e.is_upcoming}
|
|
for e in context.events_today if e.is_high_impact
|
|
][:8]
|
|
return {
|
|
"brief": _brief(report),
|
|
"headlines": headlines,
|
|
"events": events,
|
|
"sentiment": round(report.news_sentiment, 3),
|
|
"news_summary": report.news_summary,
|
|
}
|
|
|
|
|
|
def _technical_payload(report: "SignalReport",
|
|
candles: Optional[Dict[str, Any]] = None
|
|
) -> Dict[str, Any]:
|
|
indicators: Dict[str, Dict[str, Any]] = {}
|
|
key_levels: Dict[str, float] = {}
|
|
analysis = report.primary_analysis
|
|
if analysis is not None:
|
|
# Same visibility filter as main.report_to_dict: private series
|
|
# ride as _-prefixed pseudo-indicators, weight-0 = context-only.
|
|
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
|
|
key_levels = {k: round(v, 2) for k, v in analysis.key_levels.items()}
|
|
payload: Dict[str, Any] = {
|
|
"brief": _brief(report),
|
|
"indicators": indicators,
|
|
"mtf": {tf: {"signal": sig.name,
|
|
"score": round(report.mtf_scores.get(tf, 0.0), 1)}
|
|
for tf, sig in report.mtf_signals.items()},
|
|
"key_levels": key_levels,
|
|
"exhaustion": report.exhaustion_count,
|
|
}
|
|
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,
|
|
"spread": round(report.spread_estimate, 2),
|
|
}
|
|
ts = report.trade_setup
|
|
if ts is not None:
|
|
payload["setup"] = {
|
|
"direction": ts.direction,
|
|
"entry": round(ts.entry_price, 2),
|
|
"stop_loss": round(ts.stop_loss, 2),
|
|
"take_profit_1": round(ts.take_profit_1, 2),
|
|
"take_profit_2": round(ts.take_profit_2, 2),
|
|
"risk_reward": round(ts.risk_reward, 2),
|
|
"risk_dollars": round(ts.risk_pips * ts.pip_value, 2),
|
|
}
|
|
return payload
|
|
|
|
|
|
def _contrarian_payload(report: "SignalReport") -> Dict[str, Any]:
|
|
return {
|
|
"brief": _brief(report),
|
|
"bullish_reasons": report.bullish_reasons[:3],
|
|
"bearish_reasons": report.bearish_reasons[:3],
|
|
"warnings": list(report.warnings),
|
|
"news_summary": report.news_summary,
|
|
}
|
|
|
|
|
|
# ── 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 run_panel(report: "SignalReport",
|
|
context: Optional["FundamentalContext"] = None,
|
|
candles: Optional[Dict[str, Any]] = None,
|
|
) -> Optional[PanelVerdict]:
|
|
"""Run 4 personas + synthesizer; None = no verdict (fail-safe).
|
|
|
|
Requires ≥2 valid persona takes before spending the synthesizer
|
|
request; the synthesizer's reply is validated strictly and there is
|
|
deliberately NO programmatic fallback verdict — a panel that could
|
|
not really convene reports nothing rather than something invented.
|
|
"""
|
|
direction = _direction(report)
|
|
if direction is None:
|
|
return None
|
|
jobs = (
|
|
("macro", _MACRO_SYSTEM, _macro_payload(report, context)),
|
|
("technical", _TECH_SYSTEM, _technical_payload(report, candles)),
|
|
("risk", _RISK_SYSTEM, _risk_payload(report)),
|
|
("contrarian", _CONTRARIAN_SYSTEM, _contrarian_payload(report)),
|
|
)
|
|
futures = [_pool.submit(_ask_persona, name, system, payload)
|
|
for name, system, payload in jobs]
|
|
# The HTTP timeout inside the client bounds each call; the margin
|
|
# covers pool queueing. Stragglers past the wait are abandoned (their
|
|
# threads finish harmlessly) and simply don't vote.
|
|
client_timeout = getattr(get_llm_client(), "_timeout", DEFAULT_TIMEOUT)
|
|
done, _ = _fwait(futures, timeout=client_timeout + 8.0)
|
|
takes = [f.result() for f in done if not f.exception()]
|
|
takes = [t for t in takes if t is not None]
|
|
if len(takes) < 2:
|
|
logger.debug("agent panel: %d/4 persona takes — skipping synthesizer",
|
|
len(takes))
|
|
return None
|
|
order = {"macro": 0, "technical": 1, "risk": 2, "contrarian": 3}
|
|
takes.sort(key=lambda t: order.get(t.persona, 9))
|
|
reply = get_llm_client().chat_json(_SYNTH_SYSTEM, {
|
|
"direction": direction,
|
|
"takes": [{"persona": t.persona, "stance": t.stance,
|
|
"confidence": t.confidence, "note": t.note}
|
|
for t in takes],
|
|
})
|
|
if not isinstance(reply, dict):
|
|
return None
|
|
verdict = reply.get("verdict")
|
|
confidence = _clamp_confidence(reply.get("confidence"))
|
|
if verdict not in _VALID_VERDICTS or confidence is None:
|
|
return None
|
|
rationale = str(reply.get("rationale", "")).strip()[:240]
|
|
return PanelVerdict(verdict=verdict, confidence=confidence,
|
|
rationale=rationale, takes=takes,
|
|
created_ts=time.time())
|
|
|
|
|
|
# ── Cache, budget, attach ────────────────────────────────────────────
|
|
|
|
_state_lock = threading.Lock()
|
|
# Exact result per reviewed bar: repeated attaches on the same signal
|
|
# (re-scan, multi-consumer) are free.
|
|
_cache: "OrderedDict[Tuple[str, str, str, str], PanelVerdict]" = OrderedDict()
|
|
_CACHE_MAX = 32
|
|
# Reuse layer: a persisting signal (same symbol/tf/direction, newer bar)
|
|
# re-serves its last verdict for reuse_ttl_seconds instead of re-spending
|
|
# 5 requests per bar (see module docstring for the M5 dashboard math).
|
|
_last_by_dir: Dict[Tuple[str, str, str], PanelVerdict] = {}
|
|
# UTC-day budget of panel attempts (mirrors llm_client._under_cap).
|
|
_panel_day: Optional[_dt.date] = None
|
|
_panel_count: int = 0
|
|
_warned_budget = False
|
|
|
|
|
|
def reset_panel_state() -> None:
|
|
"""Drop caches and counters (tests / config reloads)."""
|
|
global _panel_day, _panel_count, _warned_budget
|
|
with _state_lock:
|
|
_cache.clear()
|
|
_last_by_dir.clear()
|
|
_panel_day = None
|
|
_panel_count = 0
|
|
_warned_budget = False
|
|
|
|
|
|
def panel_health() -> Dict[str, Any]:
|
|
"""Read-only counters for status displays."""
|
|
settings = get_app_config().agent_panel_settings
|
|
cap = int(settings.get("max_panels_per_day", DEFAULT_MAX_PANELS_PER_DAY))
|
|
today = _dt.datetime.now(_dt.timezone.utc).date()
|
|
with _state_lock:
|
|
used = _panel_count if _panel_day == today else 0
|
|
return {"used_today": used, "cap": cap}
|
|
|
|
|
|
def _under_panel_budget(cap: int) -> bool:
|
|
"""Count one panel attempt against the UTC-day budget."""
|
|
global _panel_day, _panel_count, _warned_budget
|
|
today = _dt.datetime.now(_dt.timezone.utc).date()
|
|
with _state_lock:
|
|
if _panel_day != today:
|
|
_panel_day = today
|
|
_panel_count = 0
|
|
_warned_budget = False
|
|
if _panel_count >= cap:
|
|
if not _warned_budget:
|
|
_warned_budget = True
|
|
logger.warning(
|
|
"agent panel: daily budget (%d) exhausted — no more "
|
|
"panels until UTC midnight", cap)
|
|
return False
|
|
_panel_count += 1
|
|
return True
|
|
|
|
|
|
def _write_fields(report: "SignalReport", v: PanelVerdict) -> None:
|
|
report.panel_verdict = v.verdict
|
|
report.panel_confidence = float(v.confidence)
|
|
report.panel_rationale = v.rationale
|
|
report.panel_takes = [
|
|
{"persona": t.persona, "stance": t.stance,
|
|
"confidence": t.confidence, "note": t.note} for t in v.takes]
|
|
|
|
|
|
def _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
|
|
direction = _direction(report)
|
|
if direction is None:
|
|
return False
|
|
|
|
analysis = report.primary_analysis
|
|
tf = analysis.timeframe if analysis else ""
|
|
exact_key = (report.symbol, tf, report.timestamp, direction)
|
|
dir_key = (report.symbol, tf, direction)
|
|
ttl = float(settings.get("reuse_ttl_seconds",
|
|
DEFAULT_REUSE_TTL_SECONDS))
|
|
with _state_lock:
|
|
hit = _cache.get(exact_key)
|
|
if hit is None:
|
|
prior = _last_by_dir.get(dir_key)
|
|
if prior is not None and time.time() - prior.created_ts < ttl:
|
|
hit = prior
|
|
if hit is not None:
|
|
_write_fields(report, hit)
|
|
_log_panel_verdict(report)
|
|
return True
|
|
|
|
cap = int(settings.get("max_panels_per_day",
|
|
DEFAULT_MAX_PANELS_PER_DAY))
|
|
if not _under_panel_budget(cap):
|
|
return False
|
|
|
|
if context is None:
|
|
# Best-effort; the news engine's class-level 600 s caches make
|
|
# this nearly free when warm in both bot and dashboard.
|
|
try:
|
|
from config import PlatformConfig
|
|
from news_engine import NewsEngine
|
|
context = NewsEngine(
|
|
PlatformConfig().news).get_fundamental_context()
|
|
except Exception as e:
|
|
logger.debug("agent panel: news context unavailable: %s", e)
|
|
context = None
|
|
|
|
candles = _candle_context(report, settings)
|
|
verdict = run_panel(report, context, candles=candles)
|
|
if verdict is None:
|
|
return False
|
|
_write_fields(report, verdict)
|
|
with _state_lock:
|
|
_cache[exact_key] = verdict
|
|
while len(_cache) > _CACHE_MAX:
|
|
_cache.popitem(last=False)
|
|
_last_by_dir[dir_key] = verdict
|
|
_log_panel_verdict(report)
|
|
return True
|
|
except Exception:
|
|
logger.exception("agent panel: attach failed — signal unaffected")
|
|
return False
|