# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
# healthcare/application/transformers/clinical_nlp.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator
import torch
from transformers import (
AutoTokenizer,
AutoModelForTokenClassification,
AutoModelForSeq2SeqLM,
pipeline,
Pipeline,
)
@dataclass
class ClinicalEntity:
text: str
label: str # e.g. "PROBLEM", "TREATMENT", "TEST"
score: float
start: int
end: int
class ClinicalNERExtractor:
"""
Named-entity recognition over clinical notes using a
fine-tuned BioBERT model. Extracts problems, treatments,
and tests as structured ClinicalEntity objects.
"""
MODEL_ID = "blaze999/Medical-NER"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"ner",
model=self.MODEL_ID,
aggregation_strategy="simple",
device=0 if device == "cuda" else -1,
)
def extract(self, note: str) -> list[ClinicalEntity]:
raw = self._pipe(note)
return [
ClinicalEntity(
text=e["word"],
label=e["entity_group"],
score=round(e["score"], 4),
start=e["start"],
end=e["end"],
)
for e in raw
if e["score"] >= 0.80 # confidence threshold
]
def problems(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "PROBLEM"]
def treatments(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "TREATMENT"]
class ClinicalSummarizer:
"""
Abstractive summariser for clinical discharge notes.
Uses a BART model fine-tuned on clinical text (MedBART).
"""
MODEL_ID = "Falconsai/medical_summarization"
MAX_INPUT = 1024
MAX_SUMMARY = 256
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"summarization",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def summarize(self, note: str) -> str:
chunks = self._chunk(note)
summaries = [
self._pipe(
chunk,
max_length=self.MAX_SUMMARY,
min_length=40,
do_sample=False,
)[0]["summary_text"]
for chunk in chunks
]
return " ".join(summaries)
def _chunk(self, text: str) -> Iterator[str]:
words = text.split()
for i in range(0, len(words), self.MAX_INPUT):
yield " ".join(words[i : i + self.MAX_INPUT])
class ClinicalQA:
"""
Extractive question-answering over patient notes.
Allows clinicians to query free-text records naturally.
e.g. "What medications is the patient currently taking?"
"""
MODEL_ID = "deepset/roberta-base-squad2"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"question-answering",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def ask(self, question: str, context: str) -> dict:
result = self._pipe(question=question, context=context)
return {
"answer": result["answer"],
"score": round(result["score"], 4),
"start": result["start"],
"end": result["end"],
}
# healthcare/application/transformers/clinical_nlp.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator
import torch
from transformers import (
AutoTokenizer,
AutoModelForTokenClassification,
AutoModelForSeq2SeqLM,
pipeline,
Pipeline,
)
@dataclass
class ClinicalEntity:
text: str
label: str # e.g. "PROBLEM", "TREATMENT", "TEST"
score: float
start: int
end: int
class ClinicalNERExtractor:
"""
Named-entity recognition over clinical notes using a
fine-tuned BioBERT model. Extracts problems, treatments,
and tests as structured ClinicalEntity objects.
"""
MODEL_ID = "blaze999/Medical-NER"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"ner",
model=self.MODEL_ID,
aggregation_strategy="simple",
device=0 if device == "cuda" else -1,
)
def extract(self, note: str) -> list[ClinicalEntity]:
raw = self._pipe(note)
return [
ClinicalEntity(
text=e["word"],
label=e["entity_group"],
score=round(e["score"], 4),
start=e["start"],
end=e["end"],
)
for e in raw
if e["score"] >= 0.80 # confidence threshold
]
def problems(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "PROBLEM"]
def treatments(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "TREATMENT"]
class ClinicalSummarizer:
"""
Abstractive summariser for clinical discharge notes.
Uses a BART model fine-tuned on clinical text (MedBART).
"""
MODEL_ID = "Falconsai/medical_summarization"
MAX_INPUT = 1024
MAX_SUMMARY = 256
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"summarization",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def summarize(self, note: str) -> str:
chunks = self._chunk(note)
summaries = [
self._pipe(
chunk,
max_length=self.MAX_SUMMARY,
min_length=40,
do_sample=False,
)[0]["summary_text"]
for chunk in chunks
]
return " ".join(summaries)
def _chunk(self, text: str) -> Iterator[str]:
words = text.split()
for i in range(0, len(words), self.MAX_INPUT):
yield " ".join(words[i : i + self.MAX_INPUT])
class ClinicalQA:
"""
Extractive question-answering over patient notes.
Allows clinicians to query free-text records naturally.
e.g. "What medications is the patient currently taking?"
"""
MODEL_ID = "deepset/roberta-base-squad2"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"question-answering",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def ask(self, question: str, context: str) -> dict:
result = self._pipe(question=question, context=context)
return {
"answer": result["answer"],
"score": round(result["score"], 4),
"start": result["start"],
"end": result["end"],
}
# healthcare/application/transformers/clinical_nlp.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator
import torch
from transformers import (
AutoTokenizer,
AutoModelForTokenClassification,
AutoModelForSeq2SeqLM,
pipeline,
Pipeline,
)
@dataclass
class ClinicalEntity:
text: str
label: str # e.g. "PROBLEM", "TREATMENT", "TEST"
score: float
start: int
end: int
class ClinicalNERExtractor:
"""
Named-entity recognition over clinical notes using a
fine-tuned BioBERT model. Extracts problems, treatments,
and tests as structured ClinicalEntity objects.
"""
MODEL_ID = "blaze999/Medical-NER"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"ner",
model=self.MODEL_ID,
aggregation_strategy="simple",
device=0 if device == "cuda" else -1,
)
def extract(self, note: str) -> list[ClinicalEntity]:
raw = self._pipe(note)
return [
ClinicalEntity(
text=e["word"],
label=e["entity_group"],
score=round(e["score"], 4),
start=e["start"],
end=e["end"],
)
for e in raw
if e["score"] >= 0.80 # confidence threshold
]
def problems(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "PROBLEM"]
def treatments(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "TREATMENT"]
class ClinicalSummarizer:
"""
Abstractive summariser for clinical discharge notes.
Uses a BART model fine-tuned on clinical text (MedBART).
"""
MODEL_ID = "Falconsai/medical_summarization"
MAX_INPUT = 1024
MAX_SUMMARY = 256
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"summarization",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def summarize(self, note: str) -> str:
chunks = self._chunk(note)
summaries = [
self._pipe(
chunk,
max_length=self.MAX_SUMMARY,
min_length=40,
do_sample=False,
)[0]["summary_text"]
for chunk in chunks
]
return " ".join(summaries)
def _chunk(self, text: str) -> Iterator[str]:
words = text.split()
for i in range(0, len(words), self.MAX_INPUT):
yield " ".join(words[i : i + self.MAX_INPUT])
class ClinicalQA:
"""
Extractive question-answering over patient notes.
Allows clinicians to query free-text records naturally.
e.g. "What medications is the patient currently taking?"
"""
MODEL_ID = "deepset/roberta-base-squad2"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"question-answering",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def ask(self, question: str, context: str) -> dict:
result = self._pipe(question=question, context=context)
return {
"answer": result["answer"],
"score": round(result["score"], 4),
"start": result["start"],
"end": result["end"],
}
# healthcare/application/transformers/clinical_nlp.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator
import torch
from transformers import (
AutoTokenizer,
AutoModelForTokenClassification,
AutoModelForSeq2SeqLM,
pipeline,
Pipeline,
)
@dataclass
class ClinicalEntity:
text: str
label: str # e.g. "PROBLEM", "TREATMENT", "TEST"
score: float
start: int
end: int
class ClinicalNERExtractor:
"""
Named-entity recognition over clinical notes using a
fine-tuned BioBERT model. Extracts problems, treatments,
and tests as structured ClinicalEntity objects.
"""
MODEL_ID = "blaze999/Medical-NER"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"ner",
model=self.MODEL_ID,
aggregation_strategy="simple",
device=0 if device == "cuda" else -1,
)
def extract(self, note: str) -> list[ClinicalEntity]:
raw = self._pipe(note)
return [
ClinicalEntity(
text=e["word"],
label=e["entity_group"],
score=round(e["score"], 4),
start=e["start"],
end=e["end"],
)
for e in raw
if e["score"] >= 0.80 # confidence threshold
]
def problems(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "PROBLEM"]
def treatments(self, note: str) -> list[ClinicalEntity]:
return [e for e in self.extract(note) if e.label == "TREATMENT"]
class ClinicalSummarizer:
"""
Abstractive summariser for clinical discharge notes.
Uses a BART model fine-tuned on clinical text (MedBART).
"""
MODEL_ID = "Falconsai/medical_summarization"
MAX_INPUT = 1024
MAX_SUMMARY = 256
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"summarization",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def summarize(self, note: str) -> str:
chunks = self._chunk(note)
summaries = [
self._pipe(
chunk,
max_length=self.MAX_SUMMARY,
min_length=40,
do_sample=False,
)[0]["summary_text"]
for chunk in chunks
]
return " ".join(summaries)
def _chunk(self, text: str) -> Iterator[str]:
words = text.split()
for i in range(0, len(words), self.MAX_INPUT):
yield " ".join(words[i : i + self.MAX_INPUT])
class ClinicalQA:
"""
Extractive question-answering over patient notes.
Allows clinicians to query free-text records naturally.
e.g. "What medications is the patient currently taking?"
"""
MODEL_ID = "deepset/roberta-base-squad2"
def __init__(self, device: str = "cpu") -> None:
self._pipe: Pipeline = pipeline(
"question-answering",
model=self.MODEL_ID,
device=0 if device == "cuda" else -1,
)
def ask(self, question: str, context: str) -> dict:
result = self._pipe(question=question, context=context)
return {
"answer": result["answer"],
"score": round(result["score"], 4),
"start": result["start"],
"end": result["end"],
}
# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/domain/patient_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum, auto
from typing import Optional
from uuid import UUID
class Severity(Enum):
LOW = auto()
MODERATE = auto()
HIGH = auto()
CRITICAL = auto()
def __lt__(self, other: "Severity") -> bool:
return self.value < other.value
class Flag(Enum):
ICU_CANDIDATE = auto()
ELDERLY = auto()
CHRONIC = auto()
HIGH_RISK = auto()
AUTO_FLAGGED = auto()
@dataclass(frozen=True)
class Diagnosis:
code: str # ICD-10
description: str
severity: Severity
onset_date: date
def is_critical(self) -> bool:
return self.severity == Severity.CRITICAL
def is_chronic(self) -> bool:
return (date.today() - self.onset_date).days > 365
@dataclass(frozen=True)
class Vitals:
recorded_at: datetime
heart_rate: int
systolic_bp: int
diastolic_bp: int
spo2: float
temperature: float
_NORMAL: dict = field(default_factory=lambda: {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}, compare=False, repr=False)
@property
def abnormality_index(self) -> float:
ranges = {
"heart_rate": (60, 100),
"systolic_bp": (90, 120),
"diastolic_bp": (60, 80),
"spo2": (95.0, 100.0),
"temperature": (36.1, 37.2),
}
abnormal = sum(
1 for field, (lo, hi) in ranges.items()
if not lo <= getattr(self, field) <= hi
)
return abnormal / len(ranges)
@property
def is_critical(self) -> bool:
return self.abnormality_index >= 0.6 or self.spo2 < 90.0
@property
def is_stale(self) -> bool:
age = datetime.now(timezone.utc) - self.recorded_at
return age.total_seconds() > 6 * 3600
@dataclass(frozen=True)
class PatientContext:
id: UUID
age: int
diagnoses: tuple[Diagnosis, ...]
vitals: Optional[Vitals]
risk_score: float
flags: frozenset[Flag] = frozenset()
def __post_init__(self) -> None:
if self.age <= 0:
raise ValueError("age must be positive")
if not 0.0 <= self.risk_score <= 1.0:
raise ValueError(f"risk_score {self.risk_score} out of range")
@property
def is_high_risk(self) -> bool: return self.risk_score > 0.75
@property
def is_flagged(self) -> bool: return bool(self.flags)
@property
def primary_dx(self) -> Optional[Diagnosis]:
return self.diagnoses[0] if self.diagnoses else None
@property
def comorbidities(self) -> tuple[Diagnosis, ...]:
return self.diagnoses[1:]
@property
def has_critical_vitals(self) -> bool:
return bool(self.vitals and self.vitals.is_critical)
def with_flags(self, *new_flags: Flag) -> "PatientContext":
return PatientContext(
**{**self.__dict__, "flags": self.flags | frozenset(new_flags)}
)
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
# healthcare/application/risk_scorer.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol, Sequence
from .patient_context import PatientContext
class RiskScorer(Protocol):
"""Strategy contract — any callable class satisfies this."""
def score(self, *, age: int, records: Sequence) -> float: ...
@dataclass
class WeightedRiskScorer:
"""
Weighted linear scorer.
Open/Closed: override component methods in subclasses.
"""
W_AGE: float = 0.20
W_COMORBIDITY: float = 0.40
W_VITALS: float = 0.30
W_HISTORY: float = 0.10
def score(self, *, age: int, records: Sequence) -> float:
raw = (
self._age_score(age) * self.W_AGE +
self._comorbidity_score(records) * self.W_COMORBIDITY +
self._vitals_score(records) * self.W_VITALS +
self._history_score(records) * self.W_HISTORY
)
return max(0.0, min(1.0, raw))
def _age_score(self, age: int) -> float:
if age < 18: return 0.1
if age < 45: return 0.2
if age < 65: return 0.5
if age < 80: return 0.7
return 1.0
def _comorbidity_score(self, records: Sequence) -> float:
total = sum(len(getattr(r, "diagnoses", [])) for r in records)
return min(1.0, total / 10)
def _vitals_score(self, records: Sequence) -> float:
indices = [
r.vitals.abnormality_index
for r in records if r.vitals
]
return sum(indices) / len(indices) if indices else 0.0
def _history_score(self, records: Sequence) -> float:
return 0.8 if any(getattr(r, "hospitalization", False) for r in records) else 0.1
class PediatricRiskScorer(WeightedRiskScorer):
"""Pediatric variant — overrides age weighting only (LSP)."""
W_AGE: float = 0.10
W_VITALS: float = 0.45
def _age_score(self, age: int) -> float:
return 0.6 if age < 5 else 0.3
# healthcare/application/transformers/risk_pipeline.py
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch
class ClinicalRiskTransformer:
"""
HuggingFace transformer adapter — wraps a fine-tuned
clinical BERT model as a RiskScorer strategy.
Dependency Inversion: called via the RiskScorer protocol.
"""
MODEL_ID = "HealthcareBERT/clinical-risk-classifier-v2"
def __init__(self, device: str = "cpu") -> None:
self._tokenizer = AutoTokenizer.from_pretrained(self.MODEL_ID)
self._model = AutoModelForSequenceClassification.from_pretrained(
self.MODEL_ID
).to(device)
self._pipe = pipeline(
"text-classification",
model=self._model,
tokenizer=self._tokenizer,
device=0 if device == "cuda" else -1,
)
def score(self, *, age: int, records: Sequence) -> float:
text = self._build_prompt(age, records)
result = self._pipe(text, top_k=1)[0]
# Model outputs label "HIGH_RISK" with a confidence score
return result["score"] if result["label"] == "HIGH_RISK" else 1 - result["score"]
def _build_prompt(self, age: int, records: Sequence) -> str:
dx_codes = ", ".join(
dx.code
for r in records
for dx in getattr(r, "diagnoses", [])
)
return f"Patient age {age}. Diagnoses: {dx_codes or 'none'}."
