agent_panel: ATR-normalized candle + indicator-value enrichment, hardened prompts, verdict logging

test_all.py

@@ -6952,10 +6952,14 @@ def _():
     app = load_app_config(_write_json({"agent_panel": {
         "enabled": True, "telegram": True, "dashboard": False,
         "min_quality": "A", "max_panels_per_day": 10,
-        "reuse_ttl_seconds": 600}}))
+        "reuse_ttl_seconds": 600, "candles_enabled": False,
+        "candle_bars_primary": 30, "candle_bars_confirm": 12}}))
     assert app.agent_panel_settings["enabled"] is True
     assert app.agent_panel_settings["min_quality"] == "A"
     assert app.agent_panel_settings["max_panels_per_day"] == 10
+    assert app.agent_panel_settings["candles_enabled"] is False
+    assert app.agent_panel_settings["candle_bars_primary"] == 30
+    assert app.agent_panel_settings["candle_bars_confirm"] == 12
     assert load_app_config(_write_json({})).agent_panel_settings == {}
     assert AppConfig().agent_panel_settings == {}
 
@@ -6971,6 +6975,11 @@ def _():
         {"agent_panel": {"max_panels_per_day": 0}},      # not positive
         {"agent_panel": {"max_panels_per_day": True}},   # bool is not an int
         {"agent_panel": {"reuse_ttl_seconds": -5}},      # negative
+        {"agent_panel": {"candles_enabled": "yes"}},     # not a bool
+        {"agent_panel": {"candle_bars_primary": 0}},     # not positive
+        {"agent_panel": {"candle_bars_primary": 20.5}},  # not an int
+        {"agent_panel": {"candle_bars_confirm": True}},  # bool is not an int
+        {"agent_panel": {"candle_bars_confirm": 500}},   # absurdly large
         {"agent_panel": "oops"},                         # not an object
     ):
         try:
@@ -6998,6 +7007,23 @@ class _PanelStub:
         return reply(user_payload) if callable(reply) else reply
 
 
