Upload 2 files

__init__.py

@@ -0,0 +1,58 @@
+"""
+Schemas Package - Data Structures for Three-Layer Architecture
+
+This package contains pure data structures (dataclasses) that define
+the contracts between layers:
+
+- analysis_intent.py: Layer 1 → Layer 2 contract
+- decision_result.py: Layer 2 → Layer 3 contract
+
+IMPORTANT: No business logic in this package. Only data definitions.
+"""
+
+from schemas.analysis_intent import (
+    AnalysisIntent,
+    AnalysisType,
+    AnalysisPurpose,
+    UserPreferences,
+    HardConstraints,
+    ExtractedDataSummary,
+)
+
+from schemas.decision_result import (
+    RegulatoryDecisionResult,
+    RefusalResult,
+    RefusalSeverity,
+    DataQuality,
+    DataQualityReport,
+    KineticFitSummary,
+    PredictionSummary,
+    ArrheniusResult,
+    TrendTransferResult,
+    BatchRankingItem,
+    RegulatoryNotes,
+    CalculationTrace,
+)
+
+__all__ = [
+    # Intent structures
+    "AnalysisIntent",
+    "AnalysisType",
+    "AnalysisPurpose",
+    "UserPreferences",
+    "HardConstraints",
+    "ExtractedDataSummary",
+    # Decision structures
+    "RegulatoryDecisionResult",
+    "RefusalResult",
+    "RefusalSeverity",
+    "DataQuality",
+    "DataQualityReport",
+    "KineticFitSummary",
+    "PredictionSummary",
+    "ArrheniusResult",
+    "TrendTransferResult",
+    "BatchRankingItem",
+    "RegulatoryNotes",
+    "CalculationTrace",
+]

decision_result.py

