Add Decision Superiority Business Quality Portfolio Intelligence

README.md

@@ -935,6 +935,90 @@ export SELF_IMPROVEMENT_ROLLBACK_ENABLED=true
 
 The learning loops only update database memory, confidence adjustments, proprietary reasoning examples and reversible scoring-weight versions.
 
+## Decision Superiority, Business Quality and Portfolio Intelligence
+
+This release upgrades BLUM from trade-outcome analysis to decision-quality analysis.
+
+The core question is no longer only:
+
+> Did this trade work?
+
+The new question is:
+
+> Was this the best available decision at that moment, considering alternative opportunities, business quality, portfolio context and benchmark alternatives?
+
+### Decision Superiority Engine
+
+`DecisionSuperiorityEngine` evaluates every comparable BLUM decision against the opportunity set available in the same decision window.
+
+It measures:
+
+- Opportunity Recall: how many future outperformers BLUM identified before they outperformed.
+- Opportunity Precision: how many selected opportunities became outperformers.
+- Alpha Capture Rate: how much available alpha BLUM captured versus what was available.
+- Ranking Accuracy: whether BLUM ranked the best future performers near the top.
+- Missed Opportunities: candidates BLUM ignored that later performed better.
+- Best Decisions and Worst Decisions: auditable evidence, not narrative claims.
+
+Main API:
+
+- `GET /api/decision-intelligence/dashboard`
+- `GET /api/decision-intelligence/superiority`
+- `POST /api/decision-intelligence/superiority/recalculate`
+- `GET /api/decision-intelligence/universe-snapshots`
+- `POST /api/decision-intelligence/universe-snapshots/recalculate`
+- `GET /api/decision-intelligence/missed-opportunities`
+
+### Business Quality Engine
+
+`BusinessQualityEngine` scores company quality from stored fundamental evidence. If fundamentals are missing, BLUM explicitly marks the company as `insufficient_fundamental_evidence` and penalizes the score.
+
+It evaluates:
+
+- growth quality;
+- profitability quality;
+- cash-flow quality;
+- balance-sheet quality;
+- capital-allocation quality;
+- moat quality;
+- management-quality proxies;
+- fundamental alpha patterns from historical outcomes.
+
+Main API:
+
+- `GET /api/business-quality/dashboard`
+- `GET /api/business-quality/scores`
+- `POST /api/business-quality/recalculate`
+
+### Portfolio Intelligence Engine
+
+`PortfolioIntelligenceEngine` measures whether a position improves the simulated portfolio, not only whether the position made money.
+
+It evaluates:
+
+- return contribution;
+- risk contribution;
+- drawdown contribution;
+- alpha contribution;
+- portfolio concentration;
+- correlation between positions;
+- position-sizing outcomes;
+- portfolio quality score.
+
+Main API:
+
+- `GET /api/portfolio-intelligence/dashboard`
+- `GET /api/portfolio-intelligence/quality`
+- `POST /api/portfolio-intelligence/recalculate`
+
+### Guardrails
+
+- No decision-superiority claim is valid with insufficient comparable samples.
+- BLUM must say when a better opportunity existed.
+- Business quality is not inferred as fact when fundamentals are missing.
+- Portfolio quality is penalized when P/L is concentrated in too few positions.
+- Chat responses use stored dashboard evidence only. If the evidence is missing, BLUM must answer `Insufficient evidence`.
+
 ## Docker
 
 ```bash

backend/alembic/versions/0019_decision_quality_engines.py

@@ -0,0 +1,309 @@
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision = "0019_decision_quality"
+down_revision = "0018_learning_intel_benchmark"
+branch_labels = None
+depends_on = None
+
+
+json_type = postgresql.JSONB(astext_type=sa.Text())
+
+
+def create_index(table: str, columns: list[str], suffix: str | None = None) -> None:
+    name = f"ix_{table}_{suffix or '_'.join(columns)}"
+    if len(name) > 63:
+        name = f"ix_{table[:30]}_{(suffix or '_'.join(columns))[:22]}"
+    op.create_index(name, table, columns)
+
+
+def upgrade() -> None:
+    op.create_table(
+        "decision_universe_snapshots",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("timestamp", sa.DateTime(), nullable=False),
+        sa.Column("market_regime", sa.String(length=120), nullable=True),
+        sa.Column("volatility_regime", sa.String(length=120), nullable=True),
+        sa.Column("selected_asset", sa.String(length=32), nullable=False),
+        sa.Column("selected_rank", sa.Integer(), nullable=True),
+        sa.Column("selected_score", sa.Float(), nullable=True),
+        sa.Column("total_candidates", sa.Integer(), nullable=False),
+        sa.Column("candidates_json", json_type, nullable=True),
+        sa.Column("benchmark_snapshot", json_type, nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("decision_universe_snapshots", ["selected_asset", "timestamp"], "selected_created")
+    create_index("decision_universe_snapshots", ["market_regime", "timestamp"], "regime_created")
+
+    op.create_table(
+        "opportunity_recall_metrics",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("sector", sa.String(length=120), nullable=True),
+        sa.Column("setup", sa.String(length=120), nullable=True),
+        sa.Column("regime", sa.String(length=120), nullable=True),
+        sa.Column("timeframe", sa.String(length=80), nullable=True),
+        sa.Column("captured_outperformers", sa.Integer(), nullable=False),
+        sa.Column("total_outperformers", sa.Integer(), nullable=False),
+        sa.Column("opportunity_recall", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("opportunity_recall_metrics", ["sector", "setup", "regime", "timeframe"], "scope")
+
+    op.create_table(
+        "opportunity_precision_metrics",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("sector", sa.String(length=120), nullable=True),
+        sa.Column("setup", sa.String(length=120), nullable=True),
+        sa.Column("regime", sa.String(length=120), nullable=True),
+        sa.Column("timeframe", sa.String(length=80), nullable=True),
+        sa.Column("successful_opportunities", sa.Integer(), nullable=False),
+        sa.Column("selected_opportunities", sa.Integer(), nullable=False),
+        sa.Column("opportunity_precision", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("opportunity_precision_metrics", ["sector", "setup", "regime", "timeframe"], "scope")
+
+    op.create_table(
+        "alpha_capture_metrics",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("ticker", sa.String(length=32), nullable=True),
+        sa.Column("sector", sa.String(length=120), nullable=True),
+        sa.Column("regime", sa.String(length=120), nullable=True),
+        sa.Column("timeframe", sa.String(length=80), nullable=True),
+        sa.Column("available_alpha", sa.Float(), nullable=True),
+        sa.Column("captured_alpha", sa.Float(), nullable=True),
+        sa.Column("alpha_capture_rate", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("alpha_capture_metrics", ["ticker", "sector", "regime", "timeframe"], "scope")
+
+    op.create_table(
+        "ranking_accuracy_metrics",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("sector", sa.String(length=120), nullable=True),
+        sa.Column("setup", sa.String(length=120), nullable=True),
+        sa.Column("regime", sa.String(length=120), nullable=True),
+        sa.Column("timeframe", sa.String(length=80), nullable=True),
+        sa.Column("sample_size", sa.Integer(), nullable=False),
+        sa.Column("top1_accuracy", sa.Float(), nullable=True),
+        sa.Column("top3_accuracy", sa.Float(), nullable=True),
+        sa.Column("top5_accuracy", sa.Float(), nullable=True),
+        sa.Column("ranking_correlation", sa.Float(), nullable=True),
+        sa.Column("ranking_decay", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("ranking_accuracy_metrics", ["sector", "setup", "regime", "timeframe"], "scope")
+
+    op.create_table(
+        "decision_superiority_scores",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("mode", sa.String(length=80), nullable=False),
+        sa.Column("scope", sa.String(length=120), nullable=False),
+        sa.Column("score", sa.Float(), nullable=False),
+        sa.Column("classification", sa.String(length=120), nullable=False),
+        sa.Column("opportunity_recall", sa.Float(), nullable=True),
+        sa.Column("opportunity_precision", sa.Float(), nullable=True),
+        sa.Column("alpha_capture", sa.Float(), nullable=True),
+        sa.Column("ranking_accuracy", sa.Float(), nullable=True),
+        sa.Column("benchmark_excess", sa.Float(), nullable=True),
+        sa.Column("live_validation", sa.Float(), nullable=True),
+        sa.Column("regime_consistency", sa.Float(), nullable=True),
+        sa.Column("reproducibility", sa.Float(), nullable=True),
+        sa.Column("drawdown_control", sa.Float(), nullable=True),
+        sa.Column("explanation", sa.Text(), nullable=False),
+        sa.Column("warnings_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("decision_superiority_scores", ["mode", "calculated_at"], "mode_created")
+    create_index("decision_superiority_scores", ["score", "classification"], "score_class")
+
+    op.create_table(
+        "business_quality_profiles",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("asset_id", sa.Integer(), nullable=True),
+        sa.Column("ticker", sa.String(length=32), nullable=False),
+        sa.Column("growth_quality", sa.Float(), nullable=True),
+        sa.Column("profitability_quality", sa.Float(), nullable=True),
+        sa.Column("cash_flow_quality", sa.Float(), nullable=True),
+        sa.Column("balance_sheet_quality", sa.Float(), nullable=True),
+        sa.Column("capital_allocation_quality", sa.Float(), nullable=True),
+        sa.Column("moat_quality", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.ForeignKeyConstraint(["asset_id"], ["assets.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("business_quality_profiles", ["ticker", "calculated_at"], "ticker_created")
+
+    op.create_table(
+        "management_quality_profiles",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("asset_id", sa.Integer(), nullable=True),
+        sa.Column("ticker", sa.String(length=32), nullable=False),
+        sa.Column("insider_alignment", sa.Float(), nullable=True),
+        sa.Column("execution_consistency", sa.Float(), nullable=True),
+        sa.Column("earnings_delivery", sa.Float(), nullable=True),
+        sa.Column("management_quality", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.ForeignKeyConstraint(["asset_id"], ["assets.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("management_quality_profiles", ["ticker", "calculated_at"], "ticker_created")
+
+    op.create_table(
+        "fundamental_alpha_patterns",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("pattern_name", sa.String(length=160), nullable=False),
+        sa.Column("sector", sa.String(length=120), nullable=True),
+        sa.Column("timeframe", sa.String(length=80), nullable=True),
+        sa.Column("sample_size", sa.Integer(), nullable=False),
+        sa.Column("average_forward_return", sa.Float(), nullable=True),
+        sa.Column("hit_rate", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("fundamental_alpha_patterns", ["pattern_name", "sector", "timeframe"], "scope")
+
+    op.create_table(
+        "business_quality_scores",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("asset_id", sa.Integer(), nullable=True),
+        sa.Column("ticker", sa.String(length=32), nullable=False),
+        sa.Column("sector", sa.String(length=120), nullable=True),
+        sa.Column("business_quality_score", sa.Float(), nullable=False),
+        sa.Column("growth_quality", sa.Float(), nullable=True),
+        sa.Column("profitability_quality", sa.Float(), nullable=True),
+        sa.Column("cash_flow_quality", sa.Float(), nullable=True),
+        sa.Column("balance_sheet_quality", sa.Float(), nullable=True),
+        sa.Column("capital_allocation_quality", sa.Float(), nullable=True),
+        sa.Column("moat_quality", sa.Float(), nullable=True),
+        sa.Column("management_quality", sa.Float(), nullable=True),
+        sa.Column("data_quality_score", sa.Float(), nullable=False),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.ForeignKeyConstraint(["asset_id"], ["assets.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("business_quality_scores", ["ticker", "calculated_at"], "ticker_created")
+    create_index("business_quality_scores", ["business_quality_score"], "score")
+
+    op.create_table(
+        "portfolio_contributions",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("game_id", sa.Integer(), nullable=True),
+        sa.Column("ticker", sa.String(length=32), nullable=False),
+        sa.Column("sector", sa.String(length=120), nullable=True),
+        sa.Column("return_contribution", sa.Float(), nullable=True),
+        sa.Column("risk_contribution", sa.Float(), nullable=True),
+        sa.Column("drawdown_contribution", sa.Float(), nullable=True),
+        sa.Column("alpha_contribution", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.ForeignKeyConstraint(["game_id"], ["trading_games.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("portfolio_contributions", ["game_id", "ticker", "calculated_at"], "scope")
+
+    op.create_table(
+        "portfolio_correlations",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("scope", sa.String(length=120), nullable=False),
+        sa.Column("asset_a", sa.String(length=32), nullable=False),
+        sa.Column("asset_b", sa.String(length=32), nullable=False),
+        sa.Column("correlation", sa.Float(), nullable=True),
+        sa.Column("correlation_type", sa.String(length=80), nullable=False),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("scope", "asset_a", "asset_b", name="uq_portfolio_correlation_pair"),
+    )
+    create_index("portfolio_correlations", ["scope", "correlation"], "scope_corr")
+
+    op.create_table(
+        "portfolio_alpha_scores",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("game_id", sa.Integer(), nullable=True),
+        sa.Column("ticker", sa.String(length=32), nullable=False),
+        sa.Column("portfolio_alpha_score", sa.Float(), nullable=False),
+        sa.Column("marginal_return_score", sa.Float(), nullable=True),
+        sa.Column("marginal_risk_score", sa.Float(), nullable=True),
+        sa.Column("diversification_score", sa.Float(), nullable=True),
+        sa.Column("benchmark_excess_score", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.ForeignKeyConstraint(["game_id"], ["trading_games.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("portfolio_alpha_scores", ["ticker", "calculated_at"], "ticker_created")
+
+    op.create_table(
+        "position_sizing_outcomes",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("sizing_logic", sa.String(length=120), nullable=False),
+        sa.Column("timeframe", sa.String(length=80), nullable=True),
+        sa.Column("sample_size", sa.Integer(), nullable=False),
+        sa.Column("average_r", sa.Float(), nullable=True),
+        sa.Column("drawdown_impact", sa.Float(), nullable=True),
+        sa.Column("capital_efficiency", sa.Float(), nullable=True),
+        sa.Column("evidence_json", json_type, nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("position_sizing_outcomes", ["sizing_logic", "timeframe"], "logic_timeframe")
+
+    op.create_table(
+        "portfolio_quality_scores",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("calculated_at", sa.DateTime(), nullable=False),
+        sa.Column("game_id", sa.Integer(), nullable=True),
+        sa.Column("portfolio_quality_score", sa.Float(), nullable=False),
+        sa.Column("diversification", sa.Float(), nullable=True),
+        sa.Column("concentration_risk", sa.Float(), nullable=True),
+        sa.Column("drawdown_control", sa.Float(), nullable=True),
+        sa.Column("alpha_generation", sa.Float(), nullable=True),
+        sa.Column("benchmark_excess", sa.Float(), nullable=True),
+        sa.Column("capital_efficiency", sa.Float(), nullable=True),
+        sa.Column("explanation", sa.Text(), nullable=False),
+        sa.Column("warnings_json", json_type, nullable=True),
+        sa.ForeignKeyConstraint(["game_id"], ["trading_games.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    create_index("portfolio_quality_scores", ["game_id", "calculated_at"], "game_created")
+    create_index("portfolio_quality_scores", ["portfolio_quality_score"], "score")
+
+
+def downgrade() -> None:
+    for table in [
+        "portfolio_quality_scores",
+        "position_sizing_outcomes",
+        "portfolio_alpha_scores",
+        "portfolio_correlations",
+        "portfolio_contributions",
+        "business_quality_scores",
+        "fundamental_alpha_patterns",
+        "management_quality_profiles",
+        "business_quality_profiles",
+        "decision_superiority_scores",
+        "ranking_accuracy_metrics",
+        "alpha_capture_metrics",
+        "opportunity_precision_metrics",
+        "opportunity_recall_metrics",
+        "decision_universe_snapshots",
+    ]:
+        op.drop_table(table)

backend/app/api/routes.py

@@ -16,6 +16,7 @@ from app.ingestion.news_ingestor import NewsIngestor
 from app.models import (
     AIInsight,
     AccuracySnapshot,
+    AlphaCaptureMetric,
     Asset,
     AutonomousEngineRun,
     BenchmarkRelativeOutcome,
@@ -32,6 +33,8 @@ from app.models import (
     BlumThesisOutcome,
     BlumThesisQualityScore,
     BlumTrainingExample,
+    BusinessQualityProfile,
+    BusinessQualityScore,
     ChartAnalysis,
     ChartPatternMemory,
     ChatMessage,
@@ -39,10 +42,13 @@ from app.models import (
     CompetingThesis,
     ConfidenceCalibrationBucket,
     ConfidenceAdjustment,
+    DecisionSuperiorityScore,
+    DecisionUniverseSnapshot,
     EmbeddingVector,
     EngineVote,
     EnsembleWeightVersion,
     ExternalDatasetSource,
+    FundamentalAlphaPattern,
     FundamentalSnapshot,
     HistoricalPrediction,
     HistoricalSimilarityCase,
@@ -95,6 +101,15 @@ from app.models import (
     MetaLearningEvent,
     LearningProgressSnapshot,
     LearningStrengthWeaknessMap,
+    ManagementQualityProfile,
+    OpportunityPrecisionMetric,
+    OpportunityRecallMetric,
+    PortfolioAlphaScore,
+    PortfolioContribution,
+    PortfolioCorrelation,
+    PortfolioQualityScore,
+    PositionSizingOutcome,
+    RankingAccuracyMetric,
     SelfImprovementAction,
     ThesisCompetition,
     ThesisConvictionHistory,
@@ -155,6 +170,12 @@ from app.services.learning_intelligence import (
     LearningWeaknessMapService,
     SelfImprovementActionEngine,
 )
+from app.services.decision_intelligence import (
+    BusinessQualityEngine,
+    DecisionIntelligenceDashboardService,
+    DecisionSuperiorityEngine,
+    PortfolioIntelligenceEngine,
+)
 from app.services.live import live_news, market_sentiment
 from app.services.macro import macro_overview, update_macro_snapshots
 from app.services.market_brain import build_market_brain, latest_market_brain, market_brain_history
@@ -339,6 +360,9 @@ def system_status(db: Session = Depends(get_db)) -> dict:
             "official_benchmark_comparison": True,
             "learning_weakness_map": True,
             "self_improvement_action_engine": True,
+            "decision_superiority_engine": True,
+            "business_quality_engine": True,
+            "portfolio_intelligence_engine": True,
         },
         "database_counts": {
             "assets": int(db.scalar(select(func.count(Asset.id))) or 0),
@@ -386,6 +410,21 @@ def system_status(db: Session = Depends(get_db)) -> dict:
             "learning_progress_snapshots": int(db.scalar(select(func.count(LearningProgressSnapshot.id))) or 0),
             "learning_strength_weakness_map": int(db.scalar(select(func.count(LearningStrengthWeaknessMap.id))) or 0),
             "self_improvement_actions": int(db.scalar(select(func.count(SelfImprovementAction.id))) or 0),
+            "decision_universe_snapshots": int(db.scalar(select(func.count(DecisionUniverseSnapshot.id))) or 0),
+            "opportunity_recall_metrics": int(db.scalar(select(func.count(OpportunityRecallMetric.id))) or 0),
+            "opportunity_precision_metrics": int(db.scalar(select(func.count(OpportunityPrecisionMetric.id))) or 0),
+            "alpha_capture_metrics": int(db.scalar(select(func.count(AlphaCaptureMetric.id))) or 0),
+            "ranking_accuracy_metrics": int(db.scalar(select(func.count(RankingAccuracyMetric.id))) or 0),
+            "decision_superiority_scores": int(db.scalar(select(func.count(DecisionSuperiorityScore.id))) or 0),
+            "business_quality_profiles": int(db.scalar(select(func.count(BusinessQualityProfile.id))) or 0),
+            "management_quality_profiles": int(db.scalar(select(func.count(ManagementQualityProfile.id))) or 0),
+            "fundamental_alpha_patterns": int(db.scalar(select(func.count(FundamentalAlphaPattern.id))) or 0),
+            "business_quality_scores": int(db.scalar(select(func.count(BusinessQualityScore.id))) or 0),
+            "portfolio_contributions": int(db.scalar(select(func.count(PortfolioContribution.id))) or 0),
+            "portfolio_correlations": int(db.scalar(select(func.count(PortfolioCorrelation.id))) or 0),
+            "portfolio_alpha_scores": int(db.scalar(select(func.count(PortfolioAlphaScore.id))) or 0),
+            "position_sizing_outcomes": int(db.scalar(select(func.count(PositionSizingOutcome.id))) or 0),
+            "portfolio_quality_scores": int(db.scalar(select(func.count(PortfolioQualityScore.id))) or 0),
             "model_weight_versions": int(db.scalar(select(func.count(ModelWeightVersion.id))) or 0),
             "historical_similarity_cases": int(db.scalar(select(func.count(HistoricalSimilarityCase.id))) or 0),
             "confidence_adjustments": int(db.scalar(select(func.count(ConfidenceAdjustment.id))) or 0),
@@ -431,7 +470,7 @@ def system_status(db: Session = Depends(get_db)) -> dict:
             "Hugging Face serves the previous image until the Docker build finishes successfully.",
             "The finance-domain 7B model is disabled by default unless BLUM_ENABLE_FINANCIAL_BRAIN_MODEL=true.",
             "Existing snapshots are refreshed by the autonomous engine after a successful deployment.",
-            "Browser cache can keep old static Next.js chunks; hard refresh if app_version is not 0.16.0.",
+            "Browser cache can keep old static Next.js chunks; hard refresh if app_version is not 0.17.0.",
         ],
     }
 
@@ -1033,6 +1072,74 @@ def learning_intelligence_evaluate_self_improvement(action_id: int, db: Session
     return SelfImprovementActionEngine().evaluate(db, action_id=action_id)
 
 
+@router.get("/api/decision-intelligence/dashboard")
+def decision_intelligence_dashboard(db: Session = Depends(get_db)) -> dict:
+    return DecisionIntelligenceDashboardService().dashboard(db)
+
+
+@router.get("/api/decision-intelligence/superiority")
+def decision_intelligence_superiority(db: Session = Depends(get_db)) -> dict:
+    return DecisionSuperiorityEngine().score(db, persist=False)
+
+
+@router.post("/api/decision-intelligence/superiority/recalculate")
+def decision_intelligence_superiority_recalculate(db: Session = Depends(get_db)) -> dict:
+    return DecisionSuperiorityEngine().score(db, persist=True)
+
+
+@router.get("/api/decision-intelligence/universe-snapshots")
+def decision_intelligence_universe_snapshots(db: Session = Depends(get_db)) -> dict:
+    return DecisionSuperiorityEngine().universe_snapshots(db, persist=False)
+
+
+@router.post("/api/decision-intelligence/universe-snapshots/recalculate")
+def decision_intelligence_universe_snapshots_recalculate(db: Session = Depends(get_db)) -> dict:
+    return DecisionSuperiorityEngine().universe_snapshots(db, persist=True)
+
+
+@router.get("/api/decision-intelligence/missed-opportunities")
+def decision_intelligence_missed_opportunities(db: Session = Depends(get_db)) -> list[dict]:
+    return DecisionSuperiorityEngine().top_missed_opportunities(db)
+
+
+@router.get("/api/business-quality/dashboard")
+def business_quality_dashboard(db: Session = Depends(get_db)) -> dict:
+    return BusinessQualityEngine().dashboard(db)
+
+
+@router.get("/api/business-quality/scores")
+def business_quality_scores(limit: int = Query(default=80, ge=1, le=300), db: Session = Depends(get_db)) -> dict:
+    return BusinessQualityEngine().scores(db, limit=limit, persist=False)
+
+
+@router.post("/api/business-quality/recalculate")
+def business_quality_recalculate(limit: int = Query(default=80, ge=1, le=300), db: Session = Depends(get_db)) -> dict:
+    return BusinessQualityEngine().scores(db, limit=limit, persist=True)
+
+
+@router.get("/api/portfolio-intelligence/dashboard")
+def portfolio_intelligence_dashboard(db: Session = Depends(get_db)) -> dict:
+    return PortfolioIntelligenceEngine().dashboard(db)
+
+
+@router.get("/api/portfolio-intelligence/quality")
+def portfolio_intelligence_quality(db: Session = Depends(get_db)) -> dict:
+    return PortfolioIntelligenceEngine().quality_score(db, persist=False)
+
+
+@router.post("/api/portfolio-intelligence/recalculate")
+def portfolio_intelligence_recalculate(db: Session = Depends(get_db)) -> dict:
+    engine = PortfolioIntelligenceEngine()
+    return {
+        "status": "ok",
+        "quality": engine.quality_score(db, persist=True),
+        "contributions": engine.contributions(db, persist=True),
+        "correlations": engine.correlations(db, persist=True),
+        "portfolio_alpha": engine.alpha_scores(db, persist=True),
+        "position_sizing": engine.position_sizing_outcomes(db, persist=True),
+    }
+
+
 @router.get("/model/status")
 def blum_model_status(db: Session = Depends(get_db)) -> dict:
     return model_status(db)

backend/app/core/config.py

@@ -5,7 +5,7 @@ from pydantic_settings import BaseSettings
 
 class Settings(BaseSettings):
     app_name: str = "Blum AI Financial Intelligence"
-    app_version: str = "0.16.0"
+    app_version: str = "0.17.0"
     environment: str = Field(default="demo", alias="ENVIRONMENT")
     database_url: str = Field(
         default="postgresql+psycopg2://postgres:postgres@127.0.0.1:5432/blum",

backend/app/models.py

@@ -2420,3 +2420,284 @@ class SelfImprovementAction(Base):
     after_metric: Mapped[float | None] = mapped_column(Float)
     improvement_observed: Mapped[bool | None] = mapped_column(Boolean, index=True)
     notes_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class DecisionUniverseSnapshot(Base):
+    __tablename__ = "decision_universe_snapshots"
+    __table_args__ = (
+        Index("ix_decision_universe_selected_created", "selected_asset", "timestamp"),
+        Index("ix_decision_universe_regime_created", "market_regime", "timestamp"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    timestamp: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    market_regime: Mapped[str | None] = mapped_column(String(120), index=True)
+    volatility_regime: Mapped[str | None] = mapped_column(String(120), index=True)
+    selected_asset: Mapped[str] = mapped_column(String(32), index=True)
+    selected_rank: Mapped[int | None] = mapped_column(Integer)
+    selected_score: Mapped[float | None] = mapped_column(Float)
+    total_candidates: Mapped[int] = mapped_column(Integer, default=0)
+    candidates_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+    benchmark_snapshot: Mapped[dict] = mapped_column(JsonType, default=dict)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+
+
+class OpportunityRecallMetric(Base):
+    __tablename__ = "opportunity_recall_metrics"
+    __table_args__ = (Index("ix_opportunity_recall_scope", "sector", "setup", "regime", "timeframe"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    sector: Mapped[str | None] = mapped_column(String(120), index=True)
+    setup: Mapped[str | None] = mapped_column(String(120), index=True)
+    regime: Mapped[str | None] = mapped_column(String(120), index=True)
+    timeframe: Mapped[str | None] = mapped_column(String(80), index=True)
+    captured_outperformers: Mapped[int] = mapped_column(Integer, default=0)
+    total_outperformers: Mapped[int] = mapped_column(Integer, default=0)
+    opportunity_recall: Mapped[float | None] = mapped_column(Float, index=True)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class OpportunityPrecisionMetric(Base):
+    __tablename__ = "opportunity_precision_metrics"
+    __table_args__ = (Index("ix_opportunity_precision_scope", "sector", "setup", "regime", "timeframe"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    sector: Mapped[str | None] = mapped_column(String(120), index=True)
+    setup: Mapped[str | None] = mapped_column(String(120), index=True)
+    regime: Mapped[str | None] = mapped_column(String(120), index=True)
+    timeframe: Mapped[str | None] = mapped_column(String(80), index=True)
+    successful_opportunities: Mapped[int] = mapped_column(Integer, default=0)
+    selected_opportunities: Mapped[int] = mapped_column(Integer, default=0)
+    opportunity_precision: Mapped[float | None] = mapped_column(Float, index=True)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class AlphaCaptureMetric(Base):
+    __tablename__ = "alpha_capture_metrics"
+    __table_args__ = (Index("ix_alpha_capture_scope", "ticker", "sector", "regime", "timeframe"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    ticker: Mapped[str | None] = mapped_column(String(32), index=True)
+    sector: Mapped[str | None] = mapped_column(String(120), index=True)
+    regime: Mapped[str | None] = mapped_column(String(120), index=True)
+    timeframe: Mapped[str | None] = mapped_column(String(80), index=True)
+    available_alpha: Mapped[float | None] = mapped_column(Float)
+    captured_alpha: Mapped[float | None] = mapped_column(Float)
+    alpha_capture_rate: Mapped[float | None] = mapped_column(Float, index=True)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class RankingAccuracyMetric(Base):
+    __tablename__ = "ranking_accuracy_metrics"
+    __table_args__ = (Index("ix_ranking_accuracy_scope", "sector", "setup", "regime", "timeframe"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    sector: Mapped[str | None] = mapped_column(String(120), index=True)
+    setup: Mapped[str | None] = mapped_column(String(120), index=True)
+    regime: Mapped[str | None] = mapped_column(String(120), index=True)
+    timeframe: Mapped[str | None] = mapped_column(String(80), index=True)
+    sample_size: Mapped[int] = mapped_column(Integer, default=0, index=True)
+    top1_accuracy: Mapped[float | None] = mapped_column(Float)
+    top3_accuracy: Mapped[float | None] = mapped_column(Float)
+    top5_accuracy: Mapped[float | None] = mapped_column(Float)
+    ranking_correlation: Mapped[float | None] = mapped_column(Float)
+    ranking_decay: Mapped[float | None] = mapped_column(Float)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class DecisionSuperiorityScore(Base):
+    __tablename__ = "decision_superiority_scores"
+    __table_args__ = (
+        Index("ix_decision_superiority_mode_created", "mode", "calculated_at"),
+        Index("ix_decision_superiority_score", "score", "classification"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    mode: Mapped[str] = mapped_column(String(80), default="historical_plus_live", index=True)
+    scope: Mapped[str] = mapped_column(String(120), default="global", index=True)
+    score: Mapped[float] = mapped_column(Float, default=0.0, index=True)
+    classification: Mapped[str] = mapped_column(String(120), default="Weak", index=True)
+    opportunity_recall: Mapped[float | None] = mapped_column(Float)
+    opportunity_precision: Mapped[float | None] = mapped_column(Float)
+    alpha_capture: Mapped[float | None] = mapped_column(Float)
+    ranking_accuracy: Mapped[float | None] = mapped_column(Float)
+    benchmark_excess: Mapped[float | None] = mapped_column(Float)
+    live_validation: Mapped[float | None] = mapped_column(Float)
+    regime_consistency: Mapped[float | None] = mapped_column(Float)
+    reproducibility: Mapped[float | None] = mapped_column(Float)
+    drawdown_control: Mapped[float | None] = mapped_column(Float)
+    explanation: Mapped[str] = mapped_column(Text, default="")
+    warnings_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class BusinessQualityProfile(Base):
+    __tablename__ = "business_quality_profiles"
+    __table_args__ = (Index("ix_business_quality_profiles_ticker_created", "ticker", "calculated_at"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    asset_id: Mapped[int | None] = mapped_column(ForeignKey("assets.id", ondelete="SET NULL"), index=True)
+    ticker: Mapped[str] = mapped_column(String(32), index=True)
+    growth_quality: Mapped[float | None] = mapped_column(Float)
+    profitability_quality: Mapped[float | None] = mapped_column(Float)
+    cash_flow_quality: Mapped[float | None] = mapped_column(Float)
+    balance_sheet_quality: Mapped[float | None] = mapped_column(Float)
+    capital_allocation_quality: Mapped[float | None] = mapped_column(Float)
+    moat_quality: Mapped[float | None] = mapped_column(Float)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+    asset = relationship("Asset")
+
+
+class ManagementQualityProfile(Base):
+    __tablename__ = "management_quality_profiles"
+    __table_args__ = (Index("ix_management_quality_profiles_ticker_created", "ticker", "calculated_at"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    asset_id: Mapped[int | None] = mapped_column(ForeignKey("assets.id", ondelete="SET NULL"), index=True)
+    ticker: Mapped[str] = mapped_column(String(32), index=True)
+    insider_alignment: Mapped[float | None] = mapped_column(Float)
+    execution_consistency: Mapped[float | None] = mapped_column(Float)
+    earnings_delivery: Mapped[float | None] = mapped_column(Float)
+    management_quality: Mapped[float | None] = mapped_column(Float, index=True)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+    asset = relationship("Asset")
+
+
+class FundamentalAlphaPattern(Base):
+    __tablename__ = "fundamental_alpha_patterns"
+    __table_args__ = (Index("ix_fundamental_alpha_pattern_scope", "pattern_name", "sector", "timeframe"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    pattern_name: Mapped[str] = mapped_column(String(160), index=True)
+    sector: Mapped[str | None] = mapped_column(String(120), index=True)
+    timeframe: Mapped[str | None] = mapped_column(String(80), index=True)
+    sample_size: Mapped[int] = mapped_column(Integer, default=0, index=True)
+    average_forward_return: Mapped[float | None] = mapped_column(Float)
+    hit_rate: Mapped[float | None] = mapped_column(Float)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class BusinessQualityScore(Base):
+    __tablename__ = "business_quality_scores"
+    __table_args__ = (
+        Index("ix_business_quality_scores_ticker_created", "ticker", "calculated_at"),
+        Index("ix_business_quality_scores_score", "business_quality_score"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    asset_id: Mapped[int | None] = mapped_column(ForeignKey("assets.id", ondelete="SET NULL"), index=True)
+    ticker: Mapped[str] = mapped_column(String(32), index=True)
+    sector: Mapped[str | None] = mapped_column(String(120), index=True)
+    business_quality_score: Mapped[float] = mapped_column(Float, default=0.0, index=True)
+    growth_quality: Mapped[float | None] = mapped_column(Float)
+    profitability_quality: Mapped[float | None] = mapped_column(Float)
+    cash_flow_quality: Mapped[float | None] = mapped_column(Float)
+    balance_sheet_quality: Mapped[float | None] = mapped_column(Float)
+    capital_allocation_quality: Mapped[float | None] = mapped_column(Float)
+    moat_quality: Mapped[float | None] = mapped_column(Float)
+    management_quality: Mapped[float | None] = mapped_column(Float)
+    data_quality_score: Mapped[float] = mapped_column(Float, default=0.0)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+    asset = relationship("Asset")
+
+
+class PortfolioContribution(Base):
+    __tablename__ = "portfolio_contributions"
+    __table_args__ = (Index("ix_portfolio_contributions_scope", "game_id", "ticker", "calculated_at"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    game_id: Mapped[int | None] = mapped_column(ForeignKey("trading_games.id", ondelete="CASCADE"), index=True)
+    ticker: Mapped[str] = mapped_column(String(32), index=True)
+    sector: Mapped[str | None] = mapped_column(String(120), index=True)
+    return_contribution: Mapped[float | None] = mapped_column(Float)
+    risk_contribution: Mapped[float | None] = mapped_column(Float)
+    drawdown_contribution: Mapped[float | None] = mapped_column(Float)
+    alpha_contribution: Mapped[float | None] = mapped_column(Float)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+    game = relationship("TradingGame")
+
+
+class PortfolioCorrelation(Base):
+    __tablename__ = "portfolio_correlations"
+    __table_args__ = (
+        UniqueConstraint("scope", "asset_a", "asset_b", name="uq_portfolio_correlation_pair"),
+        Index("ix_portfolio_correlations_scope", "scope", "correlation"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    scope: Mapped[str] = mapped_column(String(120), default="active_game", index=True)
+    asset_a: Mapped[str] = mapped_column(String(32), index=True)
+    asset_b: Mapped[str] = mapped_column(String(32), index=True)
+    correlation: Mapped[float | None] = mapped_column(Float, index=True)
+    correlation_type: Mapped[str] = mapped_column(String(80), default="price_return", index=True)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class PortfolioAlphaScore(Base):
+    __tablename__ = "portfolio_alpha_scores"
+    __table_args__ = (Index("ix_portfolio_alpha_scores_ticker_created", "ticker", "calculated_at"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    game_id: Mapped[int | None] = mapped_column(ForeignKey("trading_games.id", ondelete="CASCADE"), index=True)
+    ticker: Mapped[str] = mapped_column(String(32), index=True)
+    portfolio_alpha_score: Mapped[float] = mapped_column(Float, default=0.0, index=True)
+    marginal_return_score: Mapped[float | None] = mapped_column(Float)
+    marginal_risk_score: Mapped[float | None] = mapped_column(Float)
+    diversification_score: Mapped[float | None] = mapped_column(Float)
+    benchmark_excess_score: Mapped[float | None] = mapped_column(Float)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+    game = relationship("TradingGame")
+
+
+class PositionSizingOutcome(Base):
+    __tablename__ = "position_sizing_outcomes"
+    __table_args__ = (Index("ix_position_sizing_outcomes_logic", "sizing_logic", "timeframe"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    sizing_logic: Mapped[str] = mapped_column(String(120), index=True)
+    timeframe: Mapped[str | None] = mapped_column(String(80), index=True)
+    sample_size: Mapped[int] = mapped_column(Integer, default=0, index=True)
+    average_r: Mapped[float | None] = mapped_column(Float)
+    drawdown_impact: Mapped[float | None] = mapped_column(Float)
+    capital_efficiency: Mapped[float | None] = mapped_column(Float)
+    evidence_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+
+class PortfolioQualityScore(Base):
+    __tablename__ = "portfolio_quality_scores"
+    __table_args__ = (
+        Index("ix_portfolio_quality_game_created", "game_id", "calculated_at"),
+        Index("ix_portfolio_quality_score", "portfolio_quality_score"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    calculated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
+    game_id: Mapped[int | None] = mapped_column(ForeignKey("trading_games.id", ondelete="CASCADE"), index=True)
+    portfolio_quality_score: Mapped[float] = mapped_column(Float, default=0.0, index=True)
+    diversification: Mapped[float | None] = mapped_column(Float)
+    concentration_risk: Mapped[float | None] = mapped_column(Float)
+    drawdown_control: Mapped[float | None] = mapped_column(Float)
+    alpha_generation: Mapped[float | None] = mapped_column(Float)
+    benchmark_excess: Mapped[float | None] = mapped_column(Float)
+    capital_efficiency: Mapped[float | None] = mapped_column(Float)
+    explanation: Mapped[str] = mapped_column(Text, default="")
+    warnings_json: Mapped[dict] = mapped_column(JsonType, default=dict)
+
+    game = relationship("TradingGame")

backend/app/services/decision_intelligence.py

@@ -0,0 +1,927 @@
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+from datetime import date, datetime, timedelta
+from math import sqrt
+from statistics import mean, median, pstdev
+
+from sqlalchemy import desc, select
+from sqlalchemy.orm import Session
+
+from app.models import (
+    AlphaCaptureMetric,
+    Asset,
+    BusinessQualityProfile,
+    BusinessQualityScore,
+    DecisionSuperiorityScore,
+    DecisionUniverseSnapshot,
+    FundamentalAlphaPattern,
+    FundamentalSnapshot,
+    ManagementQualityProfile,
+    OpportunityPrecisionMetric,
+    OpportunityRecallMetric,
+    PortfolioAlphaScore,
+    PortfolioContribution,
+    PortfolioCorrelation,
+    PortfolioQualityScore,
+    PositionSizingOutcome,
+    PriceHistory,
+    RankingAccuracyMetric,
+    SignalSnapshot,
+    TradingGame,
+    TradingGameTrade,
+)
+from app.services.learning_intelligence import (
+    BenchmarkComparisonService,
+    game_trades,
+    latest_trading_game,
+    round_or_none,
+    statistical_confidence_label,
+)
+from app.services.trade_transparency import clamp, safe_float
+from app.services.trading_intelligence_lab import executable_trades, metric_payload, sample_context
+
+
+DECISION_INTELLIGENCE_POLICY = (
+    "Decision Intelligence measures selection quality against available alternatives, business quality and portfolio context. "
+    "It must report insufficient evidence instead of claiming decision superiority without comparable samples."
+)
+
+
+class DecisionSuperiorityEngine:
+    """Measures whether BLUM selected the best available opportunity, not only whether a trade worked."""
+
+    def dashboard(self, db: Session) -> dict:
+        return {
+            "status": "ok",
+            "policy": DECISION_INTELLIGENCE_POLICY,
+            "decision_superiority": self.score(db, persist=False),
+            "universe_snapshots": self.universe_snapshots(db, persist=False),
+            "top_missed_opportunities": self.top_missed_opportunities(db),
+            "best_decisions": self.best_decisions(db),
+            "worst_decisions": self.worst_decisions(db),
+        }
+
+    def score(self, db: Session, persist: bool = False) -> dict:
+        game = latest_trading_game(db)
+        rows = executable_trades(game_trades(db, game.id if game else None))
+        decisions = [decision_evidence_for_trade(db, row) for row in rows]
+        decisions = [item for item in decisions if item["candidate_count"] >= 2]
+        metrics = decision_superiority_metrics(decisions, rows)
+        components = decision_superiority_components(metrics, rows)
+        score = weighted_average(
+            [
+                components["opportunity_recall"],
+                components["opportunity_precision"],
+                components["alpha_capture"],
+                components["ranking_accuracy"],
+                components["benchmark_excess"],
+                components["live_validation"],
+                components["regime_consistency"],
+                components["reproducibility"],
+                components["drawdown_control"],
+            ]
+        )
+        warnings = decision_superiority_warnings(metrics, rows)
+        classification = classify_decision_superiority(score)
+        explanation = decision_superiority_explanation(score, classification, metrics, warnings)
+        payload = {
+            "status": "ok" if game else "no_game",
+            "calculated_at": datetime.utcnow().isoformat(),
+            "mode": "historical_plus_live",
+            "scope": "global",
+            "score": round(score, 2),
+            "classification": classification,
+            "components": components,
+            "metrics": metrics,
+            "warnings": warnings,
+            "sample_size": len(decisions),
+            "statistical_confidence": statistical_confidence_label(len(decisions), count_live(rows), sample_context(rows)),
+            "explanation": explanation,
+            "policy": DECISION_INTELLIGENCE_POLICY,
+        }
+        if persist:
+            db.add(
+                OpportunityRecallMetric(
+                    sector="All",
+                    setup="All",
+                    regime="All",
+                    timeframe="All",
+                    captured_outperformers=metrics.get("captured_outperformers", 0),
+                    total_outperformers=metrics.get("total_outperformers", 0),
+                    opportunity_recall=metrics.get("opportunity_recall"),
+                    evidence_json=metrics,
+                )
+            )
+            db.add(
+                OpportunityPrecisionMetric(
+                    sector="All",
+                    setup="All",
+                    regime="All",
+                    timeframe="All",
+                    successful_opportunities=metrics.get("successful_opportunities", 0),
+                    selected_opportunities=metrics.get("selected_opportunities", 0),
+                    opportunity_precision=metrics.get("opportunity_precision"),
+                    evidence_json=metrics,
+                )
+            )
+            db.add(
+                AlphaCaptureMetric(
+                    ticker="All",
+                    sector="All",
+                    regime="All",
+                    timeframe="All",
+                    available_alpha=metrics.get("available_alpha"),
+                    captured_alpha=metrics.get("captured_alpha"),
+                    alpha_capture_rate=metrics.get("alpha_capture_rate"),
+                    evidence_json=metrics,
+                )
+            )
+            db.add(
+                RankingAccuracyMetric(
+                    sector="All",
+                    setup="All",
+                    regime="All",
+                    timeframe="All",
+                    sample_size=metrics.get("sample_size", 0),
+                    top1_accuracy=metrics.get("top1_accuracy"),
+                    top3_accuracy=None,
+                    top5_accuracy=None,
+                    ranking_correlation=metrics.get("ranking_accuracy"),
+                    ranking_decay=None,
+                    evidence_json=metrics,
+                )
+            )
+            db.add(
+                DecisionSuperiorityScore(
+                    mode=payload["mode"],
+                    scope=payload["scope"],
+                    score=payload["score"],
+                    classification=classification,
+                    opportunity_recall=metrics.get("opportunity_recall"),
+                    opportunity_precision=metrics.get("opportunity_precision"),
+                    alpha_capture=metrics.get("alpha_capture_rate"),
+                    ranking_accuracy=metrics.get("ranking_accuracy"),
+                    benchmark_excess=metrics.get("benchmark_excess"),
+                    live_validation=components["live_validation"],
+                    regime_consistency=components["regime_consistency"],
+                    reproducibility=components["reproducibility"],
+                    drawdown_control=components["drawdown_control"],
+                    explanation=explanation,
+                    warnings_json={"warnings": warnings, "metrics": metrics},
+                )
+            )
+            db.commit()
+        return payload
+
+    def universe_snapshots(self, db: Session, persist: bool = False) -> dict:
+        game = latest_trading_game(db)
+        rows = executable_trades(game_trades(db, game.id if game else None))[:80]
+        snapshots = [decision_universe_snapshot_payload(db, row) for row in rows]
+        if persist:
+            for item in snapshots:
+                db.add(
+                    DecisionUniverseSnapshot(
+                        timestamp=parse_dt(item["timestamp"]) or datetime.utcnow(),
+                        market_regime=item.get("market_regime"),
+                        volatility_regime=item.get("volatility_regime"),
+                        selected_asset=item["selected_asset"],
+                        selected_rank=item.get("selected_rank"),
+                        selected_score=item.get("selected_score"),
+                        total_candidates=item.get("total_candidates", 0),
+                        candidates_json={"candidates": item.get("candidates", [])},
+                        benchmark_snapshot=item.get("benchmark_snapshot", {}),
+                    )
+                )
+            db.commit()
+        return {"status": "ok" if game else "no_game", "snapshots": snapshots, "policy": DECISION_INTELLIGENCE_POLICY}
+
+    def top_missed_opportunities(self, db: Session) -> list[dict]:
+        rows = executable_trades(game_trades(db, latest_trading_game(db).id if latest_trading_game(db) else None))
+        missed = []
+        for row in rows:
+            evidence = decision_evidence_for_trade(db, row)
+            if evidence.get("missed_best"):
+                missed.append(evidence)
+        return sorted(missed, key=lambda item: safe_float(item.get("opportunity_gap")), reverse=True)[:12]
+
+    def best_decisions(self, db: Session) -> list[dict]:
+        rows = executable_trades(game_trades(db, latest_trading_game(db).id if latest_trading_game(db) else None))
+        evidence = [decision_evidence_for_trade(db, row) for row in rows]
+        return sorted(evidence, key=lambda item: safe_float(item.get("selection_quality")), reverse=True)[:12]
+
+    def worst_decisions(self, db: Session) -> list[dict]:
+        rows = executable_trades(game_trades(db, latest_trading_game(db).id if latest_trading_game(db) else None))
+        evidence = [decision_evidence_for_trade(db, row) for row in rows]
+        return sorted(evidence, key=lambda item: safe_float(item.get("selection_quality")))[:12]
+
+
+class BusinessQualityEngine:
+    """Scores business quality from stored fundamental evidence and clearly penalizes missing data."""
+
+    def dashboard(self, db: Session, limit: int = 40) -> dict:
+        rows = self.scores(db, limit=limit, persist=False)["rows"]
+        return {
+            "status": "ok",
+            "policy": DECISION_INTELLIGENCE_POLICY,
+            "highest_quality_companies": rows[:12],
+            "strongest_moats": sorted(rows, key=lambda item: safe_float(item.get("moat_quality")), reverse=True)[:12],
+            "best_capital_allocation": sorted(rows, key=lambda item: safe_float(item.get("capital_allocation_quality")), reverse=True)[:12],
+            "highest_fundamental_alpha_score": sorted(rows, key=lambda item: safe_float(item.get("fundamental_alpha_score")), reverse=True)[:12],
+            "improving_businesses": [row for row in rows if row.get("trend_label") == "improving"][:12],
+            "deteriorating_businesses": [row for row in rows if row.get("trend_label") == "deteriorating"][:12],
+            "fundamental_alpha_patterns": self.fundamental_alpha_patterns(db, persist=False)["rows"],
+        }
+
+    def scores(self, db: Session, limit: int = 80, persist: bool = False) -> dict:
+        assets = db.scalars(
+            select(Asset)
+            .where(Asset.asset_type == "Stock", Asset.is_active.is_(True))
+            .order_by(Asset.ticker)
+            .limit(limit)
+        ).all()
+        rows = [business_quality_for_asset(db, asset) for asset in assets]
+        rows = sorted(rows, key=lambda item: safe_float(item.get("business_quality_score")), reverse=True)
+        if persist:
+            for item in rows:
+                asset = db.scalar(select(Asset).where(Asset.ticker == item["ticker"]))
+                db.add(
+                    BusinessQualityProfile(
+                        asset_id=asset.id if asset else None,
+                        ticker=item["ticker"],
+                        growth_quality=item.get("growth_quality"),
+                        profitability_quality=item.get("profitability_quality"),
+                        cash_flow_quality=item.get("cash_flow_quality"),
+                        balance_sheet_quality=item.get("balance_sheet_quality"),
+                        capital_allocation_quality=item.get("capital_allocation_quality"),
+                        moat_quality=item.get("moat_quality"),
+                        evidence_json=item.get("evidence", {}),
+                    )
+                )
+                db.add(
+                    ManagementQualityProfile(
+                        asset_id=asset.id if asset else None,
+                        ticker=item["ticker"],
+                        insider_alignment=item.get("management_components", {}).get("insider_alignment"),
+                        execution_consistency=item.get("management_components", {}).get("execution_consistency"),
+                        earnings_delivery=item.get("management_components", {}).get("earnings_delivery"),
+                        management_quality=item.get("management_quality"),
+                        evidence_json=item.get("management_components", {}),
+                    )
+                )
+                db.add(
+                    BusinessQualityScore(
+                        asset_id=asset.id if asset else None,
+                        ticker=item["ticker"],
+                        sector=item.get("sector"),
+                        business_quality_score=item["business_quality_score"],
+                        growth_quality=item.get("growth_quality"),
+                        profitability_quality=item.get("profitability_quality"),
+                        cash_flow_quality=item.get("cash_flow_quality"),
+                        balance_sheet_quality=item.get("balance_sheet_quality"),
+                        capital_allocation_quality=item.get("capital_allocation_quality"),
+                        moat_quality=item.get("moat_quality"),
+                        management_quality=item.get("management_quality"),
+                        data_quality_score=item.get("data_quality_score", 0),
+                        evidence_json=item.get("evidence", {}),
+                    )
+                )
+            db.commit()
+        return {"status": "ok", "rows": rows, "policy": DECISION_INTELLIGENCE_POLICY}
+
+    def fundamental_alpha_patterns(self, db: Session, persist: bool = False) -> dict:
+        rows = []
+        trades = executable_trades(game_trades(db, latest_trading_game(db).id if latest_trading_game(db) else None))
+        by_sector: dict[str, list[TradingGameTrade]] = defaultdict(list)
+        for trade in trades:
+            by_sector[trade.sector or "Unknown"].append(trade)
+        for sector, group in by_sector.items():
+            returns = [safe_float(row.pnl_percent if row.pnl_percent is not None else row.excess_return_vs_benchmark) for row in group]
+            rows.append(
+                {
+                    "pattern_name": "quality_plus_positive_trade_outcome",
+                    "sector": sector,
+                    "timeframe": "trade_horizon",
+                    "sample_size": len(group),
+                    "average_forward_return": round_or_none(mean(returns) if returns else None),
+                    "hit_rate": round_or_none(sum(1 for value in returns if value > 0) / max(1, len(returns))),
+                    "evidence": {"tickers": sorted({row.ticker for row in group})[:20]},
+                }
+            )
+        if persist:
+            for item in rows:
+                db.add(
+                    FundamentalAlphaPattern(
+                        pattern_name=item["pattern_name"],
+                        sector=item.get("sector"),
+                        timeframe=item.get("timeframe"),
+                        sample_size=item["sample_size"],
+                        average_forward_return=item.get("average_forward_return"),
+                        hit_rate=item.get("hit_rate"),
+                        evidence_json=item.get("evidence", {}),
+                    )
+                )
+            db.commit()
+        return {"status": "ok", "rows": rows, "policy": DECISION_INTELLIGENCE_POLICY}
+
+
+class PortfolioIntelligenceEngine:
+    """Measures whether individual decisions improve the simulated portfolio."""
+
+    def dashboard(self, db: Session) -> dict:
+        return {
+            "status": "ok",
+            "policy": DECISION_INTELLIGENCE_POLICY,
+            "portfolio_quality": self.quality_score(db, persist=False),
+            "contributions": self.contributions(db, persist=False)["rows"],
+            "correlations": self.correlations(db, persist=False)["rows"],
+            "portfolio_alpha": self.alpha_scores(db, persist=False)["rows"],
+            "position_sizing": self.position_sizing_outcomes(db, persist=False)["rows"],
+        }
+
+    def quality_score(self, db: Session, persist: bool = False) -> dict:
+        game = latest_trading_game(db)
+        rows = executable_trades(game_trades(db, game.id if game else None))
+        metrics = metric_payload(rows, "portfolio", str(game.id) if game else None, "all", None)
+        contributions = portfolio_contribution_rows(rows)
+        concentration = concentration_score(contributions)
+        diversification = clamp(100 - concentration)
+        drawdown_control = clamp(100 + safe_float(metrics.get("max_drawdown")) * 3)
+        alpha_generation = clamp(50 + safe_float(metrics.get("benchmark_excess")) * 2)
+        capital_efficiency = clamp(50 + safe_float(metrics.get("expectancy_r")) * 25)
+        score = weighted_average([diversification, drawdown_control, alpha_generation, capital_efficiency, safe_float(metrics.get("risk_reward_quality_score"), 50)])
+        warnings = portfolio_warnings(rows, concentration, metrics)
+        explanation = f"Portfolio Quality Score {score:.1f}/100. Diversification {diversification:.1f}, drawdown control {drawdown_control:.1f}, alpha generation {alpha_generation:.1f}."
+        payload = {
+            "status": "ok" if game else "no_game",
+            "score": round(score, 2),
+            "components": {
+                "diversification": round(diversification, 2),
+                "concentration_risk": round(concentration, 2),
+                "drawdown_control": round(drawdown_control, 2),
+                "alpha_generation": round(alpha_generation, 2),
+                "benchmark_excess": round_or_none(metrics.get("benchmark_excess")),
+                "capital_efficiency": round(capital_efficiency, 2),
+            },
+            "sample_size": len(rows),
+            "warnings": warnings,
+            "explanation": explanation,
+            "policy": DECISION_INTELLIGENCE_POLICY,
+        }
+        if persist:
+            db.add(
+                PortfolioQualityScore(
+                    game_id=game.id if game else None,
+                    portfolio_quality_score=payload["score"],
+                    diversification=payload["components"]["diversification"],
+                    concentration_risk=payload["components"]["concentration_risk"],
+                    drawdown_control=payload["components"]["drawdown_control"],
+                    alpha_generation=payload["components"]["alpha_generation"],
+                    benchmark_excess=payload["components"]["benchmark_excess"],
+                    capital_efficiency=payload["components"]["capital_efficiency"],
+                    explanation=explanation,
+                    warnings_json={"warnings": warnings},
+                )
+            )
+            db.commit()
+        return payload
+
+    def contributions(self, db: Session, persist: bool = False) -> dict:
+        game = latest_trading_game(db)
+        rows = portfolio_contribution_rows(executable_trades(game_trades(db, game.id if game else None)))
+        if persist:
+            for item in rows:
+                db.add(
+                    PortfolioContribution(
+                        game_id=game.id if game else None,
+                        ticker=item["ticker"],
+                        sector=item.get("sector"),
+                        return_contribution=item.get("return_contribution"),
+                        risk_contribution=item.get("risk_contribution"),
+                        drawdown_contribution=item.get("drawdown_contribution"),
+                        alpha_contribution=item.get("alpha_contribution"),
+                        evidence_json=item,
+                    )
+                )
+            db.commit()
+        return {"status": "ok" if game else "no_game", "rows": rows, "policy": DECISION_INTELLIGENCE_POLICY}
+
+    def correlations(self, db: Session, persist: bool = False) -> dict:
+        game = latest_trading_game(db)
+        tickers = sorted({row.ticker for row in executable_trades(game_trades(db, game.id if game else None))})[:14]
+        rows = correlation_rows(db, tickers)
+        if persist:
+            for item in rows:
+                db.add(
+                    PortfolioCorrelation(
+                        scope=f"game:{game.id}" if game else "global",
+                        asset_a=item["asset_a"],
+                        asset_b=item["asset_b"],
+                        correlation=item.get("correlation"),
+                        correlation_type="price_return",
+                        evidence_json=item.get("evidence", {}),
+                    )
+                )
+            db.commit()
+        return {"status": "ok" if game else "no_game", "rows": rows, "policy": DECISION_INTELLIGENCE_POLICY}
+
+    def alpha_scores(self, db: Session, persist: bool = False) -> dict:
+        game = latest_trading_game(db)
+        contributions = portfolio_contribution_rows(executable_trades(game_trades(db, game.id if game else None)))
+        rows = []
+        for item in contributions:
+            score = clamp(50 + safe_float(item.get("alpha_contribution")) * 2 - safe_float(item.get("risk_contribution")) * 0.2 + (20 if safe_float(item.get("return_contribution")) > 0 else -10))
+            rows.append({**item, "portfolio_alpha_score": round(score, 2), "marginal_return_score": round(clamp(50 + safe_float(item.get("return_contribution")) * 2), 2), "marginal_risk_score": round(clamp(100 - safe_float(item.get("risk_contribution"))), 2), "diversification_score": None, "benchmark_excess_score": round(clamp(50 + safe_float(item.get("alpha_contribution")) * 2), 2)})
+        rows = sorted(rows, key=lambda item: safe_float(item.get("portfolio_alpha_score")), reverse=True)
+        if persist:
+            for item in rows:
+                db.add(
+                    PortfolioAlphaScore(
+                        game_id=game.id if game else None,
+                        ticker=item["ticker"],
+                        portfolio_alpha_score=item["portfolio_alpha_score"],
+                        marginal_return_score=item.get("marginal_return_score"),
+                        marginal_risk_score=item.get("marginal_risk_score"),
+                        diversification_score=item.get("diversification_score"),
+                        benchmark_excess_score=item.get("benchmark_excess_score"),
+                        evidence_json=item,
+                    )
+                )
+            db.commit()
+        return {"status": "ok" if game else "no_game", "rows": rows, "policy": DECISION_INTELLIGENCE_POLICY}
+
+    def position_sizing_outcomes(self, db: Session, persist: bool = False) -> dict:
+        rows = executable_trades(game_trades(db, latest_trading_game(db).id if latest_trading_game(db) else None))
+        buckets: dict[str, list[TradingGameTrade]] = defaultdict(list)
+        for row in rows:
+            logic = "confidence_adjusted" if safe_float(row.confidence_at_entry) >= 65 else "fixed_fractional"
+            if safe_float(row.risk_percent) > 1.5:
+                logic = "higher_risk_fractional"
+            buckets[logic].append(row)
+        output = []
+        for logic, group in buckets.items():
+            r_values = [safe_float(row.realized_r_multiple) for row in group if row.realized_r_multiple is not None]
+            output.append(
+                {
+                    "sizing_logic": logic,
+                    "timeframe": "trade_horizon",
+                    "sample_size": len(group),
+                    "average_r": round_or_none(mean(r_values) if r_values else None),
+                    "drawdown_impact": round_or_none(min([safe_float(row.pnl_percent) for row in group], default=0)),
+                    "capital_efficiency": round_or_none(mean([safe_float(row.net_pnl_eur if row.net_pnl_eur is not None else row.realized_pl) for row in group]) if group else None),
+                    "evidence": {"tickers": sorted({row.ticker for row in group})[:16]},
+                }
+            )
+        if persist:
+            for item in output:
+                db.add(
+                    PositionSizingOutcome(
+                        sizing_logic=item["sizing_logic"],
+                        timeframe=item.get("timeframe"),
+                        sample_size=item["sample_size"],
+                        average_r=item.get("average_r"),
+                        drawdown_impact=item.get("drawdown_impact"),
+                        capital_efficiency=item.get("capital_efficiency"),
+                        evidence_json=item.get("evidence", {}),
+                    )
+                )
+            db.commit()
+        return {"status": "ok", "rows": output, "policy": DECISION_INTELLIGENCE_POLICY}
+
+
+class DecisionIntelligenceDashboardService:
+    def dashboard(self, db: Session) -> dict:
+        return {
+            "status": "ok",
+            "generated_at": datetime.utcnow().isoformat(),
+            "decision": DecisionSuperiorityEngine().dashboard(db),
+            "business_quality": BusinessQualityEngine().dashboard(db),
+            "portfolio": PortfolioIntelligenceEngine().dashboard(db),
+            "policy": DECISION_INTELLIGENCE_POLICY,
+        }
+
+
+def decision_universe_snapshot_payload(db: Session, trade: TradingGameTrade) -> dict:
+    evidence = decision_evidence_for_trade(db, trade)
+    return {
+        "timestamp": (trade.entry_date or trade.created_at.date()).isoformat() if hasattr((trade.entry_date or trade.created_at), "isoformat") else datetime.utcnow().isoformat(),
+        "market_regime": trade.market_regime_at_entry or "unknown",
+        "volatility_regime": volatility_regime_for_trade(trade),
+        "selected_asset": trade.ticker,
+        "selected_rank": evidence.get("selected_rank"),
+        "selected_score": evidence.get("selected_score"),
+        "total_candidates": evidence.get("candidate_count", 0),
+        "candidates": evidence.get("candidates", [])[:20],
+        "benchmark_snapshot": {"benchmark": trade.benchmark_ticker or "SPY", "benchmark_return": trade.benchmark_return_same_period or trade.benchmark_return},
+    }
+
+
+def decision_evidence_for_trade(db: Session, trade: TradingGameTrade) -> dict:
+    candidates = candidate_returns_for_trade(db, trade)
+    selected_score = safe_float(trade.sniper_score_at_entry if trade.sniper_score_at_entry is not None else trade.opportunity_score_at_entry)
+    if not any(item["ticker"] == trade.ticker for item in candidates):
+        candidates.append(
+            {
+                "ticker": trade.ticker,
+                "score": selected_score,
+                "realized_return": selected_return(trade),
+                "sector": trade.sector,
+                "selected": True,
+            }
+        )
+    candidates = dedupe_candidates(candidates)
+    ranked = sorted(candidates, key=lambda item: safe_float(item.get("score")), reverse=True)
+    realized_sorted = sorted(candidates, key=lambda item: safe_float(item.get("realized_return")), reverse=True)
+    selected = next((item for item in ranked if item["ticker"] == trade.ticker), ranked[0] if ranked else None)
+    selected_rank = next((index + 1 for index, item in enumerate(ranked) if item["ticker"] == trade.ticker), None)
+    best = realized_sorted[0] if realized_sorted else selected
+    selected_ret = safe_float(selected.get("realized_return") if selected else selected_return(trade))
+    best_ret = safe_float(best.get("realized_return") if best else selected_ret)
+    benchmark = safe_float(trade.benchmark_return_same_period if trade.benchmark_return_same_period is not None else trade.benchmark_return)
+    opportunity_gap = max(0.0, best_ret - selected_ret)
+    selected_outperformed = selected_ret > benchmark and selected_ret > 0
+    missed_best = bool(best and best.get("ticker") != trade.ticker and opportunity_gap > 1.0)
+    top_score_ticker = ranked[0]["ticker"] if ranked else None
+    best_return_ticker = best["ticker"] if best else None
+    return {
+        "trade_id": trade.id,
+        "ticker": trade.ticker,
+        "setup_type": trade.setup_type,
+        "sector": trade.sector,
+        "regime": trade.market_regime_at_entry,
+        "timeframe": trade.timeframe,
+        "selected_rank": selected_rank,
+        "selected_score": round_or_none(selected_score),
+        "selected_return": round_or_none(selected_ret),
+        "best_available_ticker": best_return_ticker,
+        "best_available_return": round_or_none(best_ret),
+        "top_ranked_ticker": top_score_ticker,
+        "benchmark_return": round_or_none(benchmark),
+        "opportunity_gap": round_or_none(opportunity_gap),
+        "selected_outperformed": selected_outperformed,
+        "missed_best": missed_best,
+        "candidate_count": len(candidates),
+        "ranking_correct": top_score_ticker == best_return_ticker if top_score_ticker and best_return_ticker else None,
+        "selection_quality": round_or_none(clamp(100 - opportunity_gap * 4 + (15 if selected_outperformed else -10))),
+        "candidates": ranked[:20],
+    }
+
+
+def candidate_returns_for_trade(db: Session, trade: TradingGameTrade) -> list[dict]:
+    start = trade.entry_date or (trade.created_at.date() if trade.created_at else None)
+    end = trade.exit_date or (start + timedelta(days=30) if start else None)
+    if not start or not end:
+        return []
+    signal_candidates = db.scalars(
+        select(SignalSnapshot)
+        .where(SignalSnapshot.created_at <= datetime.combine(start, datetime.max.time()))
+        .order_by(desc(SignalSnapshot.created_at), desc(SignalSnapshot.blum_score))
+        .limit(80)
+    ).all()
+    seen = set()
+    output = []
+    for signal in signal_candidates:
+        if signal.ticker in seen:
+            continue
+        seen.add(signal.ticker)
+        realized = ticker_return_between(db, signal.ticker, start, end)
+        if realized is None:
+            continue
+        output.append({"ticker": signal.ticker, "score": safe_float(signal.blum_score), "realized_return": round(realized, 4), "sector": signal.asset.sector if signal.asset else None, "selected": signal.ticker == trade.ticker})
+        if len(output) >= 20:
+            break
+    if len(output) < 2:
+        same_day = db.scalars(select(TradingGameTrade).where(TradingGameTrade.entry_date == start).limit(30)).all()
+        for row in same_day:
+            if row.ticker in seen:
+                continue
+            seen.add(row.ticker)
+            output.append({"ticker": row.ticker, "score": safe_float(row.sniper_score_at_entry if row.sniper_score_at_entry is not None else row.opportunity_score_at_entry), "realized_return": selected_return(row), "sector": row.sector, "selected": row.ticker == trade.ticker})
+    return output
+
+
+def decision_superiority_metrics(decisions: list[dict], rows: list[TradingGameTrade]) -> dict:
+    if not decisions:
+        return {"status": "insufficient_evidence", "sample_size": 0}
+    outperformers = [item for item in decisions if item.get("best_available_return") is not None and safe_float(item.get("best_available_return")) > safe_float(item.get("benchmark_return"))]
+    captured = [item for item in outperformers if item.get("selected_outperformed")]
+    selected_success = [item for item in decisions if item.get("selected_outperformed")]
+    available_alpha = sum(max(0, safe_float(item.get("best_available_return")) - safe_float(item.get("benchmark_return"))) for item in decisions)
+    captured_alpha = sum(max(0, safe_float(item.get("selected_return")) - safe_float(item.get("benchmark_return"))) for item in decisions)
+    ranking_known = [item for item in decisions if item.get("ranking_correct") is not None]
+    ranking_accuracy = sum(1 for item in ranking_known if item.get("ranking_correct")) / max(1, len(ranking_known))
+    benchmark_excess_values = [safe_float(row.excess_return_vs_benchmark) for row in rows if row.excess_return_vs_benchmark is not None]
+    return {
+        "status": "ok",
+        "sample_size": len(decisions),
+        "opportunity_recall": round_or_none(len(captured) / max(1, len(outperformers))),
+        "captured_outperformers": len(captured),
+        "total_outperformers": len(outperformers),
+        "opportunity_precision": round_or_none(len(selected_success) / max(1, len(decisions))),
+        "successful_opportunities": len(selected_success),
+        "selected_opportunities": len(decisions),
+        "alpha_capture_rate": round_or_none(captured_alpha / available_alpha if available_alpha else None),
+        "available_alpha": round_or_none(available_alpha),
+        "captured_alpha": round_or_none(captured_alpha),
+        "ranking_accuracy": round_or_none(ranking_accuracy),
+        "top1_accuracy": round_or_none(ranking_accuracy),
+        "missed_opportunities": sum(1 for item in decisions if item.get("missed_best")),
+        "benchmark_excess": round_or_none(mean(benchmark_excess_values) if benchmark_excess_values else None),
+    }
+
+
+def decision_superiority_components(metrics: dict, rows: list[TradingGameTrade]) -> dict:
+    if metrics.get("status") == "insufficient_evidence":
+        return {key: 0 for key in ["opportunity_recall", "opportunity_precision", "alpha_capture", "ranking_accuracy", "benchmark_excess", "live_validation", "regime_consistency", "reproducibility", "drawdown_control"]}
+    live = [row for row in rows if row.mode == "live_forward_paper"]
+    regimes = {row.market_regime_at_entry for row in rows if row.market_regime_at_entry}
+    reproducibility = mean([safe_float(row.reproducibility_score) for row in rows]) if rows else 0
+    drawdown_values = [safe_float(row.max_adverse_excursion) for row in rows if row.max_adverse_excursion is not None]
+    return {
+        "opportunity_recall": round(clamp(safe_float(metrics.get("opportunity_recall")) * 100), 2),
+        "opportunity_precision": round(clamp(safe_float(metrics.get("opportunity_precision")) * 100), 2),
+        "alpha_capture": round(clamp(safe_float(metrics.get("alpha_capture_rate")) * 100), 2),
+        "ranking_accuracy": round(clamp(safe_float(metrics.get("ranking_accuracy")) * 100), 2),
+        "benchmark_excess": round(clamp(50 + safe_float(metrics.get("benchmark_excess")) * 3), 2),
+        "live_validation": round(clamp(min(100, len(live) * 3)), 2),
+        "regime_consistency": round(clamp(len(regimes) * 18), 2),
+        "reproducibility": round(clamp(reproducibility), 2),
+        "drawdown_control": round(clamp(100 + (min(drawdown_values) if drawdown_values else 0) * 2), 2),
+    }
+
+
+def decision_superiority_warnings(metrics: dict, rows: list[TradingGameTrade]) -> list[str]:
+    warnings = []
+    if metrics.get("sample_size", 0) < 30:
+        warnings.append("Insufficient comparable decision samples for a strong superiority claim.")
+    if count_live(rows) < 30:
+        warnings.append("Live forward evidence is not mature enough to validate selection superiority.")
+    if safe_float(metrics.get("alpha_capture_rate")) < 0.5 and metrics.get("alpha_capture_rate") is not None:
+        warnings.append("BLUM is leaving more than half of available alpha uncaptured in comparable samples.")
+    if safe_float(metrics.get("ranking_accuracy")) < 0.5 and metrics.get("ranking_accuracy") is not None:
+        warnings.append("Ranking accuracy is weak; BLUM may identify candidates but order them poorly.")
+    return warnings
+
+
+def business_quality_for_asset(db: Session, asset: Asset) -> dict:
+    snapshots = db.scalars(select(FundamentalSnapshot).where(FundamentalSnapshot.asset_id == asset.id).order_by(desc(FundamentalSnapshot.period_end), desc(FundamentalSnapshot.created_at)).limit(6)).all()
+    latest = snapshots[0] if snapshots else None
+    metrics = latest.metrics if latest else {}
+    data_quality = safe_float(latest.quality_score if latest else 0)
+    revenue = metric_value(metrics, "revenue")
+    net_income = metric_value(metrics, "net_income")
+    operating_income = metric_value(metrics, "operating_income")
+    operating_cash_flow = metric_value(metrics, "operating_cash_flow")
+    capex = abs(metric_value(metrics, "capex") or 0)
+    assets = metric_value(metrics, "assets")
+    liabilities = metric_value(metrics, "liabilities")
+    equity = metric_value(metrics, "equity")
+    fcf = operating_cash_flow - capex if operating_cash_flow is not None else None
+    growth = trend_score([metric_value(row.metrics, "revenue") for row in snapshots])
+    profitability = clamp(45 + ratio_pct(net_income, revenue) * 2 + ratio_pct(operating_income, revenue))
+    cash_flow = clamp(45 + ratio_pct(fcf, revenue) * 2)
+    balance = clamp(55 + ratio_pct(equity, assets) - ratio_pct(liabilities, assets) * 0.35)
+    capital_allocation = clamp(50 + ratio_pct(fcf, assets) * 3)
+    moat = moat_score(asset, profitability, cash_flow)
+    management = management_score(data_quality, growth, profitability)
+    raw_score = weighted_average([growth, profitability, cash_flow, balance, capital_allocation, moat, management])
+    score = clamp(raw_score * (0.45 + min(0.55, data_quality / 100)))
+    status = "ready" if latest else "insufficient_fundamental_evidence"
+    trend = "improving" if growth >= 62 and profitability >= 55 else "deteriorating" if growth < 38 or profitability < 35 else "stable"
+    return {
+        "ticker": asset.ticker,
+        "name": asset.name,
+        "sector": asset.sector,
+        "business_quality_score": round(score, 2),
+        "growth_quality": round(growth, 2),
+        "profitability_quality": round(profitability, 2),
+        "cash_flow_quality": round(cash_flow, 2),
+        "balance_sheet_quality": round(balance, 2),
+        "capital_allocation_quality": round(capital_allocation, 2),
+        "moat_quality": round(moat, 2),
+        "management_quality": round(management, 2),
+        "fundamental_alpha_score": round(clamp(score * 0.65 + growth * 0.2 + cash_flow * 0.15), 2),
+        "data_quality_score": round(data_quality, 2),
+        "status": status,
+        "trend_label": trend,
+        "evidence": {
+            "period_end": latest.period_end.isoformat() if latest and latest.period_end else None,
+            "provider": latest.provider if latest else None,
+            "metrics_available": sorted(metrics.keys()) if isinstance(metrics, dict) else [],
+            "warning": None if latest else "No stored fundamental snapshot; score is penalized and should not be treated as business-quality proof.",
+        },
+        "management_components": {
+            "insider_alignment": None,
+            "execution_consistency": round(clamp((data_quality + growth) / 2), 2),
+            "earnings_delivery": None,
+        },
+    }
+
+
+def portfolio_contribution_rows(rows: list[TradingGameTrade]) -> list[dict]:
+    total_pl = sum(safe_float(row.net_pnl_eur if row.net_pnl_eur is not None else row.realized_pl) for row in rows)
+    total_risk = sum(abs(safe_float(row.max_adverse_excursion if row.max_adverse_excursion is not None else row.risk_amount)) for row in rows)
+    grouped: dict[str, list[TradingGameTrade]] = defaultdict(list)
+    for row in rows:
+        grouped[row.ticker].append(row)
+    output = []
+    for ticker, group in grouped.items():
+        pl = sum(safe_float(row.net_pnl_eur if row.net_pnl_eur is not None else row.realized_pl) for row in group)
+        risk = sum(abs(safe_float(row.max_adverse_excursion if row.max_adverse_excursion is not None else row.risk_amount)) for row in group)
+        alpha = sum(safe_float(row.excess_return_vs_benchmark) for row in group if row.excess_return_vs_benchmark is not None)
+        output.append(
+            {
+                "ticker": ticker,
+                "sector": group[0].sector,
+                "trades": len(group),
+                "return_contribution": round_or_none(pl / total_pl * 100 if total_pl else None),
+                "risk_contribution": round_or_none(risk / total_risk * 100 if total_risk else None),
+                "drawdown_contribution": round_or_none(min([safe_float(row.pnl_percent) for row in group], default=0)),
+                "alpha_contribution": round_or_none(alpha / max(1, len(group))),
+            }
+        )
+    return sorted(output, key=lambda item: abs(safe_float(item.get("return_contribution"))), reverse=True)
+
+
+def correlation_rows(db: Session, tickers: list[str]) -> list[dict]:
+    output = []
+    series = {ticker: daily_returns(db, ticker) for ticker in tickers}
+    for index, ticker_a in enumerate(tickers):
+        for ticker_b in tickers[index + 1 :]:
+            corr = pearson(series.get(ticker_a, []), series.get(ticker_b, []))
+            output.append({"asset_a": ticker_a, "asset_b": ticker_b, "correlation": round_or_none(corr), "evidence": {"points": min(len(series.get(ticker_a, [])), len(series.get(ticker_b, [])))}})
+    return sorted(output, key=lambda item: abs(safe_float(item.get("correlation"))), reverse=True)[:40]
+
+
+def candidate_scope_metric_rows(decisions: list[dict], key: str) -> list[dict]:
+    grouped: dict[str, list[dict]] = defaultdict(list)
+    for item in decisions:
+        grouped[str(item.get(key) or "Unknown")].append(item)
+    return [{"scope": key, "entity": entity, **decision_superiority_metrics(items, [])} for entity, items in grouped.items()]
+
+
+def classify_decision_superiority(score: float) -> str:
+    if score <= 20:
+        return "Weak"
+    if score <= 40:
+        return "Experimental"
+    if score <= 60:
+        return "Learning"
+    if score <= 75:
+        return "Competitive"
+    if score <= 90:
+        return "Strong Alpha Research"
+    return "Exceptional"
+
+
+def decision_superiority_explanation(score: float, classification: str, metrics: dict, warnings: list[str]) -> str:
+    if metrics.get("status") == "insufficient_evidence":
+        return "Insufficient evidence. BLUM does not yet have enough comparable decision snapshots to claim selection superiority."
+    base = f"Decision Superiority Score {score:.1f}/100 ({classification}). Opportunity recall {pct(metrics.get('opportunity_recall'))}, precision {pct(metrics.get('opportunity_precision'))}, alpha capture {pct(metrics.get('alpha_capture_rate'))}."
+    if warnings:
+        base += f" Main warning: {warnings[0]}"
+    return base
+
+
+def selected_return(trade: TradingGameTrade) -> float:
+    if trade.pnl_percent is not None:
+        return safe_float(trade.pnl_percent)
+    if trade.entry_price and trade.exit_price:
+        return (trade.exit_price / trade.entry_price - 1) * 100
+    return safe_float(trade.excess_return_vs_benchmark if trade.excess_return_vs_benchmark is not None else trade.realized_r_multiple)
+
+
+def ticker_return_between(db: Session, ticker: str, start: date, end: date) -> float | None:
+    asset = db.scalar(select(Asset).where(Asset.ticker == ticker))
+    if not asset:
+        return None
+    start_row = db.scalar(select(PriceHistory).where(PriceHistory.asset_id == asset.id, PriceHistory.date >= start).order_by(PriceHistory.date).limit(1))
+    end_row = db.scalar(select(PriceHistory).where(PriceHistory.asset_id == asset.id, PriceHistory.date <= end).order_by(desc(PriceHistory.date)).limit(1))
+    if not start_row or not end_row or not start_row.close:
+        return None
+    return (safe_float(end_row.close) / safe_float(start_row.close) - 1) * 100
+
+
+def dedupe_candidates(candidates: list[dict]) -> list[dict]:
+    best: dict[str, dict] = {}
+    for item in candidates:
+        ticker = item.get("ticker")
+        if not ticker:
+            continue
+        if ticker not in best or safe_float(item.get("score")) > safe_float(best[ticker].get("score")):
+            best[ticker] = item
+    return list(best.values())
+
+
+def weighted_average(values: list[float | None]) -> float:
+    clean = [safe_float(value) for value in values if value is not None]
+    return clamp(mean(clean) if clean else 0)
+
+
+def pct(value: float | None) -> str:
+    return "n/a" if value is None else f"{safe_float(value) * 100:.1f}%"
+
+
+def count_live(rows: list[TradingGameTrade]) -> int:
+    return sum(1 for row in rows if row.mode == "live_forward_paper")
+
+
+def volatility_regime_for_trade(trade: TradingGameTrade) -> str:
+    mae = abs(safe_float(trade.max_adverse_excursion))
+    if mae >= 10:
+        return "high_volatility"
+    if mae <= 2:
+        return "low_volatility"
+    return "normal_volatility"
+
+
+def metric_value(metrics: dict, key: str) -> float | None:
+    raw = metrics.get(key) if isinstance(metrics, dict) else None
+    if isinstance(raw, dict):
+        raw = raw.get("value")
+    if raw is None:
+        return None
+    try:
+        return float(raw)
+    except (TypeError, ValueError):
+        return None
+
+
+def ratio_pct(numerator: float | None, denominator: float | None) -> float:
+    if numerator is None or denominator in (None, 0):
+        return 0.0
+    return numerator / denominator * 100
+
+
+def trend_score(values: list[float | None]) -> float:
+    clean = [float(value) for value in values if value not in (None, 0)]
+    if len(clean) < 2:
+        return 45.0
+    latest, oldest = clean[0], clean[-1]
+    if oldest == 0:
+        return 45.0
+    return clamp(50 + (latest / oldest - 1) * 100)
+
+
+def moat_score(asset: Asset, profitability: float, cash_flow: float) -> float:
+    text = f"{asset.category} {asset.sector} {asset.industry} {asset.description}".lower()
+    boost = 0
+    for token in ["platform", "ecosystem", "semiconductor", "cloud", "security", "defense", "luxury", "healthcare", "ai"]:
+        if token in text:
+            boost += 4
+    return clamp(45 + boost + (profitability - 50) * 0.25 + (cash_flow - 50) * 0.2)
+
+
+def management_score(data_quality: float, growth: float, profitability: float) -> float:
+    return clamp(42 + data_quality * 0.25 + growth * 0.2 + profitability * 0.2)
+
+
+def concentration_score(contributions: list[dict]) -> float:
+    values = sorted([abs(safe_float(item.get("return_contribution"))) for item in contributions], reverse=True)
+    if not values:
+        return 100.0
+    return clamp(sum(values[:3]))
+
+
+def portfolio_warnings(rows: list[TradingGameTrade], concentration: float, metrics: dict) -> list[str]:
+    warnings = []
+    if len(rows) < 30:
+        warnings.append("Portfolio evidence is still low sample.")
+    if concentration >= 70:
+        warnings.append("Portfolio result is concentrated in the top contributors.")
+    if safe_float(metrics.get("benchmark_excess")) < 0:
+        warnings.append("Portfolio is underperforming its benchmark context.")
+    return warnings
+
+
+def daily_returns(db: Session, ticker: str) -> list[float]:
+    asset = db.scalar(select(Asset).where(Asset.ticker == ticker))
+    if not asset:
+        return []
+    rows = db.scalars(select(PriceHistory).where(PriceHistory.asset_id == asset.id).order_by(desc(PriceHistory.date)).limit(260)).all()
+    values = [safe_float(row.close) for row in reversed(rows) if row.close]
+    return [(values[index] / values[index - 1] - 1) * 100 for index in range(1, len(values)) if values[index - 1]]
+
+
+def pearson(a: list[float], b: list[float]) -> float | None:
+    n = min(len(a), len(b))
+    if n < 20:
+        return None
+    x, y = a[-n:], b[-n:]
+    mean_x, mean_y = mean(x), mean(y)
+    numerator = sum((xv - mean_x) * (yv - mean_y) for xv, yv in zip(x, y))
+    den_x = sqrt(sum((xv - mean_x) ** 2 for xv in x))
+    den_y = sqrt(sum((yv - mean_y) ** 2 for yv in y))
+    if den_x == 0 or den_y == 0:
+        return None
+    return numerator / (den_x * den_y)
+
+
+def parse_dt(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(value)
+        return parsed if isinstance(parsed, datetime) else datetime.combine(parsed, datetime.min.time())
+    except ValueError:
+        return None

backend/app/services/financial_chat.py

@@ -33,6 +33,7 @@ from app.services.fundamentals import fundamentals_for_asset
 from app.services.live import market_sentiment
 from app.services.learning_loop import LearningDashboardService
 from app.services.learning_intelligence import LearningIntelligenceDashboardService
+from app.services.decision_intelligence import DecisionIntelligenceDashboardService
 from app.services.market_data import market_snapshot_for_asset
 from app.services.market_sniper import MarketSniperEngine
 from app.services.reasoning_precision import (
@@ -483,6 +484,7 @@ def trading_game_context_for_chat(db: Session) -> dict:
             "live_forward": LiveForwardPaperTradingService().status(db),
             "historical_vs_live": HistoricalLiveComparisonService().compare(db),
             "learning_intelligence": LearningIntelligenceDashboardService().dashboard(db),
+            "decision_intelligence": DecisionIntelligenceDashboardService().dashboard(db),
             "pnl_breakdown": PnLBreakdownService().game_breakdown(db),
             "reality_check": TradingGameRealityCheckService().evaluate(db),
             "failures": engine.failures(db, limit=12),
@@ -1469,6 +1471,7 @@ def build_trading_game_response(language: str, context: dict) -> dict:
     pnl_breakdown = context.get("pnl_breakdown") or {}
     reality_check = context.get("reality_check") or {}
     learning_intelligence = context.get("learning_intelligence") or {}
+    decision_intelligence = context.get("decision_intelligence") or {}
     if not game:
         message = "BLUM non ha ancora un Trading Game persistito. Serve almeno un ciclo Sniper/Learning Loop per creare simulazioni P/L reali." if language == "it" else "BLUM does not have a persisted Trading Game yet. It needs at least one Sniper/Learning Loop cycle to create real P/L simulations."
         return build_error_response(language, message, [])
@@ -1495,6 +1498,7 @@ def build_trading_game_response(language: str, context: dict) -> dict:
                 "Nessuna dichiarazione di outperformance e valida se il campione e piccolo o incompleto.",
             ]},
             {"key": "learning_intelligence", "title": "Learning Intelligence", "bullets": learning_intelligence_lines(learning_intelligence, language)},
+            {"key": "decision_intelligence", "title": "Decision Intelligence", "bullets": decision_intelligence_lines(decision_intelligence, language)},
             {"key": "trade_ledger", "title": "Trade ledger", "bullets": trade_ledger_lines(ledger_rows, language)},
             {"key": "intelligence_metrics", "title": "Trading intelligence", "bullets": intelligence_metric_lines(intelligence_metrics, rolling_metrics, metrics_by_setup, language)},
             {"key": "live_forward", "title": "Storico vs live paper", "bullets": historical_live_lines(live_forward, historical_vs_live, language)},
@@ -1533,6 +1537,7 @@ def build_trading_game_response(language: str, context: dict) -> dict:
                 "No outperformance claim is valid when sample size or benchmark coverage is insufficient.",
             ]},
             {"key": "learning_intelligence", "title": "Learning Intelligence", "bullets": learning_intelligence_lines(learning_intelligence, language)},
+            {"key": "decision_intelligence", "title": "Decision Intelligence", "bullets": decision_intelligence_lines(decision_intelligence, language)},
             {"key": "trade_ledger", "title": "Trade Ledger", "bullets": trade_ledger_lines(ledger_rows, language)},
             {"key": "intelligence_metrics", "title": "Trading Intelligence", "bullets": intelligence_metric_lines(intelligence_metrics, rolling_metrics, metrics_by_setup, language)},
             {"key": "live_forward", "title": "Historical vs Live Paper", "bullets": historical_live_lines(live_forward, historical_vs_live, language)},
@@ -1562,7 +1567,7 @@ def build_trading_game_response(language: str, context: dict) -> dict:
         "executive_view": summary,
         "risk_reward_view": f"Expectancy {format_number(game.get('expectancy_r'))}R, drawdown {format_signed(game.get('max_drawdown'))}%.",
         "data_quality": {"sample_warning": sample_warning, "trades": game.get("trade_count"), "reproducibility": reproducibility, "cycles": cycle_stats, "live_sample_warning": (historical_vs_live.get("sample_warning") if isinstance(historical_vs_live, dict) else None)},
-        "learning_loop_memory": {"trading_game": game, "lessons": lessons[:6], "latest_trades": trades[:6], "ledger": ledger_rows[:8], "reality_check": reality_check, "cycles": cycle_stats, "intelligence_metrics": intelligence_metrics, "historical_vs_live": historical_vs_live, "learning_intelligence": learning_intelligence},
+        "learning_loop_memory": {"trading_game": game, "lessons": lessons[:6], "latest_trades": trades[:6], "ledger": ledger_rows[:8], "reality_check": reality_check, "cycles": cycle_stats, "intelligence_metrics": intelligence_metrics, "historical_vs_live": historical_vs_live, "learning_intelligence": learning_intelligence, "decision_intelligence": decision_intelligence},
         "answer_to_user": summary,
     }
 
@@ -1602,6 +1607,42 @@ def learning_intelligence_lines(payload: dict, language: str) -> list[str]:
     return dedupe_warnings(lines)
 
 
+def decision_intelligence_lines(payload: dict, language: str) -> list[str]:
+    if not payload or payload.get("status") == "unavailable":
+        return ["Decision Intelligence non disponibile: non invento opportunita mancate o qualita portfolio." if language == "it" else "Decision Intelligence is unavailable; I will not invent missed opportunities or portfolio-quality evidence."]
+    decision = ((payload.get("decision") or {}).get("decision_superiority") or {})
+    business = (payload.get("business_quality") or {}).get("highest_quality_companies") or []
+    portfolio = ((payload.get("portfolio") or {}).get("portfolio_quality") or {})
+    missed = (payload.get("decision") or {}).get("top_missed_opportunities") or []
+    if language == "it":
+        lines = [
+            f"Decision Superiority Score: {format_number(decision.get('score'))}/100 | {decision.get('classification', 'insufficient evidence')} | campione {decision.get('sample_size', 0)}.",
+            f"Opportunity recall {format_pct(decision.get('metrics', {}).get('opportunity_recall'))}; precision {format_pct(decision.get('metrics', {}).get('opportunity_precision'))}; alpha capture {format_pct(decision.get('metrics', {}).get('alpha_capture_rate'))}.",
+            f"Portfolio Quality Score: {format_number(portfolio.get('score'))}/100 | concentrazione {format_number((portfolio.get('components') or {}).get('concentration_risk'))}/100.",
+        ]
+        if missed:
+            top = missed[0]
+            lines.append(f"Possibile opportunita mancata: BLUM ha scelto {top.get('ticker')}, ma {top.get('best_available_ticker')} aveva un outcome migliore nello stesso confronto.")
+        if business:
+            lines.append("Business quality piu alta: " + ", ".join(f"{row.get('ticker')} {format_number(row.get('business_quality_score'))}/100" for row in business[:3]))
+        if decision.get("warnings"):
+            lines.extend(str(item) for item in decision.get("warnings", [])[:2])
+        return dedupe_warnings(lines)
+    lines = [
+        f"Decision Superiority Score: {format_number(decision.get('score'))}/100 | {decision.get('classification', 'insufficient evidence')} | sample {decision.get('sample_size', 0)}.",
+        f"Opportunity recall {format_pct(decision.get('metrics', {}).get('opportunity_recall'))}; precision {format_pct(decision.get('metrics', {}).get('opportunity_precision'))}; alpha capture {format_pct(decision.get('metrics', {}).get('alpha_capture_rate'))}.",
+        f"Portfolio Quality Score: {format_number(portfolio.get('score'))}/100 | concentration {format_number((portfolio.get('components') or {}).get('concentration_risk'))}/100.",
+    ]
+    if missed:
+        top = missed[0]
+        lines.append(f"Possible missed opportunity: BLUM selected {top.get('ticker')}, while {top.get('best_available_ticker')} had a better outcome in the same comparison.")
+    if business:
+        lines.append("Highest business quality: " + ", ".join(f"{row.get('ticker')} {format_number(row.get('business_quality_score'))}/100" for row in business[:3]))
+    if decision.get("warnings"):
+        lines.extend(str(item) for item in decision.get("warnings", [])[:2])
+    return dedupe_warnings(lines)
+
+
 def trade_ledger_lines(rows: list[dict], language: str) -> list[str]:
     if not rows:
         return ["BLUM ha metriche di gioco, ma il trade ledger dettagliato non e ancora disponibile." if language == "it" else "BLUM has game-level metrics, but the detailed trade ledger is not available yet."]
@@ -2144,6 +2185,7 @@ def summarize_trading_game_context(context: dict) -> dict:
     intelligence = (context or {}).get("intelligence_metrics") or {}
     historical_vs_live = (context or {}).get("historical_vs_live") or {}
     learning_intelligence = (context or {}).get("learning_intelligence") or {}
+    decision_intelligence = (context or {}).get("decision_intelligence") or {}
     return {
         "current_game": {
             "status": game.get("status"),
@@ -2189,6 +2231,12 @@ def summarize_trading_game_context(context: dict) -> dict:
             "weakness_map": (learning_intelligence.get("weakness_map") or {}).get("rows", [])[:6],
             "self_improvement": (learning_intelligence.get("self_improvement") or {}).get("actions", [])[:6],
         },
+        "decision_intelligence": {
+            "decision_superiority": ((decision_intelligence.get("decision") or {}).get("decision_superiority") or {}),
+            "missed_opportunities": ((decision_intelligence.get("decision") or {}).get("top_missed_opportunities") or [])[:6],
+            "business_quality": ((decision_intelligence.get("business_quality") or {}).get("highest_quality_companies") or [])[:6],
+            "portfolio_quality": ((decision_intelligence.get("portfolio") or {}).get("portfolio_quality") or {}),
+        },
     }
 
 
@@ -2388,7 +2436,7 @@ def infer_intent(message: str, mode: str | None = None) -> str:
         return "fundamental_analysis"
     if any(term in normalized for term in ["tesi", "thesis", "convinzione", "conviction", "ancora valida", "still valid", "sopravviss", "survival", "decay", "decad", "bull bear neutral", "tesi bull", "tesi bear", "tesi neutral", "motore", "engine vote", "sta migliorando", "dove ha sbagliato", "reasoning core", "batte spy", "batte qqq", "vs spy", "vs qqq"]):
         return "reasoning_memory_question"
-    if any(term in normalized for term in ["capitale virtuale", "trading game", "sta battendo", "batte il mercato", "benchmark", "drawdown", "profit factor", "expectancy", "p/l", "pl ", "peggior errore", "andato a zero", "rischio per trade", "riproducibil", "reproducib", "win rate", "quali trade", "trade hanno", "dove e entrato", "dove e uscito", "entrato blum", "uscito blum", "per azione", "fortuna", "profitto arriva", "ledger", "trade piu importante", "100 eur", "10,000", "10000", "target cycle", "ciclo capitale", "cicli capitale", "quante volte", "live paper", "forward paper", "storico vs live", "historical vs live", "sta migliorando", "intelligence growth", "missed entry", "stop hit", "target hit", "trading power", "power score", "dove e scarso", "piu scarso", "weakness", "self improvement", "auto miglior", "prossima azione", "baseline semplice", "stiamo battendo spy", "stiamo battendo qqq"]):
+    if any(term in normalized for term in ["capitale virtuale", "trading game", "sta battendo", "batte il mercato", "benchmark", "drawdown", "profit factor", "expectancy", "p/l", "pl ", "peggior errore", "andato a zero", "rischio per trade", "riproducibil", "reproducib", "win rate", "quali trade", "trade hanno", "dove e entrato", "dove e uscito", "entrato blum", "uscito blum", "per azione", "fortuna", "profitto arriva", "ledger", "trade piu importante", "100 eur", "10,000", "10000", "target cycle", "ciclo capitale", "cicli capitale", "quante volte", "live paper", "forward paper", "storico vs live", "historical vs live", "sta migliorando", "intelligence growth", "missed entry", "stop hit", "target hit", "trading power", "power score", "dove e scarso", "piu scarso", "weakness", "self improvement", "auto miglior", "prossima azione", "baseline semplice", "stiamo battendo spy", "stiamo battendo qqq", "best opportunity", "migliore opportunita", "opportunita migliore", "ha scelto il migliore", "decision superiority", "superiorita decisionale", "alpha capture", "opportunita mancata", "missed opportunity", "business quality", "qualita business", "moat", "management quality", "portfolio quality", "qualita portafoglio", "contribuisce piu alpha", "contribution", "concentrazione portfolio"]):
         return "trading_game"
     if any(term in normalized for term in ["sniper", "entrabile", "meglio aspettare", "ingresso", "entry", "risk/reward", "uscita", "exit", "target", "invalidazione", "invalidation", "profitto", "take profit"]):
         return "market_sniper"

backend/tests/test_decision_intelligence.py

@@ -0,0 +1,73 @@
+from types import SimpleNamespace
+
+from app.services.decision_intelligence import (
+    classify_decision_superiority,
+    concentration_score,
+    decision_superiority_components,
+    decision_superiority_metrics,
+)
+
+
+def test_decision_superiority_classification():
+    assert classify_decision_superiority(15) == "Weak"
+    assert classify_decision_superiority(35) == "Experimental"
+    assert classify_decision_superiority(55) == "Learning"
+    assert classify_decision_superiority(70) == "Competitive"
+    assert classify_decision_superiority(84) == "Strong Alpha Research"
+    assert classify_decision_superiority(94) == "Exceptional"
+
+
+def test_decision_metrics_capture_missed_best_opportunity():
+    decisions = [
+        {
+            "selected_outperformed": True,
+            "best_available_return": 8.0,
+            "selected_return": 6.0,
+            "benchmark_return": 2.0,
+            "ranking_correct": False,
+            "missed_best": True,
+        },
+        {
+            "selected_outperformed": False,
+            "best_available_return": 5.0,
+            "selected_return": -1.0,
+            "benchmark_return": 1.0,
+            "ranking_correct": True,
+            "missed_best": True,
+        },
+    ]
+    metrics = decision_superiority_metrics(decisions, [])
+    assert metrics["total_outperformers"] == 2
+    assert metrics["captured_outperformers"] == 1
+    assert metrics["opportunity_recall"] == 0.5
+    assert metrics["opportunity_precision"] == 0.5
+    assert metrics["missed_opportunities"] == 2
+    assert 0 < metrics["alpha_capture_rate"] < 1
+
+
+def test_decision_components_penalize_low_live_validation():
+    metrics = {
+        "opportunity_recall": 0.5,
+        "opportunity_precision": 0.5,
+        "alpha_capture_rate": 0.4,
+        "ranking_accuracy": 0.5,
+        "benchmark_excess": 1.2,
+        "status": "ok",
+    }
+    rows = [
+        SimpleNamespace(mode="historical_simulation", market_regime_at_entry="risk_on", reproducibility_score=80, max_adverse_excursion=-2),
+        SimpleNamespace(mode="historical_simulation", market_regime_at_entry="risk_on", reproducibility_score=75, max_adverse_excursion=-3),
+    ]
+    components = decision_superiority_components(metrics, rows)
+    assert components["live_validation"] == 0
+    assert components["reproducibility"] > 70
+
+
+def test_concentration_score_reads_top_contributors():
+    rows = [
+        {"ticker": "A", "return_contribution": 50},
+        {"ticker": "B", "return_contribution": 20},
+        {"ticker": "C", "return_contribution": 10},
+        {"ticker": "D", "return_contribution": 5},
+    ]
+    assert concentration_score(rows) == 80

frontend/app/learning/page.tsx

@@ -23,7 +23,7 @@ export default function LearningPage() {
     let mounted = true;
     async function load() {
       setError("");
-      const [dashResult, runsResult, predictionsResult, memoryResult, tradingStatusResult, equityResult, annotatedEquityResult, tradesResult, ledgerResult, ledgerSummaryResult, cyclesResult, currentCycleResult, cycleStatsResult, intelligenceMetricsResult, rollingMetricsResult, metricsBySetupResult, metricsByRegimeResult, metricsBySectorResult, historicalVsLiveResult, liveStatusResult, livePositionsResult, liveMetricsResult, learningIntelligenceResult, learningEvidenceResult, realityCheckResult, pnlBreakdownResult, failuresResult, lessonsResult, benchmarkResult, reproducibilityResult, reasoningStatusResult, survivalResult, convictionResult, reliabilityResult, competitionResult, ensembleResult, benchmarkRelativeResult, trainingQualityResult] = await Promise.allSettled([
+      const [dashResult, runsResult, predictionsResult, memoryResult, tradingStatusResult, equityResult, annotatedEquityResult, tradesResult, ledgerResult, ledgerSummaryResult, cyclesResult, currentCycleResult, cycleStatsResult, intelligenceMetricsResult, rollingMetricsResult, metricsBySetupResult, metricsByRegimeResult, metricsBySectorResult, historicalVsLiveResult, liveStatusResult, livePositionsResult, liveMetricsResult, learningIntelligenceResult, decisionIntelligenceResult, learningEvidenceResult, realityCheckResult, pnlBreakdownResult, failuresResult, lessonsResult, benchmarkResult, reproducibilityResult, reasoningStatusResult, survivalResult, convictionResult, reliabilityResult, competitionResult, ensembleResult, benchmarkRelativeResult, trainingQualityResult] = await Promise.allSettled([
         api.learningDashboard(),
         api.learningRuns(20),
         api.learningPredictions(36),
@@ -47,6 +47,7 @@ export default function LearningPage() {
         api.liveTradingGamePositions(),
         api.liveTradingGameMetrics(),
         api.learningIntelligenceDashboard(),
+        api.decisionIntelligenceDashboard(),
         api.tradingGameLearningEvidence(60),
         api.tradingGameRealityCheck(),
         api.tradingGamePnlBreakdown(),
@@ -88,6 +89,7 @@ export default function LearningPage() {
         livePositions: livePositionsResult.status === "fulfilled" ? livePositionsResult.value : null,
         liveMetrics: liveMetricsResult.status === "fulfilled" ? liveMetricsResult.value : null,
         learningIntelligence: learningIntelligenceResult.status === "fulfilled" ? learningIntelligenceResult.value : null,
+        decisionIntelligence: decisionIntelligenceResult.status === "fulfilled" ? decisionIntelligenceResult.value : null,
         learningEvidence: learningEvidenceResult.status === "fulfilled" ? learningEvidenceResult.value : null,
         realityCheck: realityCheckResult.status === "fulfilled" ? realityCheckResult.value : null,
         pnlBreakdown: pnlBreakdownResult.status === "fulfilled" ? pnlBreakdownResult.value : null,
@@ -145,6 +147,13 @@ export default function LearningPage() {
   const livePositions = Array.isArray(livePositionRows) ? livePositionRows : [];
   const liveMetrics = trading?.liveMetrics?.metrics ?? {};
   const learningIntelligence = trading?.learningIntelligence ?? {};
+  const decisionIntelligence = trading?.decisionIntelligence ?? {};
+  const decisionSuperiority = decisionIntelligence?.decision?.decision_superiority ?? {};
+  const missedOpportunityRows = decisionIntelligence?.decision?.top_missed_opportunities ?? [];
+  const businessQualityRows = decisionIntelligence?.business_quality?.highest_quality_companies ?? [];
+  const portfolioQuality = decisionIntelligence?.portfolio?.portfolio_quality ?? {};
+  const portfolioContributionRows = decisionIntelligence?.portfolio?.contributions ?? [];
+  const portfolioCorrelationRows = decisionIntelligence?.portfolio?.correlations ?? [];
   const tradingPower = learningIntelligence?.trading_power ?? {};
   const tradingPowerComponents = tradingPower?.components ?? {};
   const benchmarkArenaRows = learningIntelligence?.benchmarks?.rows ?? [];
@@ -270,6 +279,51 @@ export default function LearningPage() {
         </div>
       </section>
 
+      <section className="grid-3" style={{ marginTop: 12 }}>
+        <div className="panel lab-command-panel">
+          <div className="panel-head"><span>Decision Superiority</span><strong>{formatNumber(decisionSuperiority?.score)}/100</strong></div>
+          <div className="cycle-progress-track"><span style={{ width: `${Math.max(0, Math.min(100, Number(decisionSuperiority?.score ?? 0)))}%` }} /></div>
+          <div className="evidence-grid">
+            <SmallDatum label="Classification" value={decisionSuperiority?.classification ?? "insufficient evidence"} />
+            <SmallDatum label="Recall" value={formatPct(decisionSuperiority?.metrics?.opportunity_recall)} />
+            <SmallDatum label="Precision" value={formatPct(decisionSuperiority?.metrics?.opportunity_precision)} />
+            <SmallDatum label="Alpha Capture" value={formatPct(decisionSuperiority?.metrics?.alpha_capture_rate)} />
+            <SmallDatum label="Ranking" value={formatPct(decisionSuperiority?.metrics?.ranking_accuracy)} />
+            <SmallDatum label="Comparable Decisions" value={decisionSuperiority?.sample_size ?? 0} />
+          </div>
+          <p>{decisionSuperiority?.explanation ?? "BLUM needs comparable decision snapshots before it can claim it selected superior opportunities."}</p>
+          {missedOpportunityRows.length > 0 && <div className="tag-row">{missedOpportunityRows.slice(0, 4).map((row: any) => <span key={`${row.trade_id}-${row.best_available_ticker}`}>Missed {row.best_available_ticker} vs {row.ticker}</span>)}</div>}
+        </div>
+
+        <div className="panel lab-command-panel">
+          <div className="panel-head"><span>Business Quality Lab</span><strong>{businessQualityRows.length}</strong></div>
+          <div className="brain-list dense">
+            {(businessQualityRows.length ? businessQualityRows : []).slice(0, 5).map((row: any) => (
+              <div key={row.ticker}>
+                <StatusBadge label={`${row.ticker} | ${row.status}`} />
+                <div className="opportunity-line"><strong>{row.name}</strong><span>{formatNumber(row.business_quality_score)}/100</span></div>
+                <p>Growth {formatNumber(row.growth_quality)} | FCF {formatNumber(row.cash_flow_quality)} | moat {formatNumber(row.moat_quality)} | data {formatNumber(row.data_quality_score)}</p>
+              </div>
+            ))}
+            {businessQualityRows.length === 0 && <div className="empty-state compact">No business-quality evidence yet. Stored fundamental snapshots are required.</div>}
+          </div>
+        </div>
+
+        <div className="panel lab-command-panel">
+          <div className="panel-head"><span>Portfolio Intelligence</span><strong>{formatNumber(portfolioQuality?.score)}/100</strong></div>
+          <div className="evidence-grid">
+            <SmallDatum label="Diversification" value={formatNumber(portfolioQuality?.components?.diversification)} />
+            <SmallDatum label="Concentration Risk" value={formatNumber(portfolioQuality?.components?.concentration_risk)} />
+            <SmallDatum label="Drawdown Control" value={formatNumber(portfolioQuality?.components?.drawdown_control)} />
+            <SmallDatum label="Alpha Generation" value={formatNumber(portfolioQuality?.components?.alpha_generation)} />
+            <SmallDatum label="Contributors" value={portfolioContributionRows.length} />
+            <SmallDatum label="Correlations" value={portfolioCorrelationRows.length} />
+          </div>
+          <p>{portfolioQuality?.explanation ?? "BLUM needs portfolio trade evidence before contribution, concentration and portfolio-alpha quality can be measured."}</p>
+          {(portfolioQuality?.warnings ?? []).length > 0 && <div className="tag-row">{portfolioQuality.warnings.slice(0, 4).map((item: string) => <span key={item}>{item}</span>)}</div>}
+        </div>
+      </section>
+
       <section className="grid-3" style={{ marginTop: 12 }}>
         <div className="panel">
           <div className="panel-head"><span>Benchmark Arena</span><strong>{benchmarkArenaRows.length}</strong></div>

frontend/lib/api.ts

@@ -128,6 +128,18 @@ export const api = {
   generateLearningSelfImprovement: () => postJson<any>("/api/learning-intelligence/self-improvement/generate", {}),
   applyLearningSelfImprovement: (actionId: number) => postJson<any>(`/api/learning-intelligence/self-improvement/apply/${actionId}`, {}),
   evaluateLearningSelfImprovement: (actionId: number) => postJson<any>(`/api/learning-intelligence/self-improvement/evaluate/${actionId}`, {}),
+  decisionIntelligenceDashboard: () => getJson<any>("/api/decision-intelligence/dashboard"),
+  decisionSuperiority: () => getJson<any>("/api/decision-intelligence/superiority"),
+  recalculateDecisionSuperiority: () => postJson<any>("/api/decision-intelligence/superiority/recalculate", {}),
+  decisionUniverseSnapshots: () => getJson<any>("/api/decision-intelligence/universe-snapshots"),
+  recalculateDecisionUniverseSnapshots: () => postJson<any>("/api/decision-intelligence/universe-snapshots/recalculate", {}),
+  missedOpportunities: () => getJson<any>("/api/decision-intelligence/missed-opportunities"),
+  businessQualityDashboard: () => getJson<any>("/api/business-quality/dashboard"),
+  businessQualityScores: (limit = 80) => getJson<any>(`/api/business-quality/scores?limit=${limit}`),
+  recalculateBusinessQuality: (limit = 80) => postJson<any>(`/api/business-quality/recalculate?limit=${limit}`, {}),
+  portfolioIntelligenceDashboard: () => getJson<any>("/api/portfolio-intelligence/dashboard"),
+  portfolioQuality: () => getJson<any>("/api/portfolio-intelligence/quality"),
+  recalculatePortfolioIntelligence: () => postJson<any>("/api/portfolio-intelligence/recalculate", {}),
   chartAnalyzeTicker: (ticker: string, timeframe = "6M", period = "1y", includeVisual = false) => getPostChart<ChartReport>(`/chart/analyze-ticker?ticker=${encodeURIComponent(ticker)}&timeframe=${encodeURIComponent(timeframe)}&period=${encodeURIComponent(period)}&include_visual=${includeVisual ? "true" : "false"}`),
   chartTechnicalReport: (ticker: string, timeframe = "6M") => getJson<ChartReport>(`/chart/technical-report/${encodeURIComponent(ticker)}?timeframe=${encodeURIComponent(timeframe)}`),
   chartLevels: (ticker: string, timeframe = "6M") => getJson<any>(`/chart/levels/${encodeURIComponent(ticker)}?timeframe=${encodeURIComponent(timeframe)}`),

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "blum-ai-financial-intelligence-frontend",
-  "version": "0.16.0",
+  "version": "0.17.0",
   "private": true,
   "scripts": {
     "dev": "next dev -p 3000",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "blum-ai-financial-intelligence",
-  "version": "0.16.0",
+  "version": "0.17.0",
   "private": true,
   "scripts": {
     "frontend:dev": "npm --prefix frontend run dev",