+class _NullDP:
+    """Provider stub: candle fetch blows up → _candle_context degrades to
+    None (and tests stay off the network)."""
+    def fetch_ohlcv(self, *args, **kwargs):
+        raise RuntimeError("no network in tests")
+
+
+class _FakeDP:
+    """Provider stub serving synthetic OHLCV per timeframe."""
+    def __init__(self):
+        self.calls = []
+
+    def fetch_ohlcv(self, symbol, timeframe, lookback_days=None):
+        self.calls.append((symbol, timeframe))
+        return make_ohlcv(60, seed=7)
+
+
 _TAKE_OK = {"stance": "support", "confidence": 70, "note": "fine"}
 _SYNTH_OK = {"verdict": "AGREE", "confidence": 75, "rationale": "panel backs it"}
 _ALL_OK = {ap._MACRO_SYSTEM: _TAKE_OK, ap._TECH_SYSTEM: _TAKE_OK,
@@ -7148,6 +7174,7 @@ def _():
     try:
         stub = _PanelStub(dict(_ALL_OK))
         lc._client = stub
+        ap._dp = _NullDP()
         report = _panel_report()
         assert ap.attach_panel(report, _PANEL_ON,
                                context=FundamentalContext()) is True
@@ -7175,6 +7202,7 @@ def _():
         assert len(stub.calls) == 10
     finally:
         lc._client = None
+        ap._dp = None
         ap.reset_panel_state()
 
 
@@ -7183,6 +7211,7 @@ def _():
     try:
         stub = _PanelStub(dict(_ALL_OK))
         lc._client = stub
+        ap._dp = _NullDP()
         settings = {**_PANEL_ON, "max_panels_per_day": 1,
                     "reuse_ttl_seconds": 0}
         assert ap.attach_panel(_panel_report(), settings,
@@ -7201,6 +7230,7 @@ def _():
         assert ap.panel_health()["used_today"] == 1
     finally:
         lc._client = None
+        ap._dp = None
         ap.reset_panel_state()
 
 
@@ -7209,6 +7239,7 @@ def _():
     try:
         stub = _PanelStub(dict(_ALL_OK))
         lc._client = stub
+        ap._dp = _NullDP()
         report = _panel_report()
         ctx = FundamentalContext()
         assert ap.attach_panel(report, {}, context=ctx) is False, \
@@ -7233,6 +7264,7 @@ def _():
             "force bypasses enabled/surface/quality"
     finally:
         lc._client = None
+        ap._dp = None
         ap.reset_panel_state()
 
 
@@ -7311,6 +7343,7 @@ def _():
             raise RuntimeError("boom")
     try:
         lc._client = _Bomb()
+        ap._dp = _NullDP()
         report = _panel_report()
         assert ap.attach_panel(report, _PANEL_ON,
                                context=FundamentalContext()) is False
@@ -7323,9 +7356,113 @@ def _():
             logging.disable(logging.NOTSET)
     finally:
         lc._client = None
+        ap._dp = None
         ap.reset_panel_state()
 
 
+@test("agent_panel: candle normalization is ATR-relative to final close")
+def _():
+    import pandas as pd
+    df = pd.DataFrame({
+        "Open": [100.0, 110.0], "High": [112.0, 115.0],
+        "Low": [98.0, 108.0], "Close": [110.0, 112.0],
+        "Volume": [1, 1]})
+    out = ap._normalize_candles(df, atr=4.0, bars=2)
+    assert out["ref_close"] == 112.0 and out["atr"] == 4.0
+    assert out["bars"][0] == [-3.0, 0.0, -3.5, -0.5]
+    assert out["bars"][1] == [-0.5, 0.8, -1.0, 0.0]
+    capped = ap._normalize_candles(df, atr=4.0, bars=50)
+    assert len(capped["bars"]) == 2, "window capped at available bars"
+
+
+@test("agent_panel: technical payload carries values + optional candles")
+def _():
+    report = _panel_report()
+    tech = ap._technical_payload(report)
+    assert tech["indicators"], "real analysis must yield indicators"
+    assert "value" in tech["indicators"]["rsi"]
+    assert "value" in tech["indicators"]["adx"]
+    assert "candles" not in tech, "candles only when provided"
+    fake = {"1h": {"ref_close": 2400.0, "atr": 5.0,
+                   "bars": [[0.0, 0.5, -0.5, 0.2]]}}
+    tech2 = ap._technical_payload(report, candles=fake)
+    assert tech2["candles"] == fake
+    json.dumps(tech2, sort_keys=True, allow_nan=False)
+
+
+@test("agent_panel: _candle_context windows, ATR gating, disable flag")
+def _():
+    try:
+        ap._dp = _FakeDP()
+        report = _panel_report()
+        settings = {**_PANEL_ON, "candle_bars_primary": 20,
+                    "candle_bars_confirm": 10}
+        ctx = ap._candle_context(report, settings)
+        tfs = [report.primary_analysis.timeframe] + [
+            a.timeframe for a in report.confirmation_analyses]
+        assert set(ctx) == set(tfs), f"one block per analysis TF: {set(ctx)}"
+        assert len(ctx[tfs[0]]["bars"]) == 20, "primary window"
+        assert len(ctx[tfs[1]]["bars"]) == 10, "confirm window"
+        assert all(len(b) == 4 for b in ctx[tfs[0]]["bars"])
+        json.dumps(ctx, allow_nan=False)
+
+        # disabled flag → omitted entirely, zero fetches
+        ap._dp = _FakeDP()
+        assert ap._candle_context(
+            report, {**settings, "candles_enabled": False}) is None
+        assert ap._dp.calls == [], "disabled must not fetch"
+
+        # degenerate ATR → that TF skipped, the rest survive
+        ap._dp = _FakeDP()
+        report.primary_analysis.indicators["atr"].value = 0.0
+        ctx2 = ap._candle_context(report, settings)
+        assert tfs[0] not in ctx2 and tfs[1] in ctx2
+
+        # provider explosion → None, never raises
+        ap._dp = _NullDP()
+        assert ap._candle_context(report, settings) is None
+    finally:
+        ap._dp = None
+
+
+@test("agent_panel: attach injects candles into the technical payload only")
+def _():
+    try:
+        ap._dp = _FakeDP()
+        stub = _PanelStub(dict(_ALL_OK))
+        lc._client = stub
+        report = _panel_report()
+        assert ap.attach_panel(report, _PANEL_ON,
+                               context=FundamentalContext()) is True
+        by_prompt = dict(stub.calls)
+        tech_payload = by_prompt[ap._TECH_SYSTEM]
+        assert "candles" in tech_payload
+        primary_tf = report.primary_analysis.timeframe
+        assert len(tech_payload["candles"][primary_tf]["bars"]) == \
+            ap.DEFAULT_CANDLE_BARS_PRIMARY
+        assert "value" in tech_payload["indicators"]["rsi"]
+        for sys_p in (ap._MACRO_SYSTEM, ap._RISK_SYSTEM,
+                      ap._CONTRARIAN_SYSTEM):
+            assert "candles" not in by_prompt[sys_p], \
+                "candles are technical-persona evidence only"
+    finally:
+        lc._client = None
+        ap._dp = None
+        ap.reset_panel_state()
+
+
+@test("agent_panel: prompts forbid outside knowledge and cap output words")
+def _():
+    assert "OFF-LIMITS" in ap._TAKE_CONTRACT
+    assert "max 15 words" in ap._TAKE_CONTRACT
+    assert "max 30 words" in ap._SYNTH_SYSTEM
+    assert "ATR units" in ap._TECH_SYSTEM
+    for prompt in (ap._MACRO_SYSTEM, ap._TECH_SYSTEM, ap._RISK_SYSTEM,
+                   ap._CONTRARIAN_SYSTEM):
+        assert ap._TAKE_CONTRACT in prompt, \
+            "every persona must carry the anti-hallucination contract"
+
+
 print("\n═══ Module 24: signal_log.py (SQLite signal history) ═══")
 
 
@@ -7441,6 +7578,92 @@ def _():
         os.remove(path)
 
 
+@test("signal_log: panel verdict updates the scan row post-hoc")
+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, context=None, source="test", timeframe="M5")
+        assert rowid >= 1, "log() must return the rowid"
+        report.panel_verdict = "OPPOSE"
+        report.panel_confidence = 64.0
+        report.panel_rationale = "risk too thin"
+        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))
+        lg.close()
+        conn = sqlite3.connect(path)
+        row = conn.execute(
+            "SELECT panel_verdict, panel_confidence, panel_rationale,"
+            " panel_takes_json 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"
+    finally:
+        slog.reset_loggers_for_tests()
+        os.remove(path)
+
+
+@test("signal_log: log_panel_update rowid hand-off; no-ops stay silent")
+def _():
+    import sqlite3
+    import signal_log as slog
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    os.remove(path)  # want a fresh file from SignalLogger
+    try:
+        os.environ["SIGNAL_LOG_PATH"] = path
+        os.environ.pop("SIGNAL_LOG_ENABLED", None)
+        # no rowid (backtest / unlogged report) → silent no-op
+        slog.log_panel_update(_make_log_report())
+        assert not os.path.exists(path), "no-op must not create the DB"
+
+        report = _make_log_report()
+        slog.log_signal(report, source="test", timeframe="M5")
+        assert getattr(report, "_signal_log_rowid", None) is not None, \
+            "log_signal must stash the rowid on the report"
+
+        # empty verdict → no-op (row keeps its empty insert values)
+        slog.log_panel_update(report)
+        conn = sqlite3.connect(path)
+        row = conn.execute(
+            "SELECT panel_verdict, panel_takes_json FROM signals").fetchone()
+        conn.close()
+        assert row[0] in (None, "") and row[1] is None
+
+        report.panel_verdict = "AGREE"
+        report.panel_confidence = 81.0
+        report.panel_rationale = "all aligned"
+        report.panel_takes = []
+        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()
+        conn.close()
+        assert row[0] == "AGREE" and row[1] == 81.0
+        assert row[2] == "all aligned"
+        assert json.loads(row[3]) == []
+
+        # disabled → no-op, never raises
+        os.environ["SIGNAL_LOG_ENABLED"] = "false"
+        slog.log_panel_update(report)
+    finally:
+        os.environ.pop("SIGNAL_LOG_PATH", None)
+        os.environ.pop("SIGNAL_LOG_ENABLED", None)
+        slog.reset_loggers_for_tests()
+        if os.path.exists(path):
+            os.remove(path)
+
+
 @test("signal_log: null context → news JSON NULL, row still written")
 def _():
     import sqlite3