@@ -0,0 +1,225 @@
+"""
+Decision Result Schema - Scientific & Regulatory Decision Layer Output
+
+This module defines the structured output of Layer 2 (RegulatoryDecisionEngine).
+Used as input to Layer 3 (Presentation Layer).
+
+IMPORTANT: This is a DATA STRUCTURE ONLY. No business logic here.
+"""
+
+from dataclasses import dataclass, field
+from typing import List, Dict, Optional, Literal, Any
+from enum import Enum
+
+
+class DataQuality(str, Enum):
+    """Data quality assessment levels."""
+    SUFFICIENT = "sufficient"      # All requirements met
+    MARGINAL = "marginal"          # Minimum requirements met with caveats
+    INSUFFICIENT = "insufficient"  # Cannot proceed
+
+
+class RefusalSeverity(str, Enum):
+    """Severity of refusal - affects how Layer 3 presents results."""
+    HARD_REFUSAL = "hard_refusal"  # No numbers at all, only explanation
+    SOFT_WARNING = "soft_warning"  # Numbers shown with "reference only" caveat
+
+
+@dataclass
+class RefusalResult:
+    """
+    Detailed refusal information when analysis cannot proceed.
+    
+    This ensures "refusal" is a first-class output, not just "return None".
+    """
+    severity: RefusalSeverity
+    reason: str
+    regulatory_reference: str  # e.g., "ICH Q1E Section 4.2"
+    suggestions: List[str] = field(default_factory=list)  # How to fix the issue
+    missing_data: List[str] = field(default_factory=list)  # Specific missing items
+
+
+@dataclass
+class KineticFitSummary:
+    """Summary of a single kinetic fit result."""
+    condition_id: str
+    model_type: Literal["zero_order", "first_order"]
+    k: float
+    k_unit: str
+    y0: float
+    R2: float
+    SE_k: float
+    n_points: int
+    equation: str
+    confidence_level: Literal["high", "medium", "low"]
+    
+    # Audit trail
+    scipy_method: str = "linregress"
+    calculation_timestamp: str = ""
+
+
+@dataclass
+class PredictionSummary:
+    """Summary of a single prediction result."""
+    timepoint_months: int
+    point_estimate: float
+    CI_lower: float
+    CI_upper: float
+    risk_level: Literal["compliant", "marginal", "non_compliant"]
+    specification_limit: float
+    margin_to_limit: float  # spec_limit - CI_upper
+    
+    # Validity - Added for Refusal Logic
+    is_valid: bool = True
+    validity_reason: str = ""  # Reason if invalid (e.g. "Extrapolation > 2x")
+    
+    def is_compliant(self) -> bool:
+        return self.is_valid and self.CI_upper < self.specification_limit
+
+
+@dataclass
+class ArrheniusResult:
+    """Arrhenius acceleration factor calculation result."""
+    is_calculated: bool
+    reference_batch: Optional[str] = None
+    empirical_AF: Optional[float] = None
+    theoretical_AF: float = 13.4  # Default for 40°C/25°C
+    deviation_percent: Optional[float] = None
+    calculation_method: str = ""
+    
+    # Validation
+    is_valid: bool = True
+    validity_reason: str = ""
+
+
+@dataclass
+class TrendTransferResult:
+    """Cross-formulation trend transfer result."""
+    is_applicable: bool
+    source_batch: Optional[str] = None
+    target_batch: Optional[str] = None
+    transferred_k: Optional[float] = None
+    confidence_level: Literal["high", "medium", "low", "not_applicable"] = "not_applicable"
+    method: str = ""
+
+
+@dataclass
+class BatchRankingItem:
+    """Single item in batch ranking."""
+    rank: int
+    batch_id: str
+    batch_name: str
+    score: float
+    reason: str
+    k_best: Optional[float]
+    r2_best: Optional[float]
+    
+    # Evaluation Logic
+    data_completeness: Literal["full_trend", "partial", "single_point"] = "partial"
+    confidence: Literal["high", "medium", "low"] = "low"
+
+
+
+@dataclass
+class DataQualityReport:
+    """
+    Detailed data quality assessment.
+    Produced early in Layer 2 before any calculations.
+    """
+    overall_quality: DataQuality
+    n_batches: int
+    n_conditions: int
+    n_total_datapoints: int
+    
+    # Per-condition assessment
+    conditions_with_sufficient_data: List[str] = field(default_factory=list)
+    conditions_with_insufficient_data: List[str] = field(default_factory=list)
+    
+    # Specific issues
+    issues: List[str] = field(default_factory=list)
+    warnings: List[str] = field(default_factory=list)
+
+
+@dataclass
+class RegulatoryNotes:
+    """Regulatory compliance notes for the report."""
+    applicable_guidelines: List[str] = field(default_factory=lambda: [
+        "ICH Q1A(R2)",
+        "ICH Q1E",
+        "WHO TRS 953"
+    ])
+    extrapolation_statement: str = ""
+    statistical_method_statement: str = ""
+    limitations: List[str] = field(default_factory=list)
+    disclaimers: List[str] = field(default_factory=list)
+
+
+@dataclass
+class CalculationTrace:
+    """
+    Audit trail for all calculations.
+    Used for FDA Data Integrity compliance.
+    """
+    entries: List[Dict[str, Any]] = field(default_factory=list)
+    
+    def add(self, step: str, inputs: Dict, outputs: Dict, method: str):
+        self.entries.append({
+            "step": step,
+            "inputs": inputs,
+            "outputs": outputs,
+            "method": method
+        })
+
+
+@dataclass
+class RegulatoryDecisionResult:
+    """
+    Complete output of Layer 2 (Scientific & Regulatory Decision Engine).
+    
+    This is the CONTRACT between Layer 2 and Layer 3.
+    Layer 2 produces this; Layer 3 consumes this for presentation.
+    
+    IMPORTANT: Layer 3 must NOT access raw data - only this structure.
+    """
+    # Decision status
+    can_proceed: bool
+    refusal: Optional[RefusalResult] = None  # Only if can_proceed=False
+    
+    # Data assessment (always provided)
+    data_quality: DataQualityReport = field(default_factory=DataQualityReport)
+    
+    # Analysis results (only if can_proceed=True)
+    analysis_type_executed: str = ""
+    kinetic_fits: Dict[str, KineticFitSummary] = field(default_factory=dict)
+    arrhenius: Optional[ArrheniusResult] = None
+    trend_transfer: Optional[TrendTransferResult] = None
+    predictions: Dict[str, PredictionSummary] = field(default_factory=dict)
+    
+    # Batch ranking (only for batch_comparison type)
+    batch_ranking: List[BatchRankingItem] = field(default_factory=list)
+    
+    # Regulatory compliance
+    regulatory_notes: RegulatoryNotes = field(default_factory=RegulatoryNotes)
+    
+    # Audit trail (for FDA/EMA inspection)
+    calculation_trace: CalculationTrace = field(default_factory=CalculationTrace)
+    
+    # Metadata
+    engine_version: str = "2.0.0"
+    timestamp: str = ""
+
+    def get_executive_summary(self) -> str:
+        """Generate one-line executive summary for Layer 3."""
+        if not self.can_proceed:
+            return f"分析无法完成: {self.refusal.reason if self.refusal else '未知原因'}"
+        
+        if self.predictions:
+            compliant = sum(1 for p in self.predictions.values() if p.is_compliant())
+            total = len(self.predictions)
+            return f"预测完成: {compliant}/{total} 时间点符合规格"
+        
+        if self.batch_ranking:
+            top = self.batch_ranking[0] if self.batch_ranking else {}
+            return f"批次筛选完成: 推荐批次 {top.get('batch_id', 'N/A')}"
+        
+        return "分析完成"