Matching Engine¶

class MatchingStrategy(ABC):
    """Base class for all matching strategies."""

    @abstractmethod
    async def match(
        self,
        case: Case,
        clinical_data: ClinicalData,
        preferences: PatientPreferences,
        tenant_id: str,
    ) -> MatchResult:
        """Execute matching and return scored results."""
        ...

    @abstractmethod
    def get_version(self) -> str:
        """Return strategy version identifier."""
        ...

    @abstractmethod
    def get_scoring_dimensions(self) -> list[str]:
        """Return list of scoring dimensions used."""
        ...

STRATEGY_WEIGHTS = {
    "clinical_relevance": 0.25,
    "outcome_score": 0.20,
    "cost_score": 0.15,
    "semantic_match": 0.10,
    "travel_logistics": 0.10,
    "accreditation": 0.10,
    "patient_preferences": 0.10,
}

MATCH (c:Condition {code: $condition_code})-[:REQUIRES]->(proc:Procedure)
      <-[offers:OFFERS]-(prov:Provider)-[:LOCATED_IN]->(loc:Location)
WHERE prov.tenant_id = $tenant_id
  AND prov.is_active = true
OPTIONAL MATCH (prov)-[:HAS_ACCREDITATION]->(acc:Accreditation)
RETURN prov, proc, offers, loc, COLLECT(DISTINCT acc) AS accreditations

offers_data = {
    "cost_usd": relationship["cost_usd"],
    "annual_volume": relationship["annual_volume"],
    "success_rate": relationship["success_rate"],
    "average_los_days": relationship["average_los_days"],
    "wait_time_weeks": relationship["wait_time_weeks"],
    "package_includes": relationship["package_includes"],
}

async def score_outcome(
    provider: Provider,
    offers_data: dict,
    procedure_code: str,
) -> float:
    """Score provider outcomes for the specific procedure."""
    success_rate = offers_data.get("success_rate", 0)
    volume = offers_data.get("annual_volume", 0)

    # Volume-based confidence: more procedures = more reliable data
    volume_confidence = min(volume / 200, 1.0)  # Caps at 200/year

    # Weighted combination
    raw_score = (success_rate / 100) * 0.7 + volume_confidence * 0.3

    return round(raw_score, 4)

def score_language_concordance(
    patient_language: str,
    doctor_languages: list[dict],
    provider_language_services: dict,
) -> float:
    """Score language match between patient and doctor."""
    for lang in doctor_languages:
        if lang["language"].lower() == patient_language.lower():
            proficiency_scores = {
                "native": 1.00,
                "fluent": 0.85,
                "professional": 0.70,
                "conversational": 0.50,
            }
            return proficiency_scores.get(lang["proficiency"], 0.50)

    # Check provider-level interpretation services
    interpreters = provider_language_services.get("medical_interpreters", [])
    if patient_language in interpreters:
        return 0.30

    return 0.00

def apply_completeness_factor(
    raw_score: float,
    data_completeness: dict,
) -> float:
    """Adjust score based on data completeness confidence."""
    overall_completeness = data_completeness.get("overall", 0.5)

    # Minimum floor of 0.5 to avoid penalizing new doctors too heavily
    confidence = max(overall_completeness, 0.5)

    # Blend raw score toward 0.5 (neutral) based on missing data
    adjusted = raw_score * confidence + 0.5 * (1 - confidence)

    return round(adjusted, 4)

REASONING_TEMPLATES = {
    "high_volume_specialist": (
        "Dr. {name} has performed {volume} {procedure} procedures annually "
        "with a {success_rate}% success rate, placing them in the top tier "
        "of specialists at {provider_name}."
    ),
    "language_concordance": (
        "Dr. {name} speaks {language} at {proficiency} level, enabling "
        "direct communication without interpretation."
    ),
    "technique_match": (
        "Dr. {name} specializes in {technique} {procedure}, which is the "
        "recommended approach for your specific condition profile."
    ),
}

def redistribute_weights(
    base_weights: dict[str, float],
    available_dimensions: set[str],
) -> dict[str, float]:
    """Redistribute weights from unavailable dimensions."""
    unavailable = set(base_weights.keys()) - available_dimensions
    if not unavailable:
        return base_weights

    total_unavailable_weight = sum(base_weights[d] for d in unavailable)
    total_available_weight = sum(base_weights[d] for d in available_dimensions)

    redistributed = {}
    for dim in available_dimensions:
        original = base_weights[dim]
        share = original / total_available_weight
        redistributed[dim] = original + (total_unavailable_weight * share)

    return redistributed

async def match_with_shadow(
    case: Case,
    clinical_data: ClinicalData,
    preferences: PatientPreferences,
    tenant_id: str,
) -> MatchResult:
    """Run active strategy + optional shadow strategy."""
    # Active strategy (returned to patient)
    active_result = await active_strategy.match(case, clinical_data, preferences, tenant_id)

    # Shadow strategy (logged, not returned)
    shadow_flag = await flagsmith.get_flag("matching_shadow_strategy", tenant_id)
    if shadow_flag:
        shadow_strategy = get_strategy(shadow_flag)
        shadow_result = await shadow_strategy.match(case, clinical_data, preferences, tenant_id)
        await log_shadow_comparison(active_result, shadow_result, case.id)

    return active_result

