From 4dd0ca0b462799965baa847a64eabe6afb41bef2 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 23 Aug 2026 12:57:05 -0400 Subject: [PATCH] agent_panel: harden evidence and track outcomes --- CLAUDE.md | 10 +- advisory_outcome.py | 343 ++++++++++++++++++++++++++++++ agent_panel.py | 486 ++++++++++++++++++++++++++++++++----------- main.py | 5 + signal_aggregator.py | 81 +++++++- signal_log.py | 144 +++++++++++-- test_all.py | 403 +++++++++++++++++++++++++++++++++-- 7 files changed, 1313 insertions(+), 159 deletions(-) create mode 100644 advisory_outcome.py diff --git a/CLAUDE.md b/CLAUDE.md index 4d19e38..78299b4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -11,7 +11,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest ## Commands ```bash -python test_all.py # QA suite: 381 tests, 24 modules, no network (mocked) +python test_all.py # QA suite: 406 tests, 24 modules, no network (mocked) python main.py --pair XAUUSD --timeframe H1 --detail # CLI signal report (--json for full dump, # --panel to run the advisory LLM agent panel on-demand) python telegram_bot.py # the bot (refuses to boot without credentials) @@ -29,7 +29,7 @@ python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-0 # 2026-08-05: NOT shipped — backtest/research/fixedtp_2026-08-05.md) ``` -**`test_all.py` is NOT pytest** — it uses a custom `@test("name")` decorator that runs each test immediately at import; don't grep for `def test_` or try `pytest test_all.py`. Run the whole file; it prints per-module results and exits non-zero on failure. There is no way to run a single test short of commenting modules out. Tests must stay green; every new gate/knob needs tests. When patching weekend logic in tests, patch `providers.base._is_forex_weekend` — but know there are THREE independent weekend implementations (`providers/base.py`, `signal_aggregator.py`, and an inline copy in `backtest/gpu_backtest.py:_is_xau_blocked_by_calendar`); patching the provider one does not reach the other two. +**`test_all.py` is NOT pytest** — it uses a custom `@test("name")` decorator that runs each test immediately at import; don't grep for `def test_` or try `pytest test_all.py`. Run the whole file; it prints per-module results and exits non-zero on failure. There is no way to run a single test short of commenting modules out. Tests must stay green; every new gate/knob needs tests. When patching weekend logic in tests, patch `providers.base._is_forex_weekend` — but know there are THREE independent signal-path weekend implementations (`providers/base.py`, `signal_aggregator.py`, and an inline copy in `backtest/gpu_backtest.py:_is_xau_blocked_by_calendar`) plus the advisory-only continuity copy in `advisory_outcome.py`; patching one does not reach the others. Python 3.12; a `venv/` exists in-repo. Dependencies in `requirements.txt` (numpy, pandas, requests, twelvedata, python-telegram-bot ≥20, websockets). @@ -59,9 +59,9 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - **`engine.py`** — `TechnicalEngine`: 9 scored indicators (ema_trend, rsi, macd, bollinger, adx, roc, keltner, divergence, candlestick) + ATR as context. Regime classification (trending_up/down, `trending` = mixed/direction-UNKNOWN, transitioning, ranging) drives per-indicator weight multipliers and regime-aware RSI reinterpretation. Compute order matters (ATR first; MACD always computed even when not a pillar — divergence needs the histogram). Private series ride on the result as `_underscore` attrs; scorers/formatters skip anything `_`-prefixed or weight-0. - **`signal_aggregator.py`** — MTF agreement, diminishing-decay penalty system, exhaustion hard block, quality grading, then an ordered chain of post-grade gates (session, XAU SELL quality/vol/hour, regime blocks, M5 noise gates), then three setup generators: XAU ATR (A+/A), generic swing-based, plus the M5 pullback (`XAU_PB`) and chart-pattern (`XAU_CP`, `chart_patterns.py`) entry paths. The PB/CP generators run AFTER the gate chain and force quality to "B"; the CP path deliberately accepts grade-F composites and re-checks only session/day/big-candle (via warning prefixes) plus its own SELL/vol/RSI gates — it does NOT re-check exhaustion or the M5 min-score/MTF gates. **This is validated behavior, not a bug** (2026-08-04 audit: 226/586 trades in the shipped M5 12mo book are XAU_CP, +38.9R, 104 of them exhaustion-flagged and net-positive) — changing it is a rule-1 revalidation event. `TradeSetup.strategy` selects trailing params; trail params are stamped on the setup from the *producing* config so per-TF overrides reach formatters. - **`pair_signals/`** — per-pair config factories. **Always use the factory (`produce_signal` / `make_config`), never bare `PlatformConfig()`** — the factory applies the config.json overlay first, then the validated per-TF deltas (M5 hour-9 block, M15 trail arm 0.70, M15 SELL-hour block, etc.; evidence in the docstring). A test enforces factory ≡ inline-config equivalence. Confirmation map: M5→[M15,H1], M15→[M5,H1], H1→[M15,H4], H4→[H1,D1]. -- **`signal_log.py`** — SQLite signal-history logger (stdlib only). Hooked into `SignalAggregator.analyze_pair` only (the live entry point — the backtester never writes here, preserving parity), so every live scan from every surface (Telegram `telegram`, CLI `cli`, Oanda daemon `oanda_daemon`, dashboard `dashboard`) lands one row in `logs/signals.db`: signal/grade/price, MTF + trade-setup + reasons JSON, full per-TF indicator snapshots, the last 14 closed OHLCV candles per fetched TF, and the news at scan time (event risk + calendar events + scored headlines JSON, re-fetched via the NewsEngine class-level caches). Config.json `signal_log` block (`enabled`, `path`; env `SIGNAL_LOG_ENABLED`/`SIGNAL_LOG_PATH`). **Fail-safe by contract** — a logging failure can never break a scan. Panel fields arrive post-hoc: `log_signal` stashes the rowid on the report and `agent_panel.attach_panel` writes the verdict back onto the same row via `log_panel_update` (no rowid — backtest replay or logging disabled — is a silent no-op). +- **`signal_log.py` / `advisory_outcome.py`** — SQLite signal-history logger (stdlib only). Hooked into `SignalAggregator.analyze_pair` only (the live entry point — the backtester never writes here, preserving parity), so every live scan from every surface (Telegram `telegram`, CLI `cli`, Oanda daemon `oanda_daemon`, dashboard `dashboard`) lands one row in `logs/signals.db`: signal/grade/price, MTF + trade-setup + reasons JSON, full per-TF indicator snapshots, the last 14 closed OHLCV candles per fetched TF, and the news at scan time (event risk + calendar events + scored headlines JSON, re-fetched via the NewsEngine class-level caches). Config.json `signal_log` block (`enabled`, `path`; env `SIGNAL_LOG_ENABLED`/`SIGNAL_LOG_PATH`). **Fail-safe by contract** — a logging failure can never break a scan. Panel fields arrive post-hoc: `log_signal` stashes the rowid and `agent_panel.attach_panel` writes verdict + creation/cache provenance + payload hash onto the same row. A versioned advisory-only outcome then advances from later stored closed-candle snapshots using next-open reanchoring, close-only trailing, continuity checks that allow only scheduled weekend/maintenance closures, a 300-bar timeout, and both $0.30/$0.65 cost views; it is explicitly a cleaned-mid evidence model, not account execution or raw-path backtest P&L. No rowid means silent no-op. - **`news_engine.py`** — Forex Factory calendar (MyFXBook RSS fallback) + RSS keyword sentiment with gold-inverse keywords. Class-level caches (fresh engine per scan). Optional DeepSeek enrichment via `llm_client.py` (requests-only, no SDK): batched headline sentiment overriding keyword scores, gold-direction on release surprises, and a market-summary blurb (`SignalReport.news_summary`) — **display/warning-only, keyword fallback on any failure, inert without a key**; prompts are static module constants (DeepSeek prefix-cache pricing) and requests send `thinking: disabled` (V4-Flash reasoning otherwise blows the timeout). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`) — the LLM does not change that. -- **`agent_panel.py`** — ai-hedge-fund-style advisory panel: 4 DeepSeek persona calls (macro/technical/risk/contrarian, concurrent on a panel-private pool) + 1 synthesizer producing an **ADVISORY-ONLY** consensus (`AGREE/NEUTRAL/OPPOSE` + confidence + rationale) written onto `SignalReport.panel_*` fields. Attached **post-hoc** by consumers (`telegram_bot` alert path, `main --panel`, `dashboard_publisher`) — never runs inside `analyze_from_dataframes` (backtest parity, test-locked) and no gate reads its fields. Static prompts (prefix caching); spend bounded by an exact per-bar cache, a reuse TTL keyed (symbol, TF, direction) so persisting signals don't re-spend every bar, and a `max_panels_per_day` UTC budget (default 20 ≈ ≤100 requests) on top of the shared 500/day client cap. `<2` valid persona takes skips the synthesizer; any failure degrades to no panel, never a changed signal. The technical persona's payload is enriched (2026-08-15): raw indicator values plus ATR-normalized recent candles per TF (`candles_enabled`/`candle_bars_primary` 20/`candle_bars_confirm` 10 config keys; code does the arithmetic — raw prices are the model's weakest input, per `backtest/research/llm_news_eval_2026-08-06.md`). Prompts forbid training-data/outside knowledge and cap notes at 15 words / rationales at 30 to bound output tokens. +- **`agent_panel.py`** — ai-hedge-fund-style advisory panel: 4 DeepSeek persona calls (macro/technical/risk/contrarian, concurrent on a panel-private pool) + 1 synthesizer producing an **ADVISORY-ONLY** consensus (`AGREE/NEUTRAL/OPPOSE` + confidence + rationale) written onto `SignalReport.panel_*` fields. Attached **post-hoc** by consumers (`telegram_bot` alert path, `main --panel`, `dashboard_publisher`) — never runs inside `analyze_from_dataframes` (backtest parity, test-locked) and no gate reads its fields. Persona payloads are role-filtered: macro gets high-impact events plus only code-scored directional news; technical gets strategy/pattern, primary indicators, MTF, levels, and ATR-normalized candles; risk gets setup strategy, strategy cost assumption, trailing semantics, volatility/events/warnings (never the median candle range as spread); contrarian gets directional news, setup, timestamped latest-closed price, indicators, reasons, and warnings. Spend is bounded by exact cache, full-evidence-identity TTL reuse, and `max_panels_per_day`; changed evidence never inherits a direction-only cached verdict. Verdict provenance logs creation time, `MISS`/`EXACT_HIT`/`TTL_REUSE`, and a SHA-256 payload hash. Prompts forbid outside knowledge and outputs remain short. - **`market_outlook.py`** — hourly DeepSeek market brief for XAU/USD, using ATR-normalized M15/H1/H4 candles, visible indicator states, levels, news, and immutable recent signal history. It returns a written **ADVISORY-ONLY** bias/setup-forming assessment and persists it independently in `outlooks` within `logs/signals.db` together with the news, indicator/regime/level, and candle evidence sent to the model. It may be sent post-scan by Telegram or requested with `main.py --outlook`; it never imports or runs inside the signal path. A lock enforces hourly dedup plus its own UTC-day cap (default 12), and provider/news failures leave a safe partial brief while LLM/database failures degrade to no outlook. - **`telegram_bot.py`** — owner-locked bot + subscriber channel; candle-boundary auto-scan loops (M5/M15/H1/H4, boundary+10 s), alert quality threshold + fingerprint dedup (committed only after a successful send), 📰 LLM market-context line on alerts when available, WebSocket level-breach monitor fed by the streaming provider (`data_providers.streaming`, falling back to the active provider; auto-disabled when the resolved provider can't stream or has no API key — Oanda does not stream, Tiingo does). Rotating log: `logs/telegram.log`, 3-h rotation × 240 backups. - **`oanda_daemon.py` / `oanda_executor.py`** — practice-account auto-execution. Book-parity rules (see daemon docstring): no broker-side TP (trailing model — TP1/TP2 advisory), 5-bar per-TF cooldown (wall-clock approximation, so it expires over weekends where the backtester would still count bars), trail ratchets at bar close once per *observed* bar (a missed/failed cycle skips those bars entirely — the intra-gap high/low never enters the peak). Accepted divergence: one position total; orders send `positionFill=OPEN_ONLY` so a state/broker divergence cancels the order instead of silently netting. State/broker re-sync runs at boot and after any suspicious fill (`_reconcile_positions`). @@ -82,7 +82,7 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz ## 2026-08-04 QA audit — owner decisions ledger -- **Weekend calendar: FIXED (owner-approved 2026-08-04).** All three `_is_forex_weekend` copies are NY-anchored (Fri 17:00 → Sun 18:00 ET; during US DST that is Fri 21:00 UTC — measured on 5yr of Oanda bars). Kills the weekly false-"STALE DATA/feed lagging" hours. Book impact at the audit HEAD: M15 byte-identical; M5 12mo dropped exactly ONE trade — a phantom whose signal fired on the final summer-Friday bar (close 21:00 UTC) and "filled" at Sunday's reopen 49 h later (−1.08R, live-impossible), leaving 555 trades / +98.64R / PF≈1.50 @ $0.65 (was 556/+97.56/1.48). The current-HEAD canonical reference is recorded below. A Module 21 test sweeps all three copies hourly across both DST transitions. +- **Weekend calendar: FIXED (owner-approved 2026-08-04).** All three signal-path `_is_forex_weekend` copies are NY-anchored (Fri 17:00 → Sun 18:00 ET; during US DST that is Fri 21:00 UTC — measured on 5yr of Oanda bars). Kills the weekly false-"STALE DATA/feed lagging" hours. Book impact at the audit HEAD: M15 byte-identical; M5 12mo dropped exactly ONE trade — a phantom whose signal fired on the final summer-Friday bar (close 21:00 UTC) and "filled" at Sunday's reopen 49 h later (−1.08R, live-impossible), leaving 555 trades / +98.64R / PF≈1.50 @ $0.65 (was 556/+97.56/1.48). The current-HEAD canonical reference is recorded below. A Module 21 test sweeps those copies plus the advisory outcome's continuity copy hourly across both DST transitions. - **Spread in exit-trigger detection (rule 6): owner DECLINED 2026-08-04.** Backtest stop triggers stay on raw mid High/Low; the measured optimism (~4–6% of total R: −3.55R @ $0.30, −4.24R @ $0.65 on the M15 12mo book) is accepted and books stay comparable. Re-open only as a conscious re-pin campaign. - **`FB2_EMA_LO`: HARDENED 2026-08-04 (delegated).** Research override for the EMA200-proximity band, default = tuned 0.5; an unparseable value now degrades to the default with a warning instead of crashing every scan. Documented here — it is a knob that exists. - **H4 seeding divergence: parked.** Live H4 fetches only 380 bars vs the backtest's unbounded history — ~2% of sampled H4 bars flip an indicator signal (composite Δ up to 12.3). M5/M15/H1 clean at their real fetch depths. Fold into any future H4 campaign; H4 remains "unproven for auto-execution". diff --git a/advisory_outcome.py b/advisory_outcome.py new file mode 100644 index 0000000..1f6eb76 --- /dev/null +++ b/advisory_outcome.py @@ -0,0 +1,343 @@ +"""Deterministic advisory trade outcomes from stored closed candles. + +This module measures the hypothetical setup reviewed by the LLM panel. It +never changes a signal, panel verdict, order, or daemon state. The model uses +the same next-open reanchoring and close-only trailing rules as the backtester, +but its evidence is the cleaned mid-candle snapshots stored in signals.db. +""" + +from __future__ import annotations + +import math +from datetime import datetime, timedelta +from typing import Any, Dict, Iterable, List, Optional, Tuple +from zoneinfo import ZoneInfo + +MODEL = "stored-closed-mid-next-open-trail-v1" +MAX_HOLD_BARS = 300 +REFERENCE_COST = 0.30 +REALISTIC_COST = 0.65 + +_TF_ALIASES = { + "M1": "1min", "1MIN": "1min", "1min": "1min", + "M5": "5min", "5MIN": "5min", "5min": "5min", + "M15": "15min", "15MIN": "15min", "15min": "15min", + "M30": "30min", "30MIN": "30min", "30min": "30min", + "H1": "1h", "1H": "1h", "1h": "1h", + "H4": "4h", "4H": "4h", "4h": "4h", + "D1": "1day", "1D": "1day", "1d": "1day", + "1DAY": "1day", "1day": "1day", +} + +_TF_SECONDS = { + "1min": 60, + "5min": 5 * 60, + "15min": 15 * 60, + "30min": 30 * 60, + "1h": 60 * 60, + "4h": 4 * 60 * 60, + "1day": 24 * 60 * 60, +} + +_NY_TZ = ZoneInfo("America/New_York") + + +def _is_forex_weekend(now: datetime) -> bool: + """NY-anchored XAU weekend closure; test-locked to the signal copies.""" + ny = now.astimezone(_NY_TZ) + weekday = ny.weekday() + if weekday == 4: + return ny.hour >= 17 + if weekday == 5: + return True + if weekday == 6: + return ny.hour < 18 + return False + + +def normalize_timeframe(value: Any) -> str: + text = str(getattr(value, "value", value) or "") + return _TF_ALIASES.get(text, _TF_ALIASES.get(text.upper(), text)) + + +def timeframe_value(mapping: Any, timeframe: Any) -> Any: + """Read a timeframe-keyed snapshot regardless of enum/storage alias.""" + if not isinstance(mapping, dict): + return None + wanted = normalize_timeframe(timeframe) + for key, value in mapping.items(): + if normalize_timeframe(key) == wanted: + return value + return None + + +def _finite(value: Any) -> Optional[float]: + try: + number = float(value) + except (TypeError, ValueError): + return None + return number if math.isfinite(number) else None + + +def _timestamp(value: Any) -> Optional[datetime]: + try: + return datetime.fromisoformat(str(value).replace(" ", "T")) + except (TypeError, ValueError): + return None + + +def initial_outcome( + setup: Optional[Dict[str, Any]], + indicators: Optional[Dict[str, Any]], + timeframe: Any, +) -> Tuple[str, Dict[str, Any]]: + """Freeze the setup and deterministic evaluation policy.""" + if not setup: + return "NOT_APPLICABLE", { + "model": MODEL, + "status": "NOT_APPLICABLE", + "reason": "panel row has no trade setup", + } + + tf = normalize_timeframe(timeframe) + analysis = timeframe_value(indicators, tf) + signal_bar = analysis.get("timestamp") if isinstance(analysis, dict) else None + direction = str(setup.get("direction", "")) + entry = _finite(setup.get("entry")) + stop = _finite(setup.get("stop_loss")) + tp1 = _finite(setup.get("take_profit_1")) + arm = _finite(setup.get("trail_arm_r")) + distance = _finite(setup.get("trail_dist_r")) + risk = abs(entry - stop) if entry is not None and stop is not None else 0.0 + if (direction not in ("BUY", "SELL") or not signal_bar + or _timestamp(signal_bar) is None or risk <= 0 or tp1 is None + or arm is None or distance is None or arm < 0 or distance < 0): + return "UNRESOLVABLE", { + "model": MODEL, + "status": "UNRESOLVABLE", + "reason": "invalid or incomplete setup/outcome seed", + } + + state = { + "model": MODEL, + "status": "PENDING", + "timeframe": tf, + "signal_bar_at": str(signal_bar), + "setup": { + "direction": direction, + "reported_entry": entry, + "stop_distance": risk, + "tp1_distance": abs(tp1 - entry), + "strategy": setup.get("strategy"), + "setup_pattern": setup.get("setup_pattern"), + }, + "policy": { + "entry": "next_primary_bar_open_reanchored", + "max_hold_bars": MAX_HOLD_BARS, + "trail_arm_r": arm, + "trail_dist_r": distance, + "tp1_is_advisory": arm > 0, + "reference_cost": REFERENCE_COST, + "realistic_cost": REALISTIC_COST, + }, + "progress": { + "last_processed_bar": str(signal_bar), + "entry_at": None, + "entry_price": None, + "initial_stop": None, + "current_stop": None, + "peak": None, + "bars_held": 0, + "mfe_r": 0.0, + "mae_r": 0.0, + }, + "result": None, + } + return "PENDING", state + + +def _valid_bars(bars: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: + valid = [] + for bar in bars: + stamp = _timestamp(bar.get("timestamp")) if isinstance(bar, dict) else None + values = [_finite(bar.get(key)) for key in ("open", "high", "low", "close")] + if stamp is None or any(value is None for value in values): + continue + valid.append({ + "timestamp": str(bar["timestamp"]), + "_timestamp": stamp, + "open": values[0], "high": values[1], + "low": values[2], "close": values[3], + }) + valid.sort(key=lambda item: item["_timestamp"]) + return valid + + +def _has_continuity( + bars: List[Dict[str, Any]], last: datetime, timeframe: str +) -> bool: + """Reject missing tradable bars while allowing scheduled XAU closures.""" + seconds = _TF_SECONDS.get(normalize_timeframe(timeframe)) + if seconds is None: + return False + step = timedelta(seconds=seconds) + previous = last + seen = {last} + for bar in bars: + stamp = bar["_timestamp"] + if stamp <= previous or stamp in seen: + continue + expected = previous + step + while expected < stamp: + ny = expected.astimezone(_NY_TZ) + daily_maintenance = ny.weekday() < 4 and ny.hour == 17 + if not (_is_forex_weekend(expected) or daily_maintenance): + return False + expected += step + if expected != stamp: + return False + previous = stamp + seen.add(stamp) + return True + + +def _finish( + state: Dict[str, Any], reason: str, exit_at: str, exit_price: float +) -> None: + setup = state["setup"] + progress = state["progress"] + entry = float(progress["entry_price"]) + risk = float(setup["stop_distance"]) + pnl = ((exit_price - entry) if setup["direction"] == "BUY" + else (entry - exit_price)) + gross_r = pnl / risk + state["status"] = "RESOLVED" + state["result"] = { + "exit_at": exit_at, + "exit_price": exit_price, + "exit_reason": reason, + "bars_held": progress["bars_held"], + "gross_r": gross_r, + "net_r_030": (pnl - REFERENCE_COST) / risk, + "net_r_065": (pnl - REALISTIC_COST) / risk, + "mfe_r": progress["mfe_r"], + "mae_r": progress["mae_r"], + } + + +def advance_outcome( + state: Dict[str, Any], bars: Iterable[Dict[str, Any]] +) -> Dict[str, Any]: + """Advance one pending outcome with overlapping closed-bar snapshots.""" + if state.get("status") != "PENDING": + return state + try: + progress = state["progress"] + setup = state["setup"] + policy = state["policy"] + last = _timestamp(progress["last_processed_bar"]) + signal_bar = _timestamp(state["signal_bar_at"]) + if last is None or signal_bar is None: + raise ValueError("invalid progress timestamp") + + buy = setup["direction"] == "BUY" + risk = float(setup["stop_distance"]) + trail_arm = float(policy["trail_arm_r"]) + trail_distance = float(policy["trail_dist_r"]) + + valid_bars = _valid_bars(bars) + has_future = any(bar["_timestamp"] > last for bar in valid_bars) + has_overlap = any(bar["_timestamp"] == last for bar in valid_bars) + if has_future and not has_overlap: + state["status"] = "UNRESOLVABLE" + state["result"] = None + state["reason"] = "stored candle snapshots lost continuity" + return state + if has_future and not _has_continuity(valid_bars, last, + state["timeframe"]): + state["status"] = "UNRESOLVABLE" + state["result"] = None + state["reason"] = "stored candle snapshots contain an internal gap" + return state + + for bar in valid_bars: + stamp = bar["_timestamp"] + if stamp <= signal_bar or stamp <= last: + continue + + if progress["entry_price"] is None: + entry = bar["open"] + initial_stop = entry - risk if buy else entry + risk + progress.update({ + "entry_at": bar["timestamp"], + "entry_price": entry, + "initial_stop": initial_stop, + "current_stop": initial_stop, + "peak": entry, + }) + + entry = float(progress["entry_price"]) + current_stop = float(progress["current_stop"]) + initial_stop = float(progress["initial_stop"]) + progress["bars_held"] += 1 + progress["last_processed_bar"] = bar["timestamp"] + last = stamp + + if buy: + mfe = (bar["high"] - entry) / risk + mae = (entry - bar["low"]) / risk + stop_hit = bar["low"] <= current_stop + else: + mfe = (entry - bar["low"]) / risk + mae = (bar["high"] - entry) / risk + stop_hit = bar["high"] >= current_stop + progress["mfe_r"] = max(float(progress["mfe_r"]), max(0.0, mfe)) + progress["mae_r"] = max(float(progress["mae_r"]), max(0.0, mae)) + + if stop_hit: + exit_price = current_stop + if (buy and bar["open"] < current_stop) or \ + (not buy and bar["open"] > current_stop): + exit_price = bar["open"] + reason = "SL" if current_stop == initial_stop else "TRAIL" + _finish(state, reason, bar["timestamp"], exit_price) + return state + + if trail_arm <= 0: + target = entry + setup["tp1_distance"] if buy \ + else entry - setup["tp1_distance"] + target_hit = bar["high"] >= target if buy else bar["low"] <= target + if target_hit: + exit_price = target + if (buy and bar["open"] > target) or \ + (not buy and bar["open"] < target): + exit_price = bar["open"] + _finish(state, "TP1", bar["timestamp"], exit_price) + return state + else: + favorable = bar["high"] if buy else bar["low"] + peak = max(float(progress["peak"]), favorable) if buy \ + else min(float(progress["peak"]), favorable) + progress["peak"] = peak + excursion = ((peak - entry) if buy else (entry - peak)) / risk + if excursion >= trail_arm: + level = peak - trail_distance * risk if buy \ + else peak + trail_distance * risk + progress["current_stop"] = max(current_stop, level) if buy \ + else min(current_stop, level) + + if progress["bars_held"] >= int(policy["max_hold_bars"]): + _finish(state, "TIMEOUT", bar["timestamp"], bar["close"]) + return state + return state + except Exception as exc: + state["status"] = "UNRESOLVABLE" + state["result"] = None + state["reason"] = f"outcome evaluator error: {type(exc).__name__}" + return state + + +def canonical_r(state: Dict[str, Any]) -> Optional[float]: + result = state.get("result") + value = result.get("net_r_065") if isinstance(result, dict) else None + return _finite(value) diff --git a/agent_panel.py b/agent_panel.py index 3bf7771..62c29da 100644 --- a/agent_panel.py +++ b/agent_panel.py @@ -25,17 +25,17 @@ 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. +- 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, 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 +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`` @@ -43,13 +43,19 @@ see backtest/research/llm_news_eval_2026-08-06.md). Sized by config 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. +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 @@ -122,9 +128,11 @@ _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. +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. @@ -146,16 +154,18 @@ and a soft jobs report -> {"stance": "support", "confidence": 75, _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". +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); keltner = dollar distance from the mid-line; -ema_trend = 0-1 alignment score; divergence and candlestick = pattern -counts. +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, @@ -166,6 +176,10 @@ 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. @@ -183,19 +197,22 @@ BUY and exhaustion 0 -> {"stance": "support", "confidence": 70, _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". +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 -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. +stop distance versus volatility, cost as a fraction of risk, trailing +geometry, and event landmines about to hit. Stance rules: -- support: the likely reward clearly justifies the risk, the stop fits +- support: the trailing opportunity 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 +- 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). @@ -203,17 +220,21 @@ Stance rules: 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 -> +"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 possible profit is more than double the possible loss, and conditions look calm."} +"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 engine's own "bullish_reasons" and "bearish_reasons", its -"warnings", and an optional "news_summary". +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, @@ -230,9 +251,11 @@ 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. +or "NEUTRAL" (genuinely split or low conviction). The risk manager and +contrarian have intentionally defensive roles; their objections are not +automatic vetoes. 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 @@ -266,6 +289,7 @@ class PanelVerdict: rationale: str takes: List[PersonaTake] = field(default_factory=list) created_ts: float = 0.0 + payload_hash: str = "" # ── Payload builders (pure functions; llm_client json-dumps sorted) ── @@ -307,49 +331,181 @@ def _brief(report: "SignalReport") -> Dict[str, Any]: } +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]: - 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, + "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] - return { + payload: Dict[str, Any] = { "brief": _brief(report), - "headlines": headlines, + "headlines": _directional_news_payload(context), "events": events, - "sentiment": round(report.news_sentiment, 3), - "news_summary": report.news_summary, } + 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]: - 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, + "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()}, @@ -368,29 +524,25 @@ def _risk_payload(report: "SignalReport") -> Dict[str, Any]: "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), - } + setup = _trade_setup_payload(report) + if setup is not None: + payload["setup"] = setup return payload -def _contrarian_payload(report: "SignalReport") -> Dict[str, Any]: +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), - "news_summary": report.news_summary, } @@ -502,15 +654,58 @@ def _validate_take(persona: str, reply: Any) -> Optional[PersonaTake]: def _ask_persona(persona: str, system: str, - payload: Dict[str, Any]) -> Optional[PersonaTake]: + 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)), + ) + + +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, ) -> Optional[PanelVerdict]: """Run 4 personas + synthesizer; None = no verdict (fail-safe). @@ -522,12 +717,8 @@ def run_panel(report: "SignalReport", 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)), - ) + 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 @@ -558,7 +749,7 @@ def run_panel(report: "SignalReport", rationale = str(reply.get("rationale", "")).strip()[:240] return PanelVerdict(verdict=verdict, confidence=confidence, rationale=rationale, takes=takes, - created_ts=time.time()) + created_ts=time.time(), payload_hash=payload_hash) # ── Cache, budget, attach ──────────────────────────────────────────── @@ -566,13 +757,14 @@ def run_panel(report: "SignalReport", _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: "OrderedDict[Tuple[str, 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). +# 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. +_last_by_payload: Dict[str, PanelVerdict] = {} +_inflight: Dict[str, 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 @@ -583,7 +775,10 @@ def reset_panel_state() -> None: global _panel_day, _panel_count, _warned_budget with _state_lock: _cache.clear() - _last_by_dir.clear() + _last_by_payload.clear() + for event in _inflight.values(): + event.set() + _inflight.clear() _panel_day = None _panel_count = 0 _warned_budget = False @@ -619,13 +814,19 @@ def _under_panel_budget(cap: int) -> bool: return True -def _write_fields(report: "SignalReport", v: PanelVerdict) -> None: +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 def _log_panel_verdict(report: "SignalReport") -> None: @@ -674,28 +875,6 @@ def attach_panel(report: "SignalReport", 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. @@ -709,17 +888,82 @@ def attach_panel(report: "SignalReport", context = None candles = _candle_context(report, settings) - verdict = run_panel(report, context, candles=candles) - if verdict is None: - return False - _write_fields(report, verdict) + 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) + ttl = float(settings.get("reuse_ttl_seconds", + DEFAULT_REUSE_TTL_SECONDS)) + inflight_key = reuse_hash if ttl > 0 else payload_hash + hit = None + cache_status = "" + leader = False + event: Optional[threading.Event] = None 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 + hit = _cache.get(exact_key) + if hit is not None: + cache_status = "EXACT_HIT" + elif ttl > 0: + prior = _last_by_payload.get(reuse_hash) + 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_hash) + 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) + 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_hash] = 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 diff --git a/main.py b/main.py index 603f3ba..28752cf 100755 --- a/main.py +++ b/main.py @@ -241,6 +241,9 @@ def report_to_dict(report: SignalReport) -> dict: "panel_confidence": round(report.panel_confidence, 1), "panel_rationale": report.panel_rationale, "panel_takes": report.panel_takes, + "panel_created_at": report.panel_created_at, + "panel_cache_status": report.panel_cache_status, + "panel_payload_hash": report.panel_payload_hash, "bullish_reasons": report.bullish_reasons, "bearish_reasons": report.bearish_reasons, "warnings": report.warnings, @@ -257,6 +260,8 @@ def report_to_dict(report: SignalReport) -> dict: "risk_pips": round(ts.risk_pips, 1), "reward_pips": round(ts.reward_pips, 1), "strategy": ts.strategy, + "setup_pattern": ts.setup_pattern, + "round_trip_cost": ts.round_trip_cost, "trail_arm_r": ts.trail_arm_r, "trail_dist_r": ts.trail_dist_r, } diff --git a/signal_aggregator.py b/signal_aggregator.py index 54370fc..5c60c6f 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -56,8 +56,12 @@ class TradeSetup: risk_pips: float reward_pips: float pip_value: float - # Entry archetype ("GENERIC", "XAU_PB") — selects trail params. + # Entry archetype — selects trail params and explains the setup to + # advisory consumers. setup_pattern is populated only by XAU_CP. strategy: str = "GENERIC" + setup_pattern: Optional[str] = None + # Strategy cost assumption, not a broker/data-provider quote. + round_trip_cost: Optional[float] = None # Trailing exit params stamped from the PRODUCING config, so per-TF # factory overrides (e.g. M15 arm 0.70, M5 dist 0.85) reach every # consumer. None = legacy/unstamped setup — formatters fall back to @@ -97,6 +101,9 @@ class SignalReport: panel_confidence: float = 0.0 panel_rationale: str = "" panel_takes: List[Dict] = field(default_factory=list) + panel_created_at: str = "" + panel_cache_status: str = "" + panel_payload_hash: str = "" final_signal: SignalStrength = SignalStrength.NEUTRAL final_score: float = 0.0 @@ -655,6 +662,7 @@ class SignalAggregator: reward_pips=1.5 * sl_dist / pip_value, pip_value=pip_value, strategy="XAU_PB", + round_trip_cost=self.config.signals.round_trip_cost, trail_arm_r=self.config.signals.pb_trail_arm_r, trail_dist_r=self.config.signals.pb_trail_dist_r, ) @@ -792,6 +800,8 @@ class SignalAggregator: reward_pips=height / pip_value, pip_value=pip_value, strategy="XAU_CP", + setup_pattern=pat["pattern"], + round_trip_cost=self.config.signals.round_trip_cost, trail_arm_r=self.config.signals.trail_arm_r, trail_dist_r=self.config.signals.trail_dist_r, ) @@ -807,8 +817,74 @@ class SignalAggregator: f"height {height / atr:.1f} ATR") target = report.bullish_reasons if is_buy else report.bearish_reasons target.insert(0, marker) + self._clean_accepted_cp_warnings(report) _reject("accepted") + @staticmethod + def _clean_accepted_cp_warnings(report: SignalReport) -> None: + """Drop superseded composite-path warnings after XAU_CP accepts. + + This is metadata-only. Pattern detection, gates, confidence, + exhaustion state, setup geometry, and signal selection are already + final when this runs. + """ + stale_prefixes = ( + "SPREAD GATE:", + "EXHAUSTION BLOCK", + "EXHAUSTED TREND FILTER:", + "MTF OPPOSITION:", + "SELL FILTER: quality", + "SELL FILTER: volatility", + "SELL FILTER: regime", + "SELL FILTER: ", + "M5 FILTER: |score|", + "M5 FILTER: no MTF agreement", + "M5 FILTER: no confirmation TF", + "M5 FILTER: SELL quality", + "M5 FILTER: SELL blocked", + ) + extension_prefixes = ( + "TREND EXHAUSTION", + "NO PULLBACK", + "FAR FROM EMA200", + "Extended — RSI", + "Weak pullback", + ) + filtered: List[str] = [] + for warning in report.warnings: + if warning.startswith(stale_prefixes): + continue + if (warning.startswith("REGIME FILTER: '") + and warning.endswith("blocked for XAU")): + continue + if (warning.startswith("M5 FILTER: '") + and "volatility" in warning): + continue + filtered.append(warning) + + extension_items = [ + (index, warning) + for index, warning in enumerate(filtered) + if warning.startswith(extension_prefixes) + ] + if not extension_items: + report.warnings[:] = filtered + return + first_index = extension_items[0][0] + priority = {prefix: rank for rank, prefix in enumerate(extension_prefixes)} + strongest = min( + (warning for _, warning in extension_items), + key=lambda warning: next( + rank for prefix, rank in priority.items() + if warning.startswith(prefix)), + ) + kept = [ + warning for warning in filtered + if not warning.startswith(extension_prefixes) + ] + kept.insert(first_index, strongest) + report.warnings[:] = kept + # ── Public: bar-replayable entry point ───────────────────────── def analyze_from_dataframes( @@ -1195,6 +1271,8 @@ class SignalAggregator: risk_pips=round(sl_distance / pip_value, 1), reward_pips=round(tp_distance / pip_value, 1), pip_value=pip_value, + strategy="XAU_ATR", + round_trip_cost=self.config.signals.round_trip_cost, trail_arm_r=self.config.signals.trail_arm_r, trail_dist_r=self.config.signals.trail_dist_r, ) @@ -1609,6 +1687,7 @@ class SignalAggregator: risk_pips=risk_pips, reward_pips=reward_pips, pip_value=pip_value, + round_trip_cost=self.config.signals.round_trip_cost, trail_arm_r=self.config.signals.trail_arm_r, trail_dist_r=self.config.signals.trail_dist_r, ) diff --git a/signal_log.py b/signal_log.py index 38b5dc3..d7d0ffc 100644 --- a/signal_log.py +++ b/signal_log.py @@ -13,9 +13,10 @@ never writes here (live == backtest parity — CLAUDE.md rule 4). Panel fields arrive POST-HOC: the row is inserted during analyze_pair (before consumers attach the advisory panel), so log_signal stashes the rowid on the report and agent_panel.attach_panel calls -log_panel_update() to fill panel_verdict / panel_confidence / -panel_rationale / panel_takes_json on that same row. A report without a -rowid (backtest replay, logging disabled) is a silent no-op. +log_panel_update() to fill the verdict, takes, creation/cache provenance, +payload hash, and advisory-outcome seed on that same row. Later closed-candle +snapshots advance the hypothetical next-open/trailing outcome. A report +without a rowid (backtest replay, logging disabled) is a silent no-op. Deployment settings come from the config.json "signal_log" block via app_config (env vars SIGNAL_LOG_ENABLED / SIGNAL_LOG_PATH win): @@ -38,6 +39,12 @@ from datetime import datetime, timezone from typing import Any, Dict, List, Optional import app_config +from advisory_outcome import ( + advance_outcome, + canonical_r, + initial_outcome, + timeframe_value, +) logger = logging.getLogger(__name__) @@ -73,7 +80,13 @@ CREATE TABLE IF NOT EXISTS signals ( panel_verdict TEXT, panel_confidence REAL, panel_rationale TEXT, - panel_takes_json TEXT + panel_takes_json TEXT, + panel_created_at TEXT, + panel_cache_status TEXT, + panel_payload_hash TEXT, + advisory_outcome_status TEXT, + advisory_r_multiple REAL, + advisory_outcome_json TEXT ); CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts ON signals (symbol, timeframe, logged_at); @@ -101,6 +114,12 @@ _ADDITIVE_SIGNAL_COLS = ( ("panel_takes_json", "TEXT"), ("indicators_json", "TEXT"), ("candles_json", "TEXT"), + ("panel_created_at", "TEXT"), + ("panel_cache_status", "TEXT"), + ("panel_payload_hash", "TEXT"), + ("advisory_outcome_status", "TEXT"), + ("advisory_r_multiple", "REAL"), + ("advisory_outcome_json", "TEXT"), ) _ADDITIVE_OUTLOOK_COLS = ( @@ -327,6 +346,9 @@ def _report_row( "risk_pips": _finite(getattr(setup, "risk_pips", None)), "reward_pips": _finite(getattr(setup, "reward_pips", None)), "strategy": getattr(setup, "strategy", None), + "setup_pattern": getattr(setup, "setup_pattern", None), + "round_trip_cost": _finite( + getattr(setup, "round_trip_cost", None)), "trail_arm_r": _finite(getattr(setup, "trail_arm_r", None)), "trail_dist_r": _finite(getattr(setup, "trail_dist_r", None)), }) @@ -403,8 +425,17 @@ class SignalLogger: existing = {r[1] for r in self._conn.execute("PRAGMA table_info(signals)")} for name, col_type in _ADDITIVE_SIGNAL_COLS: if name not in existing: - self._conn.execute( - f"ALTER TABLE signals ADD COLUMN {name} {col_type}") + try: + self._conn.execute( + f"ALTER TABLE signals ADD COLUMN {name} {col_type}") + except sqlite3.OperationalError as exc: + # Another dashboard/bot process may have migrated the + # shared database after our PRAGMA snapshot. + if "duplicate column name" not in str(exc).lower(): + raise + self._conn.execute( + "CREATE INDEX IF NOT EXISTS idx_signals_advisory_pending" + " ON signals (symbol, advisory_outcome_status, id)") self._conn.commit() def _migrate_outlook_columns(self) -> None: @@ -414,8 +445,12 @@ class SignalLogger: } for name, col_type in _ADDITIVE_OUTLOOK_COLS: if name not in existing: - self._conn.execute( - f"ALTER TABLE outlooks ADD COLUMN {name} {col_type}") + try: + self._conn.execute( + f"ALTER TABLE outlooks ADD COLUMN {name} {col_type}") + except sqlite3.OperationalError as exc: + if "duplicate column name" not in str(exc).lower(): + raise self._conn.commit() def log( @@ -431,7 +466,13 @@ class SignalLogger: with self._lock: cur = self._conn.execute(_INSERT, row) self._conn.commit() - return int(cur.lastrowid) + rowid = int(cur.lastrowid) + try: + self.resolve_pending(str(row[2]), row[25]) + except Exception: + logger.warning("signal_log: advisory outcome resolver failed", + exc_info=True) + return rowid def update_panel( self, @@ -440,15 +481,89 @@ class SignalLogger: confidence: Optional[float], rationale: str, takes_json: str, + *, + created_at: Optional[str] = None, + cache_status: Optional[str] = None, + payload_hash: Optional[str] = None, ) -> None: - """Fill the panel columns on an already-inserted scan row.""" + """Fill panel provenance and initialize its advisory outcome.""" with self._lock: + row = self._conn.execute( + "SELECT setup_json, indicators_json, timeframe," + " advisory_outcome_status FROM signals WHERE id=?", + (rowid,), + ).fetchone() + if row is None: + return + advisory_status = row[3] + advisory_json = None + if advisory_status is None: + setup = json.loads(row[0]) if row[0] else None + indicators = json.loads(row[1]) if row[1] else None + advisory_status, state = initial_outcome( + setup, indicators, row[2]) + advisory_json = _dumps(state) self._conn.execute( "UPDATE signals SET panel_verdict=?, panel_confidence=?," - " panel_rationale=?, panel_takes_json=? WHERE id=?", - (verdict, confidence, rationale, takes_json, rowid)) + " panel_rationale=?, panel_takes_json=?," + " panel_created_at=COALESCE(?, panel_created_at)," + " panel_cache_status=COALESCE(?, panel_cache_status)," + " panel_payload_hash=COALESCE(?, panel_payload_hash)," + " advisory_outcome_status=COALESCE(advisory_outcome_status, ?)," + " advisory_outcome_json=COALESCE(advisory_outcome_json, ?)" + " WHERE id=?", + (verdict, confidence, rationale, takes_json, + created_at, cache_status, payload_hash, + advisory_status, advisory_json, rowid)) self._conn.commit() + def resolve_pending( + self, symbol: str, candles_json: Optional[str], limit: int = 100 + ) -> int: + """Advance pending advisory outcomes from a new candle snapshot.""" + if not candles_json: + return 0 + snapshots = json.loads(candles_json) + if not isinstance(snapshots, dict): + return 0 + changed = 0 + with self._lock: + self._conn.execute("BEGIN IMMEDIATE") + try: + rows = self._conn.execute( + "SELECT id, timeframe, advisory_outcome_json FROM signals" + " WHERE symbol=? AND advisory_outcome_status='PENDING'" + " ORDER BY id", + (symbol,), + ).fetchall() + eligible = 0 + for rowid, timeframe, state_json in rows: + bars = timeframe_value(snapshots, timeframe) + if not bars or not state_json: + continue + if eligible >= max(1, int(limit)): + break + eligible += 1 + state = json.loads(state_json) + before = _dumps(state) + state = advance_outcome(state, bars) + after = _dumps(state) + if after == before: + continue + status = state.get("status", "UNRESOLVABLE") + self._conn.execute( + "UPDATE signals SET advisory_outcome_status=?," + " advisory_r_multiple=?, advisory_outcome_json=?" + " WHERE id=? AND advisory_outcome_status='PENDING'", + (status, canonical_r(state), after, rowid), + ) + changed += 1 + self._conn.commit() + except Exception: + self._conn.rollback() + raise + return changed + def log_outlook(self, outlook: Any, surface: str, payload: Any = None) -> int: """Insert one advisory market outlook; returns its rowid.""" @@ -540,7 +655,10 @@ def log_panel_update(report: Any) -> None: int(rowid), verdict, _finite(getattr(report, "panel_confidence", None)), getattr(report, "panel_rationale", "") or "", - _dumps(getattr(report, "panel_takes", None) or [])) + _dumps(getattr(report, "panel_takes", None) or []), + created_at=getattr(report, "panel_created_at", "") or None, + cache_status=getattr(report, "panel_cache_status", "") or None, + payload_hash=getattr(report, "panel_payload_hash", "") or None) except Exception: logger.warning("signal_log: failed to persist panel verdict", exc_info=True) diff --git a/test_all.py b/test_all.py index 54b9719..871c3fc 100644 --- a/test_all.py +++ b/test_all.py @@ -1213,6 +1213,7 @@ def _(): assert ts.risk_reward == 2.0 assert ts.entry_price > ts.stop_loss assert ts.trail_arm_r is None and ts.trail_dist_r is None # unstamped default + assert ts.setup_pattern is None and ts.round_trip_cost is None @test("trade setups carry trail params from the producing config") @@ -1228,6 +1229,8 @@ def _(): report.current_price = 3300.0 ts = agg._generate_xau_atr_trade_setup(report, 10.0) assert ts is not None, "A+ setup with ATR 10 should generate" + assert ts.strategy == "XAU_ATR" + assert ts.round_trip_cost == config.signals.round_trip_cost assert ts.trail_arm_r == 0.70, "setup must carry the producing config's arm" assert ts.trail_dist_r == 0.85, "setup must carry the producing config's dist" @@ -2768,6 +2771,7 @@ def _(): ts = rep.trade_setup assert ts is not None, "pattern setup not generated" assert ts.strategy == "XAU_CP" and ts.direction == "BUY" + assert ts.setup_pattern == "double_bottom" assert ts.stop_loss < ts.entry_price assert abs((ts.take_profit_1 - ts.entry_price) - 19.0) < 1e-6, \ "TP1 must be the measured move" @@ -2776,6 +2780,31 @@ def _(): "tag marker must sit at reasons[0]" +@test("CP gate: accepted setup removes stale vetoes and deduplicates extension context") +def _(): + cfg = _cp_config() + df = make_cp_df(double_bottom_closes()) + rep = _cp_report(df, cfg) + rep.warnings = [ + "SPREAD GATE: old core target failed", + "EXHAUSTED TREND FILTER: C-grade blocked in exhausted trend", + "M5 FILTER: no MTF agreement — M5 signals require confirmation", + "FAR FROM EMA200 (4.0%) — over-extended", + "TREND EXHAUSTION — price extended >3.5 ATR past swing low", + "NO PULLBACK — price 3.0 ATR from EMA21", + "Extended — RSI 78 in uptrend", + "EVENT RISK: CPI in 20m", + "NEAR EMA200 (0.3%) — chop risk", + ] + SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10) + assert rep.trade_setup is not None + assert rep.warnings == [ + "TREND EXHAUSTION — price extended >3.5 ATR past swing low", + "EVENT RISK: CPI in 20m", + "NEAR EMA200 (0.3%) — chop risk", + ], rep.warnings + + @test("CP gate: MTF opposition blocks the pattern trade") def _(): cfg = _cp_config() @@ -2828,8 +2857,11 @@ def _(): cfg.signals.min_tp_cost_multiple = 1000.0 # $300 minimum target df = make_cp_df(double_bottom_closes()) rep = _cp_report(df, cfg) + rep.warnings = ["SPREAD GATE: prior warning", "NO PULLBACK — context"] + before = list(rep.warnings) SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10) assert rep.trade_setup is None + assert rep.warnings == before, "rejected CP must not rewrite diagnostics" @test("CP pipeline: fires end-to-end when core generators pass") @@ -6345,6 +6377,7 @@ class _M21cBot: def _m21c_reset(): _tb21._scan_cooldowns.clear() _tb21._pending_breach.clear() + _tb21._alert_locks.clear() _tb21._last_alert_fp.clear() _tb21._last_alert_ts.clear() _tb21._last_alert_setup.clear() @@ -6858,11 +6891,12 @@ def _(): assert pbase._is_forex_weekend(T("2026-08-01 03:00")) is True # Saturday -@test("weekend calendar: all three copies agree hourly across DST transitions") +@test("weekend calendar: all four copies agree hourly across DST transitions") def _(): import providers.base as pbase import signal_aggregator as sa_mod import gpu_backtest as _gb + import advisory_outcome as ao cfg = PlatformConfig() cfg.xau_strategy.weekend_filter = True cfg.xau_strategy.allowed_weekdays = [] # isolate the weekend gate @@ -6872,7 +6906,9 @@ def _(): a = pbase._is_forex_weekend(ts) b = sa_mod._is_forex_weekend(ts) c = _gb._is_xau_blocked_by_calendar(ts, cfg) == "WEEKEND_FILTER" - assert a == b == c, f"{ts}: base={a} aggregator={b} backtest={c}" + d = ao._is_forex_weekend(ts.to_pydatetime()) + assert a == b == c == d, \ + f"{ts}: base={a} aggregator={b} backtest={c} advisory={d}" @test("FB2_EMA_LO: garbage value degrades to tuned default, scan survives") @@ -7077,12 +7113,13 @@ def _panel_report(quality="A", signal=SignalStrength.BUY): config.primary_timeframe = Timeframe.H1 config.confirmation_timeframes = [Timeframe.M15, Timeframe.H4] agg = SignalAggregator(config) - dfs = {Timeframe.H1: make_ohlcv(300, seed=42), - Timeframe.M15: make_ohlcv(300, seed=43), - Timeframe.H4: make_ohlcv(300, seed=44)} + dfs = {Timeframe.H1: make_xau_ohlcv(300, seed=42), + Timeframe.M15: make_xau_ohlcv(300, seed=43), + Timeframe.H4: make_xau_ohlcv(300, seed=44)} report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False) report.final_signal = signal report.signal_quality = quality + report.trade_setup = None return report @@ -7090,19 +7127,32 @@ def _panel_report(quality="A", signal=SignalStrength.BUY): def _(): report = _panel_report() now = _dt_m19.datetime.now(_dt_m19.timezone.utc) + directional = [NewsSentiment( + headline=f"h{i}", source="rss", timestamp=now, + currencies_mentioned=["USD"], sentiment_score=0.2, keywords=[]) + for i in range(15)] + neutral = NewsSentiment( + headline="neutral", source="rss", timestamp=now, + currencies_mentioned=["XAU"], sentiment_score=0.05, keywords=[]) + unrelated = NewsSentiment( + headline="unrelated", source="rss", timestamp=now, + currencies_mentioned=["EUR"], sentiment_score=0.8, keywords=[]) ctx = FundamentalContext( - news_headlines=[NewsSentiment( - headline=f"h{i}", source="rss", timestamp=now, - currencies_mentioned=["USD"], sentiment_score=0.1, keywords=[]) - for i in range(15)], + news_headlines=[neutral, unrelated] + directional, events_today=[EconomicEvent( time=now, currency="USD", impact="high", event=f"e{i}", forecast="1.0", actual="2.0") for i in range(12)], ) macro = ap._macro_payload(report, ctx) assert len(macro["headlines"]) == 10, "headline cap" + assert all(h["gold_influence"] == "bearish" for h in macro["headlines"]) + assert all(h["headline"] not in ("neutral", "unrelated") + for h in macro["headlines"]) + assert "sentiment" not in macro, "neutral aggregate sentiment must be omitted" + assert "news_summary" not in macro, "unfiltered summary must not bypass filtering" assert len(macro["events"]) == 8, "event cap" assert macro["events"][0]["released"] is True + assert macro["events"][0]["time"] and macro["events"][0]["surprise"] == "positive" assert ap._macro_payload(report, None)["headlines"] == [] tech = ap._technical_payload(report) @@ -7113,19 +7163,38 @@ def _(): assert not hidden, f"private/weight-0 leaked: {hidden}" assert tech["brief"]["direction"] == "BUY" assert tech["brief"]["timeframe"] == report.primary_analysis.timeframe + assert tech["strategy"]["code"] == "NO_SETUP" assert "setup" not in ap._risk_payload(report), "no setup when None" report.trade_setup = TradeSetup( pair="XAU/USD", display_name="XAU/USD", direction="SELL", entry_price=2000.0, stop_loss=2010.0, take_profit_1=1985.0, take_profit_2=1970.0, risk_reward=1.5, risk_pips=100.0, - reward_pips=150.0, pip_value=0.1) + reward_pips=150.0, pip_value=0.1, strategy="XAU_CP", + setup_pattern="double_top", round_trip_cost=0.30, + trail_arm_r=0.6, trail_dist_r=0.85) + tech = ap._technical_payload(report) + assert tech["strategy"]["code"] == "XAU_CP" + assert tech["strategy"]["setup_pattern"] == "double_top" risk = ap._risk_payload(report) - assert risk["setup"]["risk_reward"] == 1.5 + assert risk["setup"]["risk_reward_to_advisory_tp1"] == 1.5 + assert risk["setup"]["strategy"]["setup_pattern"] == "double_top" + assert risk["setup"]["trailing"] == { + "enabled": True, "arm_r": 0.6, "distance_r": 0.85, + "tp1_tp2_are_advisory": True} + assert risk["setup"]["round_trip_cost"] == 0.3 + assert risk["setup"]["atr"] > 0 and risk["setup"]["stop_distance_atr"] > 0 + assert "spread" not in risk, "candle range must not be mislabeled as spread" assert risk["brief"]["direction"] == "SELL", "setup direction wins" - contra = ap._contrarian_payload(report) + contra = ap._contrarian_payload(report, ctx) assert len(contra["bullish_reasons"]) <= 3 + assert contra["setup"]["strategy"]["code"] == "XAU_CP" + assert contra["market_snapshot"]["basis"] == "latest_closed_candle" + assert contra["market_snapshot"]["bar_timestamp"] == report.timestamp + assert contra["indicators"] == tech["indicators"] + assert len(contra["news"]) == 10 + assert "news_summary" not in contra for payload in (macro, tech, risk, contra): json.dumps(payload, sort_keys=True, allow_nan=False) @@ -7180,6 +7249,7 @@ def _(): lc._client = stub v = ap.run_panel(_panel_report()) assert v.verdict == "OPPOSE" and v.confidence == 72.4 + assert v.payload_hash.startswith("sha256:") assert len(v.rationale) == 240, "rationale truncated" assert [t.persona for t in v.takes] == \ ["macro", "technical", "risk", "contrarian"] @@ -7216,6 +7286,8 @@ def _(): assert ap.attach_panel(report, _PANEL_ON, context=FundamentalContext()) is True assert report.panel_verdict == "AGREE" + assert report.panel_cache_status == "MISS" + assert report.panel_created_at and report.panel_payload_hash.startswith("sha256:") assert len(stub.calls) == 5 # same bar again → exact-key hit @@ -7223,6 +7295,9 @@ def _(): assert ap.attach_panel(report2, _PANEL_ON, context=FundamentalContext()) is True assert report2.panel_verdict == "AGREE" and len(stub.calls) == 5 + assert report2.panel_cache_status == "EXACT_HIT" + assert report2.panel_payload_hash == report.panel_payload_hash + assert report2.panel_created_at == report.panel_created_at # newer bar, same symbol/tf/direction → reuse-TTL hit report3 = _panel_report() @@ -7230,13 +7305,24 @@ def _(): assert ap.attach_panel(report3, _PANEL_ON, context=FundamentalContext()) is True assert len(stub.calls) == 5, "persisting signal must not re-spend" + assert report3.panel_cache_status == "TTL_REUSE" + assert report3.panel_payload_hash == report.panel_payload_hash + + # Changed evidence must not inherit a direction-only cached verdict. + changed = _panel_report() + changed.timestamp = "2099-01-01 01:05:00+00:00" + changed.current_price += 1.0 + assert ap.attach_panel(changed, _PANEL_ON, + context=FundamentalContext()) is True + assert len(stub.calls) == 10 + assert changed.panel_cache_status == "MISS" # TTL 0 → the reuse layer is off; a new bar re-runs the panel report4 = _panel_report() report4.timestamp = "2099-01-01 02:00:00+00:00" assert ap.attach_panel(report4, {**_PANEL_ON, "reuse_ttl_seconds": 0}, context=FundamentalContext()) is True - assert len(stub.calls) == 10 + assert len(stub.calls) == 15 finally: lc._client = None ap._dp = None @@ -7271,6 +7357,35 @@ def _(): ap.reset_panel_state() +@test("agent_panel: identical concurrent attaches are single-flight") +def _(): + class _SlowPanelStub(_PanelStub): + def chat_json(self, system_prompt, user_payload): + time.sleep(0.03) + return super().chat_json(system_prompt, user_payload) + try: + stub = _SlowPanelStub(dict(_ALL_OK)) + lc._client = stub + ap._dp = _NullDP() + reports = [_panel_report(), _panel_report()] + results = [] + threads = [threading.Thread( + target=lambda report=r: results.append(ap.attach_panel( + report, _PANEL_ON, context=FundamentalContext()))) + for r in reports] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + assert results == [True, True] + assert len(stub.calls) == 5, "one evidence payload must spend one panel" + assert {r.panel_cache_status for r in reports} == {"MISS", "EXACT_HIT"} + finally: + lc._client = None + ap._dp = None + ap.reset_panel_state() + + @test("agent_panel: enabled/surface/min_quality gates; force bypasses") def _(): try: @@ -7345,6 +7460,9 @@ def _(): assert report.panel_confidence == 0.0 assert report.panel_rationale == "" assert report.panel_takes == [] + assert report.panel_created_at == "" + assert report.panel_cache_status == "" + assert report.panel_payload_hash == "" assert stub.calls == [], \ "the replayable core must never invoke the panel" finally: @@ -7352,6 +7470,31 @@ def _(): ap.reset_panel_state() +@test("agent_panel: attach remains advisory-only") +def _(): + try: + lc._client = _PanelStub(dict(_ALL_OK)) + ap._dp = _NullDP() + report = _panel_report() + before = ( + report.final_signal, report.final_score, report.final_confidence, + report.signal_quality, report.action_recommendation, + report.trade_setup, list(report.warnings), + ) + assert ap.attach_panel(report, _PANEL_ON, + context=FundamentalContext()) is True + after = ( + report.final_signal, report.final_score, report.final_confidence, + report.signal_quality, report.action_recommendation, + report.trade_setup, list(report.warnings), + ) + assert after == before, "panel must mutate only advisory panel fields" + finally: + lc._client = None + ap._dp = None + ap.reset_panel_state() + + @test("report_to_dict carries panel keys → dashboard snapshot propagation") def _(): from main import report_to_dict as _rtd @@ -7482,6 +7625,11 @@ def _(): ap._CONTRARIAN_SYSTEM): assert "candles" not in by_prompt[sys_p], \ "candles are technical-persona evidence only" + contra = by_prompt[ap._CONTRARIAN_SYSTEM] + assert contra["setup"] is None + assert contra["market_snapshot"]["basis"] == "latest_closed_candle" + assert contra["indicators"] + assert "news" in contra finally: lc._client = None ap._dp = None @@ -7494,6 +7642,11 @@ def _(): assert "max 15 words" in ap._TAKE_CONTRACT assert "max 30 words" in ap._SYNTH_SYSTEM assert "ATR units" in ap._TECH_SYSTEM + assert "setup_pattern" in ap._TECH_SYSTEM + assert "tp1_tp2_are_advisory" not in ap._RISK_SYSTEM # prose, not wire key + assert "reference levels" in ap._RISK_SYSTEM + assert "gold_influence" in ap._MACRO_SYSTEM + assert "market_snapshot" in ap._CONTRARIAN_SYSTEM for prompt in (ap._MACRO_SYSTEM, ap._TECH_SYSTEM, ap._RISK_SYSTEM, ap._CONTRARIAN_SYSTEM): assert ap._TAKE_CONTRACT in prompt, \ @@ -7537,6 +7690,7 @@ def _make_log_report(): entry_price=2450.50, stop_loss=2445.00, take_profit_1=2458.00, take_profit_2=2468.00, risk_reward=1.6, risk_pips=5.5, reward_pips=8.8, pip_value=1.0, strategy="XAU_PB", + setup_pattern=None, round_trip_cost=0.30, trail_arm_r=0.70, trail_dist_r=0.85, ) report.primary_analysis = TechnicalAnalysis( @@ -7625,6 +7779,8 @@ def _(): assert mtf["scores"][Timeframe.M5.value] == 0.6 setup = json.loads(row["setup_json"]) assert setup["strategy"] == "XAU_PB" + assert setup["setup_pattern"] is None + assert setup["round_trip_cost"] == 0.30 assert setup["entry"] == 2450.50 and setup["trail_arm_r"] == 0.70 reasons = json.loads(row["reasons_json"]) assert "MTF agree" in reasons["bullish"] @@ -7670,21 +7826,32 @@ def _(): report.panel_verdict = "OPPOSE" report.panel_confidence = 64.0 report.panel_rationale = "risk too thin" + report.panel_created_at = "2026-08-01T12:00:03+00:00" + report.panel_cache_status = "MISS" + report.panel_payload_hash = "sha256:" + "a" * 64 report.panel_takes = [{"persona": "risk", "stance": "oppose", "confidence": 70.0, "note": "n"}] lg.update_panel(rowid, report.panel_verdict, 64.0, report.panel_rationale, - slog._dumps(report.panel_takes)) + slog._dumps(report.panel_takes), + created_at=report.panel_created_at, + cache_status=report.panel_cache_status, + payload_hash=report.panel_payload_hash) lg.close() conn = sqlite3.connect(path) row = conn.execute( "SELECT panel_verdict, panel_confidence, panel_rationale," - " panel_takes_json FROM signals WHERE id=?", + " panel_takes_json, panel_created_at, panel_cache_status," + " panel_payload_hash, advisory_outcome_status" + " FROM signals WHERE id=?", (rowid,)).fetchone() conn.close() assert row[0] == "OPPOSE" and row[1] == 64.0 assert row[2] == "risk too thin" assert json.loads(row[3])[0]["persona"] == "risk" + assert row[4] == report.panel_created_at and row[5] == "MISS" + assert row[6] == report.panel_payload_hash + assert row[7] == "PENDING" finally: slog.reset_loggers_for_tests() os.remove(path) @@ -7721,15 +7888,21 @@ def _(): report.panel_confidence = 81.0 report.panel_rationale = "all aligned" report.panel_takes = [] + report.panel_created_at = "2026-08-01T12:00:05+00:00" + report.panel_cache_status = "EXACT_HIT" + report.panel_payload_hash = "sha256:" + "b" * 64 slog.log_panel_update(report) conn = sqlite3.connect(path) row = conn.execute( "SELECT panel_verdict, panel_confidence, panel_rationale," - " panel_takes_json FROM signals").fetchone() + " panel_takes_json, panel_created_at, panel_cache_status," + " panel_payload_hash, advisory_outcome_status FROM signals").fetchone() conn.close() assert row[0] == "AGREE" and row[1] == 81.0 assert row[2] == "all aligned" assert json.loads(row[3]) == [] + assert row[4] == report.panel_created_at and row[5] == "EXACT_HIT" + assert row[6] == report.panel_payload_hash and row[7] == "PENDING" # disabled → no-op, never raises os.environ["SIGNAL_LOG_ENABLED"] = "false" @@ -7765,6 +7938,174 @@ def _(): os.remove(path) +@test("advisory outcome: next-open trailing ignores TP1 and ratchets at close") +def _(): + from advisory_outcome import advance_outcome, initial_outcome + setup = { + "direction": "BUY", "entry": 100.0, "stop_loss": 90.0, + "take_profit_1": 105.0, "strategy": "GENERIC", + "setup_pattern": None, "trail_arm_r": 0.6, + "trail_dist_r": 0.75, + } + indicators = {"5min": {"timestamp": "2026-08-03T10:00:00+00:00"}} + status, state = initial_outcome(setup, indicators, "M5") + assert status == "PENDING" + state = advance_outcome(state, [{ + "timestamp": "2026-08-03T10:00:00+00:00", + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }, { + "timestamp": "2026-08-03T10:05:00+00:00", + "open": 101.0, "high": 112.0, "low": 100.0, "close": 110.0, + }]) + assert state["status"] == "PENDING", "advisory TP1 is not a hard exit" + assert state["progress"]["entry_price"] == 101.0 + assert state["progress"]["initial_stop"] == 91.0 + assert state["progress"]["current_stop"] == 104.5 + state = advance_outcome(state, [{ + "timestamp": "2026-08-03T10:05:00+00:00", + "open": 101.0, "high": 112.0, "low": 100.0, "close": 110.0, + }, { + "timestamp": "2026-08-03T10:10:00+00:00", + "open": 108.0, "high": 109.0, "low": 104.0, "close": 105.0, + }]) + assert state["status"] == "RESOLVED" + result = state["result"] + assert result["exit_reason"] == "TRAIL" and result["bars_held"] == 2 + assert abs(result["net_r_065"] - 0.285) < 1e-12 + + _, broken = initial_outcome(setup, indicators, "M5") + broken = advance_outcome(broken, [{ + "timestamp": "2026-08-03T10:10:00+00:00", + "open": 108.0, "high": 109.0, "low": 104.0, "close": 105.0, + }]) + assert broken["status"] == "UNRESOLVABLE" + assert "continuity" in broken["reason"] + + _, gapped = initial_outcome(setup, indicators, "M5") + gapped = advance_outcome(gapped, [{ + "timestamp": "2026-08-03T10:00:00+00:00", + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }, { + "timestamp": "2026-08-03T10:10:00+00:00", + "open": 101.0, "high": 102.0, "low": 100.0, "close": 101.0, + }]) + assert gapped["status"] == "UNRESOLVABLE" + assert "internal gap" in gapped["reason"] + + weekend_indicators = { + "5min": {"timestamp": "2026-08-07T20:55:00+00:00"}} + _, weekend = initial_outcome(setup, weekend_indicators, "M5") + weekend = advance_outcome(weekend, [{ + "timestamp": "2026-08-07T20:55:00+00:00", + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }, { + "timestamp": "2026-08-09T22:00:00+00:00", + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }]) + assert weekend["status"] == "PENDING", \ + "the scheduled FX weekend closure is not a data gap" + + maintenance_indicators = { + "5min": {"timestamp": "2026-08-03T20:55:00+00:00"}} + _, maintenance = initial_outcome(setup, maintenance_indicators, "M5") + maintenance = advance_outcome(maintenance, [{ + "timestamp": "2026-08-03T20:55:00+00:00", + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }, { + "timestamp": "2026-08-03T22:00:00+00:00", + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }]) + assert maintenance["status"] == "PENDING", \ + "Oanda's scheduled daily XAU closure is not a data gap" + + for timeframe, normalized, signal_at, next_at in ( + ("M30", "30min", "2026-08-03T10:00:00+00:00", + "2026-08-03T10:30:00+00:00"), + ("D1", "1day", "2026-08-03T00:00:00+00:00", + "2026-08-04T00:00:00+00:00"), + ): + stored_key = "1d" if timeframe == "D1" else normalized + alias_indicators = {stored_key: {"timestamp": signal_at}} + _, alias_state = initial_outcome(setup, alias_indicators, timeframe) + alias_state = advance_outcome(alias_state, [{ + "timestamp": signal_at, + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }, { + "timestamp": next_at, + "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0, + }]) + assert alias_state["status"] == "PENDING", timeframe + + +@test("signal_log: pending advisory outcome resolves from later snapshots") +def _(): + import sqlite3 + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + lg = slog.SignalLogger(path) + report = _make_log_report() + rowid = lg.log(report, source="test", timeframe="M5") + lg.update_panel( + rowid, "OPPOSE", 70.0, "test", "[]", + created_at="2026-08-01T12:00:03+00:00", + cache_status="MISS", payload_hash="sha256:" + "c" * 64, + ) + bars = slog._dumps({"5min": [{ + "timestamp": "2026-08-01T12:00:00+00:00", + "open": 2450.0, "high": 2451.0, "low": 2449.0, + "close": 2450.5, "volume": 1.0, + }, { + "timestamp": "2026-08-01T12:05:00+00:00", + "open": 2450.5, "high": 2451.0, "low": 2444.0, + "close": 2445.0, "volume": 1.0, + }]}) + assert lg.resolve_pending("XAU/USD", bars) == 1 + row = lg._conn.execute( + "SELECT advisory_outcome_status, advisory_r_multiple," + " advisory_outcome_json FROM signals WHERE id=?", (rowid,) + ).fetchone() + assert row[0] == "RESOLVED" and row[1] < -1.0 + outcome = json.loads(row[2]) + assert outcome["result"]["exit_reason"] == "SL" + assert outcome["model"] == "stored-closed-mid-next-open-trail-v1" + # Overlapping snapshots are idempotent after resolution. + assert lg.resolve_pending("XAU/USD", bars) == 0 + + d1_report = _make_log_report() + d1_report.timestamp = "2026-08-03T00:00:00+00:00" + d1_report.primary_analysis.timeframe = Timeframe.D1.value + d1_report.primary_analysis.timestamp = d1_report.timestamp + d1_rowid = lg.log(d1_report, source="test", timeframe="D1") + lg.update_panel( + d1_rowid, "AGREE", 70.0, "test", "[]", + created_at="2026-08-03T00:00:03+00:00", + cache_status="MISS", payload_hash="sha256:" + "d" * 64, + ) + d1_bars = slog._dumps({"1d": [{ + "timestamp": "2026-08-03T00:00:00+00:00", + "open": 2450.0, "high": 2451.0, "low": 2449.0, + "close": 2450.5, "volume": 1.0, + }, { + "timestamp": "2026-08-04T00:00:00+00:00", + "open": 2450.5, "high": 2451.0, "low": 2449.0, + "close": 2450.0, "volume": 1.0, + }]}) + assert lg.resolve_pending("XAU/USD", d1_bars) == 1 + d1_state = json.loads(lg._conn.execute( + "SELECT advisory_outcome_json FROM signals WHERE id=?", + (d1_rowid,)).fetchone()[0]) + assert d1_state["status"] == "PENDING" + assert d1_state["progress"]["entry_at"] == \ + "2026-08-04T00:00:00+00:00" + lg.close() + finally: + slog.reset_loggers_for_tests() + if os.path.exists(path): + os.remove(path) + + @test("signal_log: module entry honors SIGNAL_LOG_ENABLED/PATH env") def _(): import signal_log as slog @@ -8025,11 +8366,34 @@ def _(): " watch TEXT, note TEXT)") con.commit() con.close() - logger = slog.SignalLogger(path) + barrier = threading.Barrier(4) + loggers = [] + errors = [] + + def open_logger(): + try: + barrier.wait() + loggers.append(slog.SignalLogger(path)) + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=open_logger) for _ in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + assert not errors, f"concurrent additive migration failed: {errors}" + assert len(loggers) == 4 + logger = loggers[0] columns = { row[1] for row in logger._conn.execute("PRAGMA table_info(signals)") } - assert {"indicators_json", "candles_json"} <= columns + assert { + "indicators_json", "candles_json", "panel_created_at", + "panel_cache_status", "panel_payload_hash", + "advisory_outcome_status", "advisory_r_multiple", + "advisory_outcome_json", + } <= columns outlook_columns = { row[1] for row in logger._conn.execute("PRAGMA table_info(outlooks)") } @@ -8048,7 +8412,8 @@ def _(): "candles": {"1h": {"bars": [[-0.2, 0.1, -0.3, 0.0]]}}, } rowid = logger.log_outlook(result, surface="test", payload=payload) - logger.close() + for open_instance in loggers: + open_instance.close() con = sqlite3.connect(path) row = con.execute( "SELECT surface, bias, setup_forming, confidence, watch, note, "