feat: add skill opinion outcome performance statistics (#2119)

* feat: add skill opinion outcome evaluation core

* feat: add skill opinion outcome evaluation core

* fix: unify expected-start resolution

* changelog

* fix: validate persisted daily start sessions

* fix: preserve legacy local backtest windows

* feat: add skill opinion outcome statistics

* docs: clarify outcome statistics stage boundary

* fix: clarify backtest-only legacy start fallback

* fix: prevent pending outcome retry starvation

* fix: preserve explicit backtest start contract

* fix: rotate failed outcome candidates

docs/CHANGELOG.md

@@ -8,7 +8,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 > For user-friendly release highlights, see the [GitHub Releases](https://github.com/ZhuLinsen/daily_stock_analysis/releases) page.
 
 ## [Unreleased]
+- [修复] Outcome 候选按上次尝试时间公平调度，避免持续新增的缺失 key 使旧 `pending` outcome 永久得不到重试。
+- [新功能] 新增按 skill、horizon 与 outcome engine version 独立聚合的只读 Skill Opinion 表现统计；少于 30 条 evaluated 样本时仅返回观察性计数，不输出表现指标或调整运行时权重。
 - [修复] 统一等价股票代码的本地日线候选与同源窗口解析；冲突沪深交易所代码不再降级匹配裸码，回测仅接受快照或交易日历确认的起点，并在同一起点中优先完整的单一代码窗口。
+- [新功能] 新增按 individual SkillAgent 自身 signal、版本化 engine 与本地已存同源日线窗口计算并持久化 `skill_opinion_outcomes` 的核心服务。
 - [新功能] 新增按 individual SkillAgent 自身 signal、版本化 engine 与本地已存同源日线窗口计算并持久化 `skill_opinion_outcomes` 的核心服务；本阶段不提供管理员 API、表现统计、样本充足度或权重调整。
 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
 <!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->

docs/multi-strategy-contract.md

@@ -16,11 +16,19 @@
 
 每条 outcome 只使用 sample 自己的 canonical `signal`，不得读取最终 Agent decision、`skill_consensus` 或其他 skill 的 signal。`strong_buy` / `buy` 按 bullish 评价，`strong_sell` / `sell` 按 bearish 评价；方向收益严格大于零才是 `hit`，零收益是 `miss`。`hold` 在价格窗口完整后保存为 `observational`，不产生方向正确性。
 
-历史分析日期来自 `enhanced_context.date`，缺失时才回退到历史记录创建日期。Backtest 与 Outcome 统一通过共享 resolver 解析股票身份、重建受支持的旧市场快照并确定权威起始 session：优先使用市场一致且合法的 `market_phase_summary.effective_daily_bar_date`；缺少该字段时，只有 phase 与交易日历能够证明起点才进行推导，否则 fail closed，不允许选择任意更早的本地日线。共享窗口 resolver 只接受权威起始 session 的 bar，在同日起点中优先完整窗口，且起始与 forward bars 必须来自同一 stored code shape，不得跨候选拼接。
+历史分析日期来自 `enhanced_context.date`，缺失时才回退到历史记录创建日期。Backtest 与 Outcome 共享股票身份解析、受支持的旧市场快照重建和权威起始 session 判定：优先使用市场一致且合法的 `market_phase_summary.effective_daily_bar_date`；缺少该字段时，只有 phase 与交易日历能够证明起点才进行推导，否则 fail closed，不允许选择任意更早的本地日线。Outcome 只接受通过判定的 `expected_start_date`。Backtest 为兼容既有历史，可在非 session `effective_daily_bar_date` 对应的精确本地 bar 已存在时，通过显式 `backtest_start_date` 只读回放；该 fallback 不属于权威 Outcome 样本，不触发无效日期 refill，也不进入 Skill Outcome 统计或权重校准。共享窗口 resolver 在指定起点中优先完整窗口，且起始与 forward bars 必须来自同一 stored code shape，不得跨候选拼接。
 
-权威起始 session 已确定、但对应起始 bar 尚未写入，或未来本地日线不足时，保存为可重试 `pending`。损坏或晚于分析日期的 `effective_daily_bar_date`、股票市场与快照市场冲突，以及无法由可信 phase 与交易日历证明起点等永久无效元数据，保存为终态 `unable`，不得伪装成 `missing_start_bar` 持续重试。同一 engine version 下只有 `pending` 可更新，`evaluated`、`observational`、`unable` 均不可覆盖；规则变化必须提升 engine version。历史删除在同一写事务内按 outcome → sample → history 显式清理，不能依赖 SQLite 外键开关。
+对 Outcome 而言，权威起始 session 已确定、但对应起始 bar 尚未写入，或未来本地日线不足时，保存为可重试 `pending`。候选 key 按上次尝试时间（缺失 outcome 时按 sample 创建时间）公平调度；每次重试会刷新 `pending.updated_at` 并将其移至队尾，避免持续新增的缺失 key 饿死旧重试，也避免重试反向阻塞新样本。损坏或晚于分析日期的 `effective_daily_bar_date`、股票市场与快照市场冲突，以及无法由可信 phase 与交易日历证明起点等永久无效元数据，保存为终态 `unable`，不得伪装成 `missing_start_bar` 持续重试。同一 engine version 下只有 `pending` 可更新，`evaluated`、`observational`、`unable` 均不可覆盖；规则变化必须提升 engine version。历史删除在同一写事务内按 outcome → sample → history 显式清理，不能依赖 SQLite 外键开关。
 
-本 PR 基于已合并的 #2073，只提供 Outcome evaluator、repository 和 service 核心，不新增管理员 API、Schema、OpenAPI 或主 Pipeline 自动触发，也不提供表现统计、样本充足度、排名和权重调整。若后续需要运维入口，应以实际调用方和权限契约为依据独立审查。
+Outcome 核心阶段（#2116）基于已合并的 #2073，只提供 Outcome evaluator、repository 和 service 核心，当时未包含表现统计、样本充足度、排名或权重调整。其后的只读统计阶段在下节单独定义；该阶段仍不新增管理员 API、Schema、OpenAPI 或主 Pipeline 自动触发，也不调整运行时权重。若后续需要运维入口，应以实际调用方和权限契约为依据独立审查。
+
+### Skill Opinion Outcome 表现统计
+
+Outcome 统计是只读数据面，按 `skill_id + horizon + engine_version` 独立分 bucket。任何 bucket 都不能借用同 skill 的其他 horizon、其他 skill、其他 engine version 或全局样本解锁指标。当前固定门槛为 `evaluated >= 30`；只有 individual skill opinion 自身 signal 产生的 `hit` / `miss` 计入 evaluated，`pending`、`observational` 和 `unable` 只保留计数，不计入样本充足度。
+
+样本不足时，bucket 的 `sample_status` 为 `observational`，计数继续返回，但 `hit_rate_pct`、`miss_rate_pct`、`avg_directional_return_pct` 和 `unable_rate_pct` 全部为 `null`，不得输出排名或推导权重。样本充足时，hit/miss rate 以 `hit + miss` 为分母，平均方向收益只使用 evaluated rows；unable rate 以终态记录 `evaluated + observational + unable` 为分母，临时 `pending` 不得稀释永久失败比例。
+
+当前统计 service 不修改 `BacktestService.get_skill_summary()`、`AgentMemory` 或 `SkillAggregator`，也不新增 API、Pipeline 自动触发和 Web 展示。把统计接入保守权重属于独立后续变更，接入前无论统计表是否已有记录，运行时仍保持现有中性权重。
 
 ## 术语与边界
 

src/repositories/skill_opinion_outcome_repo.py

@@ -29,6 +29,23 @@ class SkillOpinionOutcomeCandidate:
     existing_outcome: Optional[SkillOpinionOutcomeRecord]
 
 
+@dataclass(frozen=True)
+class SkillOpinionPerformanceBucket:
+    """Raw persisted counts for one skill, horizon, and engine version."""
+
+    skill_id: str
+    horizon: str
+    engine_version: str
+    total: int
+    pending: int
+    evaluated: int
+    observational: int
+    unable: int
+    hit: int
+    miss: int
+    avg_directional_return_pct: Optional[float]
+
+
 class SkillOpinionOutcomeRepository:
     """Read candidates and persist outcomes through the shared write guard."""
 
@@ -87,10 +104,6 @@ class SkillOpinionOutcomeRepository:
                     .outerjoin(SkillOpinionOutcomeRecord, join_condition)
                     .where(and_(*conditions))
                     .order_by(
-                        case(
-                            (SkillOpinionOutcomeRecord.id.is_(None), 0),
-                            else_=1,
-                        ),
                         func.coalesce(
                             SkillOpinionOutcomeRecord.updated_at,
                             SkillOpinionSampleRecord.created_at,
@@ -112,10 +125,10 @@ class SkillOpinionOutcomeRepository:
         horizon_rank = {horizon: index for index, horizon in enumerate(horizons)}
         candidates.sort(
             key=lambda item: (
-                item.existing_outcome is not None,
                 self._candidate_time(item),
                 int(item.sample.id),
                 horizon_rank[item.horizon],
+                item.existing_outcome is not None,
             )
         )
         return candidates[:limit]
@@ -188,6 +201,81 @@ class SkillOpinionOutcomeRepository:
                 .limit(1)
             ).scalar_one_or_none()
 
+    def list_performance_buckets(
+        self,
+        *,
+        engine_version: str,
+        skill_id: Optional[str] = None,
+        horizons: Optional[Sequence[str]] = None,
+    ) -> List[SkillOpinionPerformanceBucket]:
+        """Aggregate persisted outcome facts without applying sample policy."""
+
+        status = SkillOpinionOutcomeRecord.eval_status
+        outcome = SkillOpinionOutcomeRecord.outcome
+        conditions = [
+            SkillOpinionOutcomeRecord.engine_version == engine_version
+        ]
+        if skill_id is not None:
+            conditions.append(SkillOpinionSampleRecord.skill_id == skill_id)
+        if horizons is not None:
+            conditions.append(
+                SkillOpinionOutcomeRecord.horizon.in_(list(horizons))
+            )
+        with self.db.get_session() as session:
+            rows = session.execute(
+                select(
+                    SkillOpinionSampleRecord.skill_id,
+                    SkillOpinionOutcomeRecord.horizon,
+                    SkillOpinionOutcomeRecord.engine_version,
+                    func.count(SkillOpinionOutcomeRecord.id),
+                    func.sum(case((status == "pending", 1), else_=0)),
+                    func.sum(case((status == "evaluated", 1), else_=0)),
+                    func.sum(case((status == "observational", 1), else_=0)),
+                    func.sum(case((status == "unable", 1), else_=0)),
+                    func.sum(case((outcome == "hit", 1), else_=0)),
+                    func.sum(case((outcome == "miss", 1), else_=0)),
+                    func.avg(
+                        case(
+                            (
+                                status == "evaluated",
+                                SkillOpinionOutcomeRecord.directional_return_pct,
+                            ),
+                            else_=None,
+                        )
+                    ),
+                )
+                .join(
+                    SkillOpinionSampleRecord,
+                    SkillOpinionSampleRecord.id
+                    == SkillOpinionOutcomeRecord.skill_opinion_sample_id,
+                )
+                .where(and_(*conditions))
+                .group_by(
+                    SkillOpinionSampleRecord.skill_id,
+                    SkillOpinionOutcomeRecord.horizon,
+                    SkillOpinionOutcomeRecord.engine_version,
+                )
+            ).all()
+
+        return [
+            SkillOpinionPerformanceBucket(
+                skill_id=str(row[0]),
+                horizon=str(row[1]),
+                engine_version=str(row[2]),
+                total=int(row[3] or 0),
+                pending=int(row[4] or 0),
+                evaluated=int(row[5] or 0),
+                observational=int(row[6] or 0),
+                unable=int(row[7] or 0),
+                hit=int(row[8] or 0),
+                miss=int(row[9] or 0),
+                avg_directional_return_pct=(
+                    float(row[10]) if row[10] is not None else None
+                ),
+            )
+            for row in rows
+        ]
+
     @staticmethod
     def _candidate_time(candidate: SkillOpinionOutcomeCandidate) -> datetime:
         outcome = candidate.existing_outcome

src/services/backtest_service.py

@@ -135,10 +135,7 @@ class BacktestService:
                     else []
                 )
                 expected_start_date = start_resolution.expected_start_date
-                local_start_date = (
-                    expected_start_date
-                    or start_resolution.legacy_local_start_date
-                )
+                local_start_date = start_resolution.backtest_start_date
                 daily_window = None
                 if local_start_date is not None:
                     daily_window = resolve_stock_daily_window(

src/services/skill_opinion_outcome_service.py

@@ -109,6 +109,16 @@ class SkillOpinionOutcomeService:
                     "error_type": type(exc).__name__,
                 }
                 errors.append(error)
+                try:
+                    self._record_retry_attempt(candidate)
+                except Exception:
+                    logger.warning(
+                        "Failed to advance Skill opinion outcome retry marker: "
+                        "sample_id=%s horizon=%s",
+                        error["sample_id"],
+                        error["horizon"],
+                        exc_info=True,
+                    )
                 logger.warning(
                     "Skill opinion outcome evaluation deferred after transient failure: "
                     "sample_id=%s horizon=%s error_type=%s",
@@ -130,6 +140,21 @@ class SkillOpinionOutcomeService:
             "engine_version": SKILL_OPINION_OUTCOME_ENGINE_VERSION,
         }
 
+    def _record_retry_attempt(
+        self,
+        candidate: SkillOpinionOutcomeCandidate,
+    ) -> None:
+        self.repo.persist_outcome(
+            {
+                "skill_opinion_sample_id": int(candidate.sample.id),
+                "horizon": candidate.horizon,
+                "engine_version": SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+                "eval_status": "pending",
+                "outcome": None,
+                "direction_correct": None,
+            }
+        )
+
     def _evaluate_candidate(
         self,
         candidate: SkillOpinionOutcomeCandidate,

src/services/skill_opinion_performance_service.py

@@ -0,0 +1,153 @@
+# -*- coding: utf-8 -*-
+"""Read-only statistics over persisted individual SkillAgent outcomes."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Sequence
+
+from src.core.skill_opinion_outcome_evaluator import (
+    SUPPORTED_SKILL_OUTCOME_HORIZONS,
+)
+from src.repositories.skill_opinion_outcome_repo import (
+    SkillOpinionOutcomeRepository,
+    SkillOpinionPerformanceBucket,
+)
+from src.services.skill_opinion_outcome_service import (
+    SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+)
+from src.storage import DatabaseManager
+
+
+MIN_SKILL_OUTCOME_SAMPLE_SIZE = 30
+
+
+class SkillOpinionPerformanceService:
+    """Apply sample-sufficiency policy to raw Outcome aggregates."""
+
+    def __init__(
+        self,
+        *,
+        repo: Optional[SkillOpinionOutcomeRepository] = None,
+        db_manager: Optional[DatabaseManager] = None,
+    ):
+        self.repo = repo or SkillOpinionOutcomeRepository(db_manager)
+
+    def get_stats(
+        self,
+        *,
+        skill_id: Optional[str] = None,
+        horizons: Optional[Sequence[str]] = None,
+        engine_version: str = SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+    ) -> Dict[str, Any]:
+        """Return low-sensitive statistics for the current engine version."""
+
+        skill_id_norm = (
+            self._required_text(skill_id, "skill_id")
+            if skill_id is not None
+            else None
+        )
+        horizons_norm = self._normalize_horizons(horizons)
+        engine_version_norm = self._required_text(
+            engine_version,
+            "engine_version",
+        )
+        buckets = self.repo.list_performance_buckets(
+            engine_version=engine_version_norm,
+            skill_id=skill_id_norm,
+            horizons=horizons_norm,
+        )
+        horizon_rank = {
+            horizon: index
+            for index, horizon in enumerate(
+                SUPPORTED_SKILL_OUTCOME_HORIZONS
+            )
+        }
+        buckets.sort(
+            key=lambda bucket: (
+                -bucket.total,
+                bucket.skill_id,
+                horizon_rank[bucket.horizon],
+            )
+        )
+        return {
+            "engine_version": engine_version_norm,
+            "minimum_evaluated_sample_size": MIN_SKILL_OUTCOME_SAMPLE_SIZE,
+            "buckets": [self._serialize_bucket(bucket) for bucket in buckets],
+        }
+
+    @staticmethod
+    def _serialize_bucket(
+        bucket: SkillOpinionPerformanceBucket,
+    ) -> Dict[str, Any]:
+        sample_sufficient = (
+            bucket.evaluated >= MIN_SKILL_OUTCOME_SAMPLE_SIZE
+        )
+        direction_denominator = bucket.hit + bucket.miss
+        terminal_denominator = (
+            bucket.evaluated + bucket.observational + bucket.unable
+        )
+        return {
+            "skill_id": bucket.skill_id,
+            "horizon": bucket.horizon,
+            "engine_version": bucket.engine_version,
+            "total": bucket.total,
+            "pending": bucket.pending,
+            "evaluated": bucket.evaluated,
+            "observational": bucket.observational,
+            "unable": bucket.unable,
+            "hit": bucket.hit,
+            "miss": bucket.miss,
+            "sample_sufficient": sample_sufficient,
+            "sample_status": (
+                "sufficient" if sample_sufficient else "observational"
+            ),
+            "hit_rate_pct": (
+                round(bucket.hit / direction_denominator * 100, 2)
+                if sample_sufficient and direction_denominator
+                else None
+            ),
+            "miss_rate_pct": (
+                round(bucket.miss / direction_denominator * 100, 2)
+                if sample_sufficient and direction_denominator
+                else None
+            ),
+            "avg_directional_return_pct": (
+                round(bucket.avg_directional_return_pct, 4)
+                if sample_sufficient
+                and bucket.avg_directional_return_pct is not None
+                else None
+            ),
+            "unable_rate_pct": (
+                round(bucket.unable / terminal_denominator * 100, 2)
+                if sample_sufficient and terminal_denominator
+                else None
+            ),
+        }
+
+    @staticmethod
+    def _required_text(value: Any, field_name: str) -> str:
+        text = str(value or "").strip()
+        if not text:
+            raise ValueError(f"{field_name} must not be blank")
+        return text
+
+    @staticmethod
+    def _normalize_horizons(
+        values: Optional[Sequence[str]],
+    ) -> Optional[List[str]]:
+        if values is None:
+            return None
+        if not values:
+            raise ValueError("horizons must not be empty")
+
+        normalized: List[str] = []
+        for value in values:
+            horizon = str(value or "").strip()
+            if horizon not in SUPPORTED_SKILL_OUTCOME_HORIZONS:
+                raise ValueError(
+                    "horizon must be one of "
+                    + ", ".join(SUPPORTED_SKILL_OUTCOME_HORIZONS)
+                )
+            if horizon not in normalized:
+                normalized.append(horizon)
+        return normalized

src/services/stock_daily_start_resolver.py

@@ -25,7 +25,13 @@ class DailyStockStartResolution:
     identity: Optional[DailyStockIdentity]
     expected_start_date: Optional[date]
     failure_reason: Optional[str]
-    legacy_local_start_date: Optional[date] = None
+    legacy_backtest_start_date: Optional[date] = None
+
+    @property
+    def backtest_start_date(self) -> Optional[date]:
+        """Return the authoritative start or Backtest-only legacy fallback."""
+
+        return self.expected_start_date or self.legacy_backtest_start_date
 
 
 def resolve_stock_daily_start(
@@ -99,7 +105,7 @@ def resolve_stock_daily_start(
                 identity,
                 None,
                 "invalid_effective_daily_bar_date",
-                legacy_local_start_date=effective_date,
+                legacy_backtest_start_date=effective_date,
             )
         return DailyStockStartResolution(identity, effective_date, None)
 

tests/test_backtest_service.py

@@ -19,6 +19,10 @@ from src.config import Config
 from src.core.backtest_engine import OVERALL_SENTINEL_CODE
 from src.repositories.backtest_repo import BacktestRepository
 from src.services.backtest_service import BacktestService
+from src.services.stock_daily_start_resolver import (
+    DailyStockStartResolution,
+    resolve_stock_daily_start,
+)
 from src.storage import AnalysisHistory, BacktestResult, BacktestSummary, DatabaseManager, StockDaily
 
 
@@ -1232,6 +1236,87 @@ class BacktestServiceTestCase(unittest.TestCase):
             self.assertEqual(result.start_price, 100.0)
             self.assertEqual(result.end_close, 103.0)
 
+    def test_non_session_resolution_exposes_backtest_only_local_start(
+        self,
+    ) -> None:
+        resolution = resolve_stock_daily_start(
+            stock_code="600520",
+            context_snapshot=_phase_snapshot(
+                date(2024, 1, 6),
+                phase="postmarket",
+                market="cn",
+                effective_date=date(2024, 1, 6),
+                trigger_source="api",
+            ),
+            analysis_date=date(2024, 1, 6),
+        )
+
+        self.assertIsNone(resolution.expected_start_date)
+        self.assertEqual(
+            resolution.failure_reason,
+            "invalid_effective_daily_bar_date",
+        )
+        self.assertEqual(
+            resolution.backtest_start_date,
+            date(2024, 1, 6),
+        )
+
+    def test_run_backtest_consumes_explicit_backtest_only_start(self) -> None:
+        analysis_date = date(2024, 1, 6)
+        self._seed_analysis(
+            query_id="q_explicit_backtest_only_start",
+            code="600521",
+            analysis_date=analysis_date,
+            created_at=datetime(2024, 1, 6, 0, 0, 0),
+            operation_advice="买入",
+            trend_prediction="看多",
+            start_close=100.0,
+            forward_bars=[
+                StockDaily(
+                    code="600521",
+                    date=date(2024, 1, 8),
+                    high=103.0,
+                    low=99.0,
+                    close=102.0,
+                )
+            ],
+            phase="postmarket",
+        )
+        identity = resolve_stock_daily_start(
+            stock_code="600521",
+            context_snapshot=_phase_snapshot(
+                analysis_date,
+                phase="postmarket",
+                market="cn",
+                effective_date=analysis_date,
+                trigger_source="api",
+            ),
+            analysis_date=analysis_date,
+        ).identity
+        resolution = DailyStockStartResolution(
+            identity=identity,
+            expected_start_date=None,
+            failure_reason="invalid_effective_daily_bar_date",
+            legacy_backtest_start_date=analysis_date,
+        )
+
+        with patch(
+            "src.services.backtest_service.resolve_stock_daily_start",
+            return_value=resolution,
+        ):
+            stats = BacktestService(self.db).run_backtest(
+                code="600521",
+                force=False,
+                eval_window_days=1,
+                min_age_days=0,
+                analysis_date_from=analysis_date,
+                analysis_date_to=analysis_date,
+                limit=10,
+            )
+
+        self.assertEqual(stats["completed"], 1)
+        self.assertEqual(stats["insufficient"], 0)
+
     def test_run_backtest_rebuilds_legacy_cn_snapshot_for_jp_history(self) -> None:
         legacy_snapshot = json.dumps(
             {

tests/test_skill_opinion_outcome_stats.py

@@ -0,0 +1,386 @@
+# -*- coding: utf-8 -*-
+"""Tests for read-only Skill Opinion Outcome performance statistics."""
+
+from __future__ import annotations
+
+import os
+from itertools import count
+
+import pytest
+
+from src.config import Config
+from src.repositories.skill_opinion_outcome_repo import (
+    SkillOpinionOutcomeRepository,
+)
+from src.services.skill_opinion_performance_service import (
+    SkillOpinionPerformanceService,
+)
+from src.services.skill_opinion_outcome_service import (
+    SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+)
+from src.storage import (
+    AnalysisHistory,
+    DatabaseManager,
+    SkillOpinionOutcomeRecord,
+    SkillOpinionSampleRecord,
+)
+
+_ROW_SEQUENCE = count(1)
+
+
+@pytest.fixture()
+def isolated_db(tmp_path):
+    old_database_path = os.environ.get("DATABASE_PATH")
+    os.environ["DATABASE_PATH"] = str(tmp_path / "skill_opinion_outcome_stats.db")
+    Config.reset_instance()
+    DatabaseManager.reset_instance()
+    db = DatabaseManager.get_instance()
+    try:
+        yield db
+    finally:
+        DatabaseManager.reset_instance()
+        Config.reset_instance()
+        if old_database_path is None:
+            os.environ.pop("DATABASE_PATH", None)
+        else:
+            os.environ["DATABASE_PATH"] = old_database_path
+
+
+def _add_outcome(
+    db: DatabaseManager,
+    *,
+    skill_id: str = "alpha",
+    horizon: str = "1d",
+    engine_version: str = SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+    eval_status: str,
+    outcome: str | None = None,
+    directional_return_pct: float | None = None,
+) -> None:
+    with db.session_scope() as session:
+        history = AnalysisHistory(
+            query_id=(
+                f"stats-{skill_id}-{horizon}-{eval_status}-{next(_ROW_SEQUENCE)}"
+            ),
+            code="600519",
+            report_type="simple",
+            operation_advice="hold",
+        )
+        session.add(history)
+        session.flush()
+        sample = SkillOpinionSampleRecord(
+            analysis_history_id=history.id,
+            stock_code="600519",
+            skill_id=skill_id,
+            signal="buy" if eval_status != "observational" else "hold",
+            confidence=0.8,
+            sample_schema_version="skill-opinion-sample-v1",
+        )
+        session.add(sample)
+        session.flush()
+        session.add(
+            SkillOpinionOutcomeRecord(
+                skill_opinion_sample_id=sample.id,
+                horizon=horizon,
+                engine_version=engine_version,
+                eval_status=eval_status,
+                outcome=outcome,
+                direction_correct=(
+                    outcome == "hit" if eval_status == "evaluated" else None
+                ),
+                directional_return_pct=directional_return_pct,
+                unable_reason=(
+                    "invalid_metadata" if eval_status == "unable" else None
+                ),
+            )
+        )
+
+
+def test_repository_aggregates_raw_bucket_counts(isolated_db) -> None:
+    _add_outcome(
+        isolated_db,
+        eval_status="evaluated",
+        outcome="hit",
+        directional_return_pct=5.0,
+    )
+    _add_outcome(
+        isolated_db,
+        eval_status="evaluated",
+        outcome="miss",
+        directional_return_pct=-2.0,
+    )
+    _add_outcome(isolated_db, eval_status="observational", outcome="observational")
+    _add_outcome(isolated_db, eval_status="unable")
+    _add_outcome(isolated_db, eval_status="pending")
+
+    buckets = SkillOpinionOutcomeRepository(
+        isolated_db
+    ).list_performance_buckets(
+        engine_version=SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+    )
+
+    assert len(buckets) == 1
+    bucket = buckets[0]
+    assert bucket.skill_id == "alpha"
+    assert bucket.horizon == "1d"
+    assert bucket.total == 5
+    assert bucket.pending == 1
+    assert bucket.evaluated == 2
+    assert bucket.observational == 1
+    assert bucket.unable == 1
+    assert bucket.hit == 1
+    assert bucket.miss == 1
+    assert bucket.avg_directional_return_pct == pytest.approx(1.5)
+
+
+def test_service_keeps_insufficient_bucket_observational(isolated_db) -> None:
+    _add_outcome(
+        isolated_db,
+        eval_status="evaluated",
+        outcome="hit",
+        directional_return_pct=5.0,
+    )
+    _add_outcome(
+        isolated_db,
+        eval_status="evaluated",
+        outcome="miss",
+        directional_return_pct=-2.0,
+    )
+    _add_outcome(isolated_db, eval_status="observational", outcome="observational")
+    _add_outcome(isolated_db, eval_status="unable")
+    _add_outcome(isolated_db, eval_status="pending")
+
+    stats = SkillOpinionPerformanceService(
+        db_manager=isolated_db
+    ).get_stats()
+
+    assert stats["engine_version"] == SKILL_OPINION_OUTCOME_ENGINE_VERSION
+    assert stats["minimum_evaluated_sample_size"] == 30
+    assert len(stats["buckets"]) == 1
+    bucket = stats["buckets"][0]
+    assert bucket["total"] == 5
+    assert bucket["evaluated"] == 2
+    assert bucket["observational"] == 1
+    assert bucket["unable"] == 1
+    assert bucket["pending"] == 1
+    assert bucket["hit"] == 1
+    assert bucket["miss"] == 1
+    assert bucket["sample_sufficient"] is False
+    assert bucket["sample_status"] == "observational"
+    assert bucket["hit_rate_pct"] is None
+    assert bucket["miss_rate_pct"] is None
+    assert bucket["avg_directional_return_pct"] is None
+    assert bucket["unable_rate_pct"] is None
+
+
+def test_service_unlocks_metrics_at_exact_sample_threshold(isolated_db) -> None:
+    for _ in range(18):
+        _add_outcome(
+            isolated_db,
+            eval_status="evaluated",
+            outcome="hit",
+            directional_return_pct=2.0,
+        )
+    for _ in range(12):
+        _add_outcome(
+            isolated_db,
+            eval_status="evaluated",
+            outcome="miss",
+            directional_return_pct=-1.0,
+        )
+    _add_outcome(isolated_db, eval_status="observational", outcome="observational")
+    _add_outcome(isolated_db, eval_status="unable")
+    _add_outcome(isolated_db, eval_status="unable")
+    _add_outcome(isolated_db, eval_status="pending")
+
+    bucket = SkillOpinionPerformanceService(
+        db_manager=isolated_db
+    ).get_stats()["buckets"][0]
+
+    assert bucket["total"] == 34
+    assert bucket["evaluated"] == 30
+    assert bucket["sample_sufficient"] is True
+    assert bucket["sample_status"] == "sufficient"
+    assert bucket["hit_rate_pct"] == 60.0
+    assert bucket["miss_rate_pct"] == 40.0
+    assert bucket["avg_directional_return_pct"] == 0.8
+    assert bucket["unable_rate_pct"] == 6.06
+
+
+def test_non_evaluated_rows_do_not_unlock_metrics(isolated_db) -> None:
+    for _ in range(29):
+        _add_outcome(
+            isolated_db,
+            eval_status="evaluated",
+            outcome="hit",
+            directional_return_pct=1.0,
+        )
+    for _ in range(5):
+        _add_outcome(
+            isolated_db,
+            eval_status="observational",
+            outcome="observational",
+        )
+        _add_outcome(isolated_db, eval_status="unable")
+        _add_outcome(isolated_db, eval_status="pending")
+
+    bucket = SkillOpinionPerformanceService(
+        db_manager=isolated_db
+    ).get_stats()["buckets"][0]
+
+    assert bucket["total"] == 44
+    assert bucket["evaluated"] == 29
+    assert bucket["sample_sufficient"] is False
+    assert bucket["sample_status"] == "observational"
+    assert bucket["hit_rate_pct"] is None
+    assert bucket["miss_rate_pct"] is None
+    assert bucket["avg_directional_return_pct"] is None
+    assert bucket["unable_rate_pct"] is None
+
+
+def test_service_filters_exact_bucket_identity(isolated_db) -> None:
+    _add_outcome(
+        isolated_db,
+        skill_id="alpha",
+        horizon="1d",
+        eval_status="evaluated",
+        outcome="hit",
+        directional_return_pct=1.0,
+    )
+    _add_outcome(
+        isolated_db,
+        skill_id="alpha",
+        horizon="3d",
+        eval_status="evaluated",
+        outcome="miss",
+        directional_return_pct=-1.0,
+    )
+    _add_outcome(
+        isolated_db,
+        skill_id="beta",
+        horizon="3d",
+        eval_status="evaluated",
+        outcome="hit",
+        directional_return_pct=2.0,
+    )
+    _add_outcome(
+        isolated_db,
+        skill_id="alpha",
+        horizon="3d",
+        engine_version="skill-opinion-outcome-v2",
+        eval_status="evaluated",
+        outcome="hit",
+        directional_return_pct=3.0,
+    )
+
+    stats = SkillOpinionPerformanceService(
+        db_manager=isolated_db
+    ).get_stats(
+        skill_id="alpha",
+        horizons=["3d"],
+        engine_version=SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+    )
+
+    assert len(stats["buckets"]) == 1
+    bucket = stats["buckets"][0]
+    assert bucket["skill_id"] == "alpha"
+    assert bucket["horizon"] == "3d"
+    assert bucket["engine_version"] == SKILL_OPINION_OUTCOME_ENGINE_VERSION
+    assert bucket["evaluated"] == 1
+    assert bucket["miss"] == 1
+
+
+def test_sibling_buckets_cannot_combine_to_unlock_metrics(
+    isolated_db,
+) -> None:
+    for skill_id, horizon, engine_version in [
+        ("alpha", "1d", SKILL_OPINION_OUTCOME_ENGINE_VERSION),
+        ("alpha", "3d", SKILL_OPINION_OUTCOME_ENGINE_VERSION),
+        ("beta", "1d", SKILL_OPINION_OUTCOME_ENGINE_VERSION),
+        ("alpha", "1d", "skill-opinion-outcome-v2"),
+    ]:
+        for _ in range(16):
+            _add_outcome(
+                isolated_db,
+                skill_id=skill_id,
+                horizon=horizon,
+                engine_version=engine_version,
+                eval_status="evaluated",
+                outcome="hit",
+                directional_return_pct=1.0,
+            )
+
+    service = SkillOpinionPerformanceService(db_manager=isolated_db)
+    current_buckets = service.get_stats()["buckets"]
+    future_buckets = service.get_stats(
+        engine_version="skill-opinion-outcome-v2"
+    )["buckets"]
+
+    assert len(current_buckets) == 3
+    assert all(bucket["evaluated"] == 16 for bucket in current_buckets)
+    assert all(
+        bucket["sample_sufficient"] is False
+        for bucket in current_buckets
+    )
+    assert len(future_buckets) == 1
+    assert future_buckets[0]["evaluated"] == 16
+    assert future_buckets[0]["sample_sufficient"] is False
+
+
+@pytest.mark.parametrize(
+    "filters",
+    [
+        {"skill_id": "   "},
+        {"horizons": []},
+        {"horizons": ["2d"]},
+        {"engine_version": "   "},
+    ],
+)
+def test_service_rejects_invalid_filters(isolated_db, filters) -> None:
+    with pytest.raises(ValueError):
+        SkillOpinionPerformanceService(
+            db_manager=isolated_db
+        ).get_stats(**filters)
+
+
+def test_service_orders_buckets_by_total_then_canonical_identity(
+    isolated_db,
+) -> None:
+    for skill_id, horizon, repetitions in [
+        ("zeta", "3d", 3),
+        ("alpha", "10d", 2),
+        ("alpha", "1d", 2),
+        ("beta", "5d", 3),
+    ]:
+        for _ in range(repetitions):
+            _add_outcome(
+                isolated_db,
+                skill_id=skill_id,
+                horizon=horizon,
+                eval_status="evaluated",
+                outcome="hit",
+                directional_return_pct=1.0,
+            )
+
+    buckets = SkillOpinionPerformanceService(
+        db_manager=isolated_db
+    ).get_stats()["buckets"]
+
+    assert [
+        (bucket["skill_id"], bucket["horizon"], bucket["total"])
+        for bucket in buckets
+    ] == [
+        ("beta", "5d", 3),
+        ("zeta", "3d", 3),
+        ("alpha", "1d", 2),
+        ("alpha", "10d", 2),
+    ]
+
+
+def test_service_returns_empty_buckets_for_valid_empty_filter(
+    isolated_db,
+) -> None:
+    stats = SkillOpinionPerformanceService(
+        db_manager=isolated_db
+    ).get_stats(skill_id="missing")
+
+    assert stats["buckets"] == []

tests/test_skill_opinion_outcomes.py

@@ -543,6 +543,111 @@ def test_pending_is_retried_but_terminal_outcome_is_immutable(isolated_db) -> No
     assert _stored_outcome(isolated_db, sample_id).stock_return_pct == pytest.approx(5.0)
 
 
+def test_old_pending_retry_is_not_starved_by_new_missing_candidates(
+    isolated_db,
+) -> None:
+    _, pending_sample_id = _add_sample(isolated_db, skill_id="pending")
+    _, missing_sample_id = _add_sample(isolated_db, skill_id="missing")
+    repo = SkillOpinionOutcomeRepository(isolated_db)
+    pending_fields = {
+        "skill_opinion_sample_id": pending_sample_id,
+        "horizon": "1d",
+        "engine_version": SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+        "eval_status": "pending",
+        "outcome": None,
+        "direction_correct": None,
+        "unable_reason": "insufficient_future_data",
+    }
+    repo.persist_outcome(pending_fields)
+
+    with isolated_db.session_scope() as session:
+        pending_outcome = session.query(SkillOpinionOutcomeRecord).filter_by(
+            skill_opinion_sample_id=pending_sample_id,
+            horizon="1d",
+            engine_version=SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+        ).one()
+        pending_outcome.updated_at = datetime(2024, 1, 1, 12, 0, 0)
+        session.get(SkillOpinionSampleRecord, missing_sample_id).created_at = datetime(
+            2024, 1, 2, 12, 0, 0
+        )
+
+    first = repo.list_candidate_keys(
+        horizons=["1d"],
+        engine_version=SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+        limit=1,
+    )
+    assert [item.sample.id for item in first] == [pending_sample_id]
+
+    repo.persist_outcome(pending_fields)
+    second = repo.list_candidate_keys(
+        horizons=["1d"],
+        engine_version=SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+        limit=1,
+    )
+    assert [item.sample.id for item in second] == [missing_sample_id]
+
+
+@pytest.mark.parametrize("existing_pending", [False, True])
+def test_failed_candidate_attempt_rotates_behind_newer_missing_candidate(
+    isolated_db,
+    monkeypatch,
+    existing_pending,
+) -> None:
+    _, failed_sample_id = _add_sample(isolated_db, skill_id="failed")
+    _, missing_sample_id = _add_sample(isolated_db, skill_id="missing")
+    repo = SkillOpinionOutcomeRepository(isolated_db)
+    if existing_pending:
+        repo.persist_outcome(
+            {
+                "skill_opinion_sample_id": failed_sample_id,
+                "horizon": "1d",
+                "engine_version": SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+                "eval_status": "pending",
+                "outcome": None,
+                "direction_correct": None,
+                "unable_reason": "insufficient_future_data",
+            }
+        )
+
+    with isolated_db.session_scope() as session:
+        session.get(SkillOpinionSampleRecord, failed_sample_id).created_at = datetime(
+            2024, 1, 1, 12, 0, 0
+        )
+        session.get(SkillOpinionSampleRecord, missing_sample_id).created_at = datetime(
+            2024, 1, 2, 12, 0, 0
+        )
+        if existing_pending:
+            session.query(SkillOpinionOutcomeRecord).filter_by(
+                skill_opinion_sample_id=failed_sample_id,
+                horizon="1d",
+                engine_version=SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+            ).one().updated_at = datetime(2024, 1, 1, 12, 0, 0)
+
+    service = SkillOpinionOutcomeService(db_manager=isolated_db)
+
+    def raise_transient_failure(_candidate):
+        raise RuntimeError("transient evaluator failure")
+
+    monkeypatch.setattr(
+        service,
+        "_evaluate_candidate",
+        raise_transient_failure,
+    )
+
+    result = service.run_outcomes(horizons=["1d"], limit=1)
+
+    assert result["failed"] == 1
+    retry_marker = _stored_outcome(isolated_db, failed_sample_id)
+    assert retry_marker is not None
+    assert retry_marker.eval_status == "pending"
+    next_candidates = repo.list_candidate_keys(
+        horizons=["1d"],
+        engine_version=SKILL_OPINION_OUTCOME_ENGINE_VERSION,
+        limit=1,
+    )
+    assert [item.sample.id for item in next_candidates] == [missing_sample_id]
+
+
 def test_history_deletion_removes_outcomes_before_samples(isolated_db) -> None:
     history_id, sample_id = _add_sample(isolated_db)
     repo = SkillOpinionOutcomeRepository(isolated_db)