class ProcedureRequirement(BaseModel):
    """A document requirement for a procedure."""
    id: UUID
    tenant_id: UUID
    procedure_code: str                     # CPT code
    requirement_type: str                   # "diagnostic", "lab", "imaging", "clearance"
    name: str                               # "Complete Blood Count"
    description: str                        # Detailed description for matching
    max_age_days: int                       # Maximum acceptable age of the document
    is_mandatory: bool = True
    alternatives: list[str] = []            # Alternative acceptable documents

class ProviderRequirementOverride(BaseModel):
    """Provider-specific override to a base requirement."""
    id: UUID
    tenant_id: UUID
    provider_id: UUID
    procedure_requirement_id: UUID
    max_age_days_override: Optional[int]    # Provider accepts older docs
    is_mandatory_override: Optional[bool]   # Provider makes it optional
    additional_notes: Optional[str]         # Provider-specific instructions
    additional_requirements: list[dict]     # Extra tests this provider needs

class MatchResult(BaseModel):
    """Complete match result for a case."""
    id: UUID
    case_id: UUID
    tenant_id: UUID
    strategy_version: str
    strategy_weights: dict[str, float]
    providers: list[ProviderMatch]
    doctors: Optional[list[DoctorMatch]]    # Only if DOCTORS_IN_MATCHING
    executed_at: datetime
    execution_time_ms: int
    shadow_strategy_version: Optional[str]

class ProviderMatch(BaseModel):
    """Individual provider match with scores."""
    provider_id: UUID
    rank: int
    overall_score: float                    # 0.0 - 1.0
    dimension_scores: dict[str, float]      # Per-dimension scores
    weights_used: dict[str, float]          # Actual weights (after redistribution)
    reasoning: str                          # Human-readable explanation
    strengths: list[str]
    considerations: list[str]

class DoctorMatch(BaseModel):
    """Individual doctor match within a provider."""
    doctor_id: UUID
    provider_id: UUID
    rank: int
    overall_score: float
    language_concordance: float
    procedure_metrics: dict
    data_completeness_score: float
    reasoning: str

# config/scoring.yaml
pfs:
  weights:
    medical: 0.45        # Risk factors + missing mandatory docs
    travel: 0.25         # Transport tier (T1-T4)
    logistics: 0.15      # Passport, visa, timeline feasibility
    intent_modifier: 0.15 # Decision stage + emotional readiness

# Step 1: Scale match_score from 0-1 to 0-100
base_hss = match_score * 100.0

# Step 2: Apply cost multiplier based on Patient Pricing Band (PV) alignment
if cost_within_pv:
    cost_multiplier = 1.0              # Full score
elif cost_in_soft_ceiling:             # Within 120% of PV max
    cost_multiplier = 0.5              # Partial score
else:
    cost_multiplier = 0.0              # HARD GATE — outside PV

# Step 3: Apply facility readiness penalty
facility_penalty = 0.15 if critical_facility_gap else 0.0

# Step 4: Combine
HSS = (base_hss * cost_multiplier) - (facility_penalty * 100)
HSS = clamp(0, 100)

# config/scoring.yaml
hss:
  cost_alignment:
    soft_ceiling_multiplier: 1.2  # 120% of PV max
    soft_ceiling_score: 0.5       # Partial score for soft-ceiling match
    outside_range_score: 0.0      # Fully outside PV = gated out

  facility_readiness:
    critical_gap_penalty: 0.15    # Subtract 15 points from HSS

# Normalize inputs to 0-1
pfs_n = pfs / 100.0
hss_n = hss / 100.0

# Weighted linear combination
linear = (
    0.30 * pfs_n                    # Patient readiness
  + 0.35 * hss_n                    # Hospital fit
  + 0.15 * preference_alignment     # Cultural/dietary/language match (0-1)
  + 0.20 * historical_conversion    # Learned signal from feedback_records (0-1)
  + bias                            # -0.5 (shifts sigmoid center)
)

# Sigmoid produces 0-1 probability
FMS = sigmoid(linear * 4)

Patient intake (Triage Agent, 5 layers)
    ↓
PFS computed (per turn, from layer state + risk items)
    ↓
PFS band gates matching:
    not_ready → no matching
    needs_attention → matching blocked
    conditionally_ready / ready → proceed
    ↓
Matching Engine (GraphEnhancedWeightedV1 → 0-1 per provider)
    ↓
HSS computed (per provider: cost gate + facility penalty → 0-100)
    ↓
FMS computed (PFS × HSS × preferences × history → 0-1 sigmoid)
    ↓
Results ranked by FMS, displayed with HSS breakdown