fix(agent_loop): Bayesian likelihoods + trading-day horizons + debate de-risk (P2.2–P2.4)

src/agent_loop/decision_pipeline.py

@@ -81,6 +81,12 @@
         self._consecutive_loss_threshold = cfg.get("consecutive_loss_threshold", 3)
         self._consecutive_loss_factor = cfg.get("consecutive_loss_size_factor", 0.5)
         self._overnight_risk_budget = cfg.get("overnight_risk_budget_pct", 0.05)
+        # De-risk the LLM-debate dependency: by default a buy is refused when the
+        # debate engine is unavailable. When enabled, the deterministic stack may
+        # issue a damped buy off the Bayesian prescreen so a missing/failed oracle
+        # degrades gracefully instead of silently dropping the signal (#56).
+        self._allow_degraded_buys = cfg.get("allow_degraded_buys", False)
+        self._degraded_buy_min_prob = cfg.get("degraded_buy_min_prob", 0.60)
 
         # Circuit breaker state (Fix 2)
         self._portfolio_drawdown_halt_pct = cfg.get("portfolio_drawdown_halt_pct", 0.05)
@@ -96,6 +102,30 @@
 
         logger.info("DecisionPipeline initialized")
 
+    def _degraded_buy_record(
+        self, signal: AggregatedSignal, prescreen_p: float
+    ) -> dict[str, Any] | None:
+        """Synthesize a conservative buy when the LLM debate is unavailable.
+
+        Returns ``None`` (refuse the buy — the safe default) unless
+        ``allow_degraded_buys`` is set and the Bayesian prescreen clears
+        ``degraded_buy_min_prob``. Scores are damped — there is no debate
+        corroboration — so a degraded buy is weaker than a debated one.
+        """
+        if not self._allow_degraded_buys or prescreen_p < self._degraded_buy_min_prob:
+            return None
+        return {
+            "bull_score": round(prescreen_p * 0.8, 4),
+            "bear_score": round((1.0 - prescreen_p) * 0.8, 4),
+            "reasoning": (
+                f"Degraded buy (no debate): Bayesian prescreen P(bull)={prescreen_p:.2f}"
+            ),
+            "risk_veto": False,
+            "final_action": signal.direction.value.lower(),
+            "verdict": {"degraded": True},
+            "degraded": True,
+        }
+
     def _bayesian_prescreen(
         self,
         signal: AggregatedSignal,
@@ -402,55 +432,68 @@
                 return None
 
         # --- UST: Bayesian prescreen (cheap, before expensive debate) ---
-        if signal.direction.value.lower() in ("buy", "add"):
-            preliminary_p = self._bayesian_prescreen(
+        is_buy = signal.direction.value.lower() in ("buy", "add")
+        prescreen_p = 1.0
+        if is_buy:
+            prescreen_p = self._bayesian_prescreen(
                 signal, thesis, mkt, portfolio, available_cash
             )
-            if preliminary_p < self._prescreen_threshold:
+            if prescreen_p < self._prescreen_threshold:
                 logger.info(
                     "Bayesian prescreen: P(bull)=%.2f < %.2f — skip debate %s",
-                    preliminary_p,
+                    prescreen_p,
                     self._prescreen_threshold,
                     signal.symbol,
                 )
                 return None
 
         # --- UST: LLM budget check (before expensive debate) ---
-        if self._budget_tracker and not self._budget_tracker.can_call("gemini_web"):
-            if signal.direction.value.lower() in ("buy", "add"):
-                logger.info(
-                    "LLM budget exhausted — skip debate for buy %s", signal.symbol
-                )
-                return None
-            # Sell/reduce: pass through without debate
+        budget_exhausted = bool(
+            self._budget_tracker and not self._budget_tracker.can_call("gemini_web")
+        )
+        if budget_exhausted and not is_buy:
             logger.info(
                 "LLM budget exhausted — sell/reduce %s passes through", signal.symbol
             )
 
-        # Run debate — no fallback for buys if debate engine unavailable
-        debate_record = self._run_debate(signal, mkt, thesis)
+        # Run the debate unless the budget is exhausted for a buy.
+        debate_record = (
+            None
+            if (budget_exhausted and is_buy)
+            else self._run_debate(signal, mkt, thesis)
+        )
         if debate_record is None:
-            # Debate unavailable: buys are blocked, sells pass through
-            if signal.direction.value.lower() in ("buy", "add"):
-                logger.warning(
-                    "Debate engine unavailable — refusing buy for %s "
-                    "(no degraded recommendation without debate)",
+            if is_buy:
+                # Debate unavailable (no engine, or budget exhausted). Default is
+                # to refuse; allow_degraded_buys lets the deterministic stack
+                # issue a damped buy off the prescreen instead (#56).
+                debate_record = self._degraded_buy_record(signal, prescreen_p)
+                if debate_record is None:
+                    logger.warning(
+                        "Debate unavailable — refusing buy for %s "
+                        "(set allow_degraded_buys for graceful degradation)",
+                        signal.symbol,
+                    )
+                    return None
+                logger.info(
+                    "Debate unavailable — degraded buy for %s (prescreen P=%.2f)",
                     signal.symbol,
+                    prescreen_p,
                 )
-                return None
-            # Sell/reduce: allow passthrough without debate
-            logger.info(
-                "Debate engine unavailable — sell/reduce for %s passes through",
-                signal.symbol,
-            )
-            debate_record = {
-                "bull_score": 0.0,
-                "bear_score": signal.confidence,
-                "reasoning": signal.reason,
-                "risk_veto": False,
-                "final_action": signal.direction.value.lower(),
-                "verdict": {},
-            }
+            else:
+                # Sell/reduce: allow passthrough without debate
+                logger.info(
+                    "Debate engine unavailable — sell/reduce for %s passes through",
+                    signal.symbol,
+                )
+                debate_record = {
+                    "bull_score": 0.0,
+                    "bear_score": signal.confidence,
+                    "reasoning": signal.reason,
+                    "risk_veto": False,
+                    "final_action": signal.direction.value.lower(),
+                    "verdict": {},
+                }
 
         # Record LLM call in budget tracker
         if self._budget_tracker and debate_record:

src/agent_loop/outcome_tracker.py

@@ -80,6 +80,26 @@ def __init__(self, db_path: Path | str | None = None) -> None:
         self._db_path = Path(db_path) if db_path else _DEFAULT_DB_PATH
         self._db_path.parent.mkdir(parents=True, exist_ok=True)
         self._init_db()
+        self._calendar: Any = None
+
+    def _add_trading_days(self, start: datetime, n: int):
+        """Return the date ``n`` *trading* days after ``start``.
+
+        T+1/T+3/T+5 are trading-session horizons, not calendar days — a T+1
+        lookup after a Friday signal must land on the next session (Monday),
+        not Saturday. Falls back to calendar days if the calendar is unavailable.
+        """
+        try:
+            if self._calendar is None:
+                from src.data.trading_calendar import TradingCalendar
+
+                self._calendar = TradingCalendar()
+            d = start.date()
+            for _ in range(n):
+                d = self._calendar.next_trading_day(d)
+            return d
+        except Exception:
+            return (start + timedelta(days=n)).date()
 
     def _init_db(self) -> None:
         with self._conn() as conn:
@@ -304,15 +324,19 @@ async def evaluate_pending(
                 continue
 
             created_at = datetime.fromisoformat(created_at_str)
-            age_days = (now - created_at).days
+            today = now.date()
+
+            # Trading-day horizons (T+N = N trading sessions, not calendar days)
+            t1_date = self._add_trading_days(created_at, 1)
+            t3_date = self._add_trading_days(created_at, 3)
+            t5_date = self._add_trading_days(created_at, 5)
 
             updates: dict[str, Any] = {}
             new_status = status
 
             # T+1 evaluation
-            if t1_price is None and age_days >= 1:
+            if t1_price is None and t1_date <= today:
                 try:
-                    t1_date = created_at + timedelta(days=1)
                     price = await price_fetcher(symbol, t1_date.strftime("%Y-%m-%d"))
                     if price:
                         t1_return = (price - entry_price) / entry_price
@@ -323,9 +347,8 @@ async def evaluate_pending(
                     pass
 
             # T+3 evaluation
-            if t3_price is None and age_days >= 3:
+            if t3_price is None and t3_date <= today:
                 try:
-                    t3_date = created_at + timedelta(days=3)
                     price = await price_fetcher(symbol, t3_date.strftime("%Y-%m-%d"))
                     if price:
                         t3_return = (price - entry_price) / entry_price
@@ -336,9 +359,8 @@ async def evaluate_pending(
                     pass
 
             # T+5 evaluation (final)
-            if t5_price is None and age_days >= 5:
+            if t5_price is None and t5_date <= today:
                 try:
-                    t5_date = created_at + timedelta(days=5)
                     price = await price_fetcher(symbol, t5_date.strftime("%Y-%m-%d"))
                     if price:
                         t5_return = (price - entry_price) / entry_price
@@ -545,10 +567,24 @@ async def record_missed_opportunity(self, missed: MissedOpportunity) -> None:
     def get_calibration_data(
         self, lookback_days: int = 90, min_samples: int = 10
     ) -> dict[str, tuple[float, float]]:
-        """Get empirical likelihood ratios for Bayesian calibration.
+        """Get empirical likelihoods for Bayesian calibration.
+
+        Returns dict mapping ``"{source}/{confidence_bucket}"`` to
+        ``(P(bullish reading | true bull), P(bullish reading | true bear))``.
+
+        These are genuine **likelihoods** P(evidence | state), estimated as
+        conditional frequencies — NOT the source's hit-rate. The realized state
+        is reconstructed per signal from its predicted direction and whether the
+        prediction was correct:
+
+            realized bull = (predicted up AND correct) OR (predicted down AND wrong)
 
-        Returns dict mapping "{source}/{confidence_bucket}" to
-        (P(signal|bull), P(signal|bear)) tuples.
+        so ``P(bullish reading | bull) = #(predicted up & realized bull) / #(realized bull)``
+        and likewise for bear. They are Laplace-smoothed and do **not** sum to 1
+        (the previous ``p_bear = 1 - p_bull`` was a posterior-style identity that
+        does not hold for likelihoods). The log-likelihood-ratio the engine uses
+        is ``log(p_bull / p_bear)``, positive only when a source reads bullish
+        more often in true bull states than in true bear states.
 
         Only returns buckets with >= min_samples completed outcomes.
         """
@@ -564,23 +600,38 @@ def get_calibration_data(
                             ELSE 'weak'
                           END as bucket,
                           COUNT(*) as total,
-                          SUM(CASE WHEN direction_correct = 1 THEN 1 ELSE 0 END) as correct
+                          -- realized bull = up-call correct OR down-call wrong
+                          SUM(CASE WHEN (direction IN ('buy', 'add')
+                                         AND direction_correct = 1)
+                                     OR (direction IN ('sell', 'reduce')
+                                         AND direction_correct = 0)
+                                   THEN 1 ELSE 0 END) as n_bull,
+                          -- bullish reading that landed in a true bull state
+                          SUM(CASE WHEN direction IN ('buy', 'add')
+                                    AND direction_correct = 1
+                                   THEN 1 ELSE 0 END) as bull_in_bull,
+                          -- bullish reading that landed in a true bear state
+                          SUM(CASE WHEN direction IN ('buy', 'add')
+                                    AND direction_correct = 0
+                                   THEN 1 ELSE 0 END) as bull_in_bear
                    FROM tracked_signals
                    WHERE status = 'complete'
+                   AND direction_correct IS NOT NULL
                    AND created_at > ?
                    GROUP BY source, bucket
                    HAVING total >= ?""",
                 (cutoff, min_samples),
             ).fetchall()
 
-        for source, bucket, total, correct in rows:
-            key = f"{source}/{bucket}"
-            p_given_bull = correct / total if total > 0 else 0.5
-            p_given_bear = 1.0 - p_given_bull
-            # Avoid extreme values
-            p_given_bull = max(0.1, min(0.9, p_given_bull))
-            p_given_bear = max(0.1, min(0.9, p_given_bear))
-            calibration[key] = (p_given_bull, p_given_bear)
+        for source, bucket, total, n_bull, bull_in_bull, bull_in_bear in rows:
+            n_bear = total - n_bull
+            # Laplace-smoothed conditional frequencies → proper likelihoods.
+            p_given_bull = (bull_in_bull + 1) / (n_bull + 2)
+            p_given_bear = (bull_in_bear + 1) / (n_bear + 2)
+            # Clamp to keep the LLR finite/stable.
+            p_given_bull = max(0.05, min(0.95, p_given_bull))
+            p_given_bear = max(0.05, min(0.95, p_given_bear))
+            calibration[f"{source}/{bucket}"] = (p_given_bull, p_given_bear)
 
         logger.info(
             "Calibration data: %d buckets with >= %d samples",

tests/unit/test_decision_pipeline_degraded.py

@@ -0,0 +1,38 @@
+"""Tests for #56 — graceful degradation of the LLM-debate dependency.
+
+By default a buy is refused when the debate engine is unavailable; with
+``allow_degraded_buys`` the deterministic stack may issue a damped buy off the
+Bayesian prescreen instead of silently dropping the signal.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from src.agent_loop.decision_pipeline import DecisionPipeline
+
+
+def _buy_signal():
+    return SimpleNamespace(direction=SimpleNamespace(value="buy"))
+
+
+class TestDegradedBuy:
+    def test_refused_by_default(self):
+        dp = DecisionPipeline()
+        assert dp._degraded_buy_record(_buy_signal(), 0.9) is None
+
+    def test_enabled_issues_damped_buy_above_threshold(self):
+        dp = DecisionPipeline(
+            config={"allow_degraded_buys": True, "degraded_buy_min_prob": 0.6}
+        )
+        rec = dp._degraded_buy_record(_buy_signal(), 0.9)
+        assert rec is not None
+        assert rec["degraded"] is True
+        assert rec["bull_score"] > rec["bear_score"]
+        assert rec["bull_score"] < 0.9  # damped — no debate corroboration
+
+    def test_enabled_but_below_threshold_refused(self):
+        dp = DecisionPipeline(
+            config={"allow_degraded_buys": True, "degraded_buy_min_prob": 0.6}
+        )
+        assert dp._degraded_buy_record(_buy_signal(), 0.5) is None