diff --git a/src/agent_loop/decision_pipeline.py b/src/agent_loop/decision_pipeline.py index 70744d9..232999f 100644 --- a/src/agent_loop/decision_pipeline.py +++ b/src/agent_loop/decision_pipeline.py @@ -81,6 +81,12 @@ def __init__( 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 @@ def __init__( 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 @@ async def evaluate( 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: diff --git a/src/agent_loop/outcome_tracker.py b/src/agent_loop/outcome_tracker.py index 5e29903..ce45a2f 100644 --- a/src/agent_loop/outcome_tracker.py +++ b/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", diff --git a/tests/unit/test_decision_pipeline_degraded.py b/tests/unit/test_decision_pipeline_degraded.py new file mode 100644 index 0000000..0cb85b1 --- /dev/null +++ b/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 diff --git a/tests/unit/test_outcome_tracker.py b/tests/unit/test_outcome_tracker.py new file mode 100644 index 0000000..5dd70e6 --- /dev/null +++ b/tests/unit/test_outcome_tracker.py @@ -0,0 +1,117 @@ +"""Tests for OutcomeTracker — the calibration feedback heart. + +Covers #55 (trading-day T+N horizons) and #54 (proper Bayesian likelihoods, +P(evidence | state) by conditional frequency — not the old hit-rate / 1-p form). +""" + +from __future__ import annotations + +from datetime import UTC, date, datetime, timedelta + +import pytest + +from src.agent_loop.outcome_tracker import OutcomeTracker + + +@pytest.fixture +def tracker(tmp_path): + return OutcomeTracker(db_path=tmp_path / "ot.db") + + +def _insert(tracker, *, source, confidence, direction, correct, n): + created = datetime.now(UTC).isoformat() + with tracker._conn() as conn: + for i in range(n): + conn.execute( + """INSERT INTO tracked_signals + (signal_id, symbol, direction, source, confidence, + created_at, status, direction_correct) + VALUES (?, '600519', ?, ?, ?, ?, 'complete', ?)""", + ( + f"{source}-{direction}-{correct}-{confidence}-{i}", + direction, + source, + confidence, + created, + correct, + ), + ) + conn.commit() + + +# --------------------------------------------------------------------------- +# #55 — trading-day horizons +# --------------------------------------------------------------------------- + + +class _FakeCal: + def next_trading_day(self, d): + nxt = d + timedelta(days=1) + while nxt.weekday() >= 5: # skip Sat/Sun + nxt += timedelta(days=1) + return nxt + + +class _BrokenCal: + def next_trading_day(self, d): + raise RuntimeError("calendar unavailable") + + +class TestTradingDayHorizons: + def test_horizons_use_trading_days(self, tracker): + tracker._calendar = _FakeCal() + fri = datetime(2024, 1, 5, tzinfo=UTC) # a Friday + assert tracker._add_trading_days(fri, 1) == date(2024, 1, 8) # → Monday + assert tracker._add_trading_days(fri, 5) == date(2024, 1, 12) # 5 sessions + + def test_fallback_to_calendar_days_when_calendar_breaks(self, tracker): + tracker._calendar = _BrokenCal() + start = datetime(2024, 1, 5, tzinfo=UTC) + assert tracker._add_trading_days(start, 3) == date(2024, 1, 8) + + +# --------------------------------------------------------------------------- +# #54 — Bayesian likelihoods are real conditional frequencies +# --------------------------------------------------------------------------- + + +class TestBayesianLikelihood: + def test_discriminating_source_pbull_far_above_pbear(self, tracker): + # Predicts up in bull states, down in bear states → highly informative. + _insert(tracker, source="good", confidence=0.8, direction="buy", correct=1, n=8) + _insert( + tracker, source="good", confidence=0.8, direction="sell", correct=1, n=8 + ) + p_bull, p_bear = tracker.get_calibration_data(min_samples=10)["good/strong"] + assert p_bull > 0.8 and p_bear < 0.2 # strong positive log-likelihood-ratio + + def test_useless_source_pbull_near_pbear(self, tracker): + # Random: equally right/wrong in both directions → no edge. + for direction in ("buy", "sell"): + _insert( + tracker, + source="rng", + confidence=0.8, + direction=direction, + correct=1, + n=5, + ) + _insert( + tracker, + source="rng", + confidence=0.8, + direction=direction, + correct=0, + n=5, + ) + p_bull, p_bear = tracker.get_calibration_data(min_samples=10)["rng/strong"] + assert abs(p_bull - p_bear) < 0.05 # LLR ≈ 0 + + def test_likelihoods_are_not_complementary(self, tracker): + # A bull-biased source: high P(bull-read|bull) AND high P(bull-read|bear). + # The old code forced p_bear = 1 - p_bull; proper likelihoods need not. + _insert(tracker, source="bias", confidence=0.8, direction="buy", correct=1, n=8) + _insert(tracker, source="bias", confidence=0.8, direction="buy", correct=0, n=4) + p_bull, p_bear = tracker.get_calibration_data(min_samples=10)["bias/strong"] + assert p_bull > p_bear + assert (p_bull + p_bear) > 1.1 # would be exactly 1.0 under the old bug