v2.0: phd_research_os_v2/layer5/scorer.py

phd_research_os_v2/layer5/scorer.py

@@ -0,0 +1,186 @@
+"""
+Layer 5: Code-Computed Calibrated Scoring
+===========================================
+The LLM provides COMPONENTS. The CODE computes FINAL SCORES.
+Three separate scores: evidence_quality, truth_likelihood, qualifier_strength.
+"""
+
+import json
+from ..core.database import get_db, to_fixed, from_fixed, now_iso
+
+# Study quality weights (Quantum-Bio V2 taxonomy)
+STUDY_QUALITY_WEIGHTS = {
+    "in_vivo": 1000,
+    "direct_physical_measurement": 1000,
+    "mathematical_proof": 950,
+    "in_vitro": 850,
+    "first_principles_simulation": 800,
+    "phenomenological_simulation": 600,
+    "review": 400,
+    "perspective": 200,
+    # Legacy mappings
+    "primary_experimental": 1000,
+    "simulation": 600,
+    "review_non_systematic": 400,
+    "meta_analysis": 1000,
+    "case_study": 300,
+}
+
+JOURNAL_TIER_WEIGHTS = {1: 1000, 2: 850, 3: 700, 0: 500}  # 0 = preprint
+
+SECTION_MODIFIERS = {
+    "abstract": 700,
+    "introduction": 800,
+    "methods": 1000,
+    "results": 1000,
+    "results_discussion": 900,
+    "discussion": 750,
+    "conclusion": 800,
+    "supplement": 1000,
+    "unknown": 850,
+    None: 850,
+}
+
+
+class CalibratedScorer:
+    """
+    Code-computed scoring engine.
+    
+    The LLM NEVER sets final confidence directly.
+    This module computes all scores from components.
+    """
+    
+    def __init__(self, db_path: str = None):
+        self.db_path = db_path
+    
+    def score_claim(self, claim: dict, source: dict = None) -> dict:
+        """
+        Compute three separate scores for a claim.
+        All arithmetic uses fixed-point integers (×1000).
+        """
+        # ── Components ──
+        evidence_strength = claim.get("evidence_strength", 500)
+        study_type = source.get("study_type", "unknown") if source else "unknown"
+        journal_tier = source.get("journal_tier", 2) if source else 2
+        section = claim.get("source_section", "unknown")
+        missing_fields = claim.get("missing_fields", [])
+        if isinstance(missing_fields, str):
+            missing_fields = json.loads(missing_fields)
+        qualifiers = claim.get("qualifiers", [])
+        if isinstance(qualifiers, str):
+            qualifiers = json.loads(qualifiers)
+        parse_confidence = claim.get("parse_confidence", 1000)
+        is_null = claim.get("is_null_result", False)
+        is_inherited = claim.get("is_inherited_citation", False)
+        
+        # ── Score 1: Evidence Quality ──
+        sq_weight = STUDY_QUALITY_WEIGHTS.get(study_type, 600)
+        jt_weight = JOURNAL_TIER_WEIGHTS.get(journal_tier, 700)
+        completeness = 700 if missing_fields else 1000
+        section_mod = SECTION_MODIFIERS.get(section, 850)
+        
+        evidence_quality = (evidence_strength * sq_weight // 1000
+                           * jt_weight // 1000
+                           * completeness // 1000
+                           * section_mod // 1000)
+        
+        # Parser confidence CAPS evidence quality
+        evidence_quality = min(evidence_quality, parse_confidence)
+        
+        # Statistical evidence gate
+        practical_sig = True
+        effect_size = claim.get("stat_effect_size")
+        sample_size = claim.get("stat_sample_size")
+        if effect_size is not None and sample_size is not None:
+            if sample_size > 1000 and abs(effect_size) < 0.1:
+                evidence_quality = min(evidence_quality, 400)
+                practical_sig = False
+        
+        # ── Score 2: Truth Likelihood ──
+        # Start from evidence quality, adjust for corroboration and conflicts
+        truth_likelihood = evidence_quality
+        
+        # Null result penalty
+        if is_null:
+            truth_likelihood = min(truth_likelihood, 500)
+        
+        # Inherited citation penalty
+        if is_inherited:
+            truth_likelihood -= 200
+        
+        truth_likelihood = max(0, min(1000, truth_likelihood))
+        
+        # ── Score 3: Qualifier Strength ──
+        qualifier_strength = 1000
+        if qualifiers:
+            qualifier_strength -= len(qualifiers) * 100
+        if is_null:
+            qualifier_strength = min(qualifier_strength, 500)
+        if is_inherited:
+            qualifier_strength -= 200
+        qualifier_strength = max(0, min(1000, qualifier_strength))
+        
+        # ── Composite ──
+        composite = (evidence_quality + truth_likelihood + qualifier_strength) // 3
+        
+        return {
+            "evidence_quality": evidence_quality,
+            "truth_likelihood": truth_likelihood,
+            "qualifier_strength_score": qualifier_strength,
+            "composite_confidence": composite,
+            "practical_significance": practical_sig,
+            "components": {
+                "evidence_strength": evidence_strength,
+                "study_quality_weight": sq_weight,
+                "journal_tier_weight": jt_weight,
+                "completeness_penalty": completeness,
+                "section_modifier": section_mod,
+                "parse_confidence": parse_confidence,
+            }
+        }
+    
+    def rescore_all_claims(self) -> int:
+        """Rescore all claims in the database. Returns number rescored."""
+        conn = get_db(self.db_path)
+        claims = conn.execute("SELECT * FROM claims").fetchall()
+        count = 0
+        
+        for row in claims:
+            claim = dict(row)
+            claim["missing_fields"] = json.loads(claim.get("missing_fields", "[]"))
+            claim["qualifiers"] = json.loads(claim.get("qualifiers", "[]"))
+            
+            # Get source info
+            source = None
+            if claim.get("source_doi"):
+                src_row = conn.execute(
+                    "SELECT * FROM sources WHERE doi = ?", (claim["source_doi"],)
+                ).fetchone()
+                if src_row:
+                    source = dict(src_row)
+            
+            scores = self.score_claim(claim, source)
+            
+            conn.execute("""
+                UPDATE claims SET
+                    evidence_quality = ?,
+                    truth_likelihood = ?,
+                    qualifier_strength_score = ?,
+                    composite_confidence = ?,
+                    practical_significance = ?,
+                    updated_at = ?
+                WHERE claim_id = ?
+            """, (
+                scores["evidence_quality"],
+                scores["truth_likelihood"],
+                scores["qualifier_strength_score"],
+                scores["composite_confidence"],
+                int(scores["practical_significance"]),
+                now_iso(),
+                claim["claim_id"],
+            ))
+            count += 1
+        
+        conn.commit()
+        conn.close()
+        return count