feat(core): движок шкал и скоринг PHQ-9, GAD-7, PCL-5, AUDIT-C, PC-PTSD-5, C-SSRS, PHQ-4 #28

Merged
Eleboba merged 1 commits from feat/3-scales into dev 2026-10-03 14:57:45 +03:00
10 changed files with 1017 additions and 0 deletions
+388
View File
@@ -0,0 +1,388 @@
"""Psychometric scales: YAML schema, loading and deterministic scoring (issue #3).
Scale content (item wording, answer options, severity bands, flags) lives in `config/scales/*.yaml`
(ADR-008) and is approved by the team psychiatrist. This module only defines the schema and the
scoring algorithm, so thresholds can be changed without touching code.
Scoring is fully deterministic (ADR-007): the LLM never computes scores or decides severity.
"""
from __future__ import annotations
import operator
from collections.abc import Callable
from functools import lru_cache
from pathlib import Path
from typing import Literal
import yaml
from pydantic import BaseModel, ConfigDict, Field, model_validator
from app.config import get_settings
Sex = Literal["m", "f"]
Op = Literal[">=", ">", "<=", "<", "=="]
_OPS: dict[str, Callable[[int, int], bool]] = {
">=": operator.ge,
">": operator.gt,
"<=": operator.le,
"<": operator.lt,
"==": operator.eq,
}
class _Strict(BaseModel):
# Typos in YAML keys must fail loudly instead of silently dropping a threshold.
model_config = ConfigDict(extra="forbid", frozen=True)
class Option(_Strict):
value: int
label: str
class Item(_Strict):
id: str
text: str
# Empty list = inherit scale-level `options` (filled in by ScaleDef validation).
options: list[Option] = Field(default_factory=list)
# True = item must be asked with buttons only, never free text through the LLM (e.g. PHQ-9 item 9).
buttons_only: bool = False
# Reverse-keyed item: scored as (max_option + min_option - value). None of the current scales use it.
reverse: bool = False
@property
def min_value(self) -> int:
return min(o.value for o in self.options)
@property
def max_value(self) -> int:
return max(o.value for o in self.options)
class Range(_Strict):
min: int
max: int
class SeverityLevel(_Strict):
code: str
label: str
class Band(SeverityLevel):
"""Severity band over the total score, inclusive on both ends."""
min: int
max: int
class Condition(_Strict):
"""Predicate over answers. Exactly one target must be set: item, sum_of, total, severity_in or any_of.
Examples (YAML):
{item: phq9_9, op: ">=", value: 1} # single item
{sum_of: [phq4_1, phq4_2], op: ">=", value: 3} # subscale (GAD-2 inside PHQ-4)
{total: true, op: ">=", value: 33} # total score
{severity_in: [positive]} # computed severity code
{any_of: [{item: q4, op: ">=", value: 1}, ...]}
"""
item: str | None = None
sum_of: list[str] | None = None
total: bool = False
severity_in: list[str] | None = None
any_of: list[Condition] | None = None
op: Op = ">="
value: int | None = None
@model_validator(mode="after")
def _one_target(self) -> Condition:
targets = [
self.item is not None,
self.sum_of is not None,
self.total,
self.severity_in is not None,
self.any_of is not None,
]
if sum(targets) != 1:
raise ValueError("condition must have exactly one of: item, sum_of, total, severity_in, any_of")
needs_value = self.item is not None or self.sum_of is not None or self.total
if needs_value and self.value is None:
raise ValueError("item/sum_of/total conditions require `value`")
return self
def item_refs(self) -> set[str]:
refs = set(self.sum_of or [])
if self.item is not None:
refs.add(self.item)
for sub in self.any_of or []:
refs |= sub.item_refs()
return refs
class Flag(_Strict):
code: str
when: Condition
class SeverityRule(_Strict):
"""Rule for `scoring: rules` scales (C-SSRS): the first matching rule gives the severity."""
code: str
label: str
when: Condition
class ScaleDef(_Strict):
id: str
title: str
recall_period: str
instruction: str | None = None
license_note: str | None = None
translation_note: str | None = None
notes: str | None = None
# `sum`: severity from total-score bands. `rules`: severity from ordered item rules (C-SSRS triage).
scoring: Literal["sum", "rules"] = "sum"
# Default answer options for items that do not define their own.
options: list[Option] = Field(default_factory=list)
items: list[Item]
range: Range
severity: list[Band] = Field(default_factory=list)
# Sex-specific bands (AUDIT-C). When sex is unknown, `severity` is used.
by_sex: dict[Sex, list[Band]] | None = None
severity_rules: list[SeverityRule] = Field(default_factory=list)
default_severity: SeverityLevel | None = None
flags: list[Flag] = Field(default_factory=list)
@model_validator(mode="before")
@classmethod
def _inherit_options(cls, data: object) -> object:
if isinstance(data, dict) and data.get("options"):
for item in data.get("items", []):
if isinstance(item, dict) and not item.get("options"):
item["options"] = data["options"]
return data
@model_validator(mode="after")
def _validate(self) -> ScaleDef:
ids = [i.id for i in self.items]
if len(ids) != len(set(ids)):
raise ValueError(f"{self.id}: duplicate item ids")
for item in self.items:
if not item.options:
raise ValueError(f"{self.id}.{item.id}: no answer options")
values = [o.value for o in item.options]
if len(values) != len(set(values)):
raise ValueError(f"{self.id}.{item.id}: duplicate option values")
# Declared range must match what the items can actually produce.
lo = sum(i.min_value for i in self.items)
hi = sum(i.max_value for i in self.items)
if (self.range.min, self.range.max) != (lo, hi):
raise ValueError(f"{self.id}: range {self.range.min}..{self.range.max} != items {lo}..{hi}")
if self.scoring == "sum":
if not self.severity:
raise ValueError(f"{self.id}: sum scoring requires `severity` bands")
self._check_bands(self.severity, "severity")
for sex, bands in (self.by_sex or {}).items():
self._check_bands(bands, f"by_sex.{sex}")
else:
if not self.severity_rules or self.default_severity is None:
raise ValueError(f"{self.id}: rules scoring requires `severity_rules` and `default_severity`")
known = set(ids)
conditions = [f.when for f in self.flags] + [r.when for r in self.severity_rules]
for cond in conditions:
unknown = cond.item_refs() - known
if unknown:
raise ValueError(f"{self.id}: condition references unknown items {sorted(unknown)}")
return self
def _check_bands(self, bands: list[Band], where: str) -> None:
# Bands must tile the whole range without gaps or overlaps, otherwise some totals get no severity.
expected = self.range.min
for band in sorted(bands, key=lambda b: b.min):
if band.min != expected or band.max < band.min:
raise ValueError(f"{self.id}.{where}: bands must be contiguous from {self.range.min}")
expected = band.max + 1
if expected != self.range.max + 1:
raise ValueError(f"{self.id}.{where}: bands must end at {self.range.max}")
@property
def item_ids(self) -> list[str]:
return [i.id for i in self.items]
def item(self, item_id: str) -> Item:
for item in self.items:
if item.id == item_id:
return item
raise KeyError(f"{self.id}: unknown item {item_id!r}")
class ScaleResult(BaseModel):
model_config = ConfigDict(frozen=True)
scale: str
total: int
max_total: int
severity_code: str
severity_label: str
complete: bool
answered: int
flags: list[str] = Field(default_factory=list)
# ---------------------------------------------------------------- loading
def _scales_dir(config_dir: Path | None) -> Path:
return (config_dir or get_settings().config_dir) / "scales"
@lru_cache
def _load_dir(scales_dir: Path) -> dict[str, ScaleDef]:
scales: dict[str, ScaleDef] = {}
for path in sorted(scales_dir.glob("*.yaml")):
with path.open(encoding="utf-8") as fh:
scale = ScaleDef.model_validate(yaml.safe_load(fh))
if scale.id != path.stem:
raise ValueError(f"{path.name}: scale id {scale.id!r} must match file name")
scales[scale.id] = scale
return scales
def load_all(config_dir: Path | None = None) -> dict[str, ScaleDef]:
"""All scales from `<config_dir>/scales/*.yaml`, keyed by id. Cached; see `clear_cache()`."""
return dict(_load_dir(_scales_dir(config_dir)))
def load_scale(scale_id: str, config_dir: Path | None = None) -> ScaleDef:
scales = _load_dir(_scales_dir(config_dir))
if scale_id not in scales:
raise KeyError(f"unknown scale {scale_id!r}")
return scales[scale_id]
def clear_cache() -> None:
"""Drop cached scale definitions (after editing YAML at runtime or in tests)."""
_load_dir.cache_clear()
# ---------------------------------------------------------------- scoring
def _validate_answers(scale: ScaleDef, answers: dict[str, int]) -> dict[str, int]:
"""Check every answer is a known item with an allowed option value; return item -> scored value."""
scored: dict[str, int] = {}
for item_id, value in answers.items():
try:
item = scale.item(item_id)
except KeyError as exc:
raise ValueError(str(exc)) from exc
# bool is a subclass of int; reject it so `True` is not silently accepted as 1.
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError(f"{scale.id}.{item_id}: answer must be int, got {type(value).__name__}")
if value not in {o.value for o in item.options}:
raise ValueError(f"{scale.id}.{item_id}: invalid value {value}")
scored[item_id] = item.max_value + item.min_value - value if item.reverse else value
return scored
def _eval(cond: Condition, scored: dict[str, int], total: int, severity_code: str | None) -> bool:
"""Evaluate a condition on answered items only.
Unanswered items never satisfy an item condition. Sums over partially answered item sets use
the answered part, which is a lower bound of the true sum: a positive `>=` result is still valid,
a negative one is not conclusive (the caller sees `complete=False`).
"""
if cond.any_of is not None:
return any(_eval(sub, scored, total, severity_code) for sub in cond.any_of)
if cond.severity_in is not None:
return severity_code in cond.severity_in
assert cond.value is not None # guaranteed by Condition validation
if cond.item is not None:
if cond.item not in scored:
return False
lhs = scored[cond.item]
elif cond.sum_of is not None:
if not any(i in scored for i in cond.sum_of):
return False
lhs = sum(scored.get(i, 0) for i in cond.sum_of)
else:
lhs = total
return _OPS[cond.op](lhs, cond.value)
def _band_for(bands: list[Band], total: int) -> Band:
for band in bands:
if band.min <= total <= band.max:
return band
raise ValueError(f"total {total} outside severity bands") # unreachable: bands tile the range
def score(scale_id: str, answers: dict[str, int], sex: Sex | None = None) -> ScaleResult:
"""Score a questionnaire deterministically.
Algorithm:
1. Validate: every answer must belong to a known item and be one of its option values (ValueError).
Reverse-keyed items are flipped.
2. Total = plain sum of item values (standard scoring for PHQ-9, GAD-7, PCL-5, AUDIT-C, PC-PTSD-5,
PHQ-4; C-SSRS yes=1/no=0, total is just the count of "yes").
3. Severity:
- `scoring: sum` — the band that contains the total. AUDIT-C uses `by_sex[sex]` bands
(positive >=4 men, >=3 women); with unknown sex the default bands use the lower, more
sensitive female cut-off so a risky pattern is not missed.
- `scoring: rules` (C-SSRS screen) — ordered rules, first match wins (high > moderate > low),
otherwise `default_severity` (none). Triage: q1-2 ideation -> low, q3 method -> moderate,
q4 intent / q5 plan / q6 behaviour in past 3 months -> high.
4. Flags — every flag whose condition holds (e.g. PHQ-9 item 9 >= 1 -> `phq9_item9_positive`,
which downstream triggers C-SSRS and the crisis branch).
Incomplete answers: we do NOT impute missing items (no prorating / mean substitution). Prorating
can invent a score the patient never reported, and for safety items (PHQ-9 item 9, C-SSRS) a
missing answer must not look like "no risk". So the total is a lower bound over answered items,
`complete=False` tells consumers the severity may be underestimated, and flags/rules still fire
on what was answered (a positive item 9 is acted upon even in a half-finished PHQ-9).
"""
scale = load_scale(scale_id)
scored = _validate_answers(scale, answers)
total = sum(scored.values())
if scale.scoring == "rules":
assert scale.default_severity is not None # guaranteed by ScaleDef validation
matched = next((r for r in scale.severity_rules if _eval(r.when, scored, total, None)), None)
level = matched or scale.default_severity
code, label = level.code, level.label
else:
# sex=None or a scale without sex-specific bands -> default bands.
bands = (scale.by_sex or {}).get(sex, scale.severity) if sex else scale.severity
band = _band_for(bands, total)
code, label = band.code, band.label
flags = [f.code for f in scale.flags if _eval(f.when, scored, total, code)]
return ScaleResult(
scale=scale.id,
total=total,
max_total=scale.range.max,
severity_code=code,
severity_label=label,
complete=len(scored) == len(scale.items),
answered=len(scored),
flags=flags,
)
def delta(prev: ScaleResult, cur: ScaleResult) -> int:
"""Change of total score between two measurements of the same scale (positive = worse).
Used by rules like "PHQ-9 grew by >= 5" or "PCL-5 grew by >= 10". Both results must be complete:
a lower-bound total from a partial questionnaire would produce a fake improvement or deterioration.
"""
if prev.scale != cur.scale:
raise ValueError(f"cannot compare different scales: {prev.scale} vs {cur.scale}")
if not (prev.complete and cur.complete):
raise ValueError("delta requires complete results")
return cur.total - prev.total
View File
+50
View File
@@ -0,0 +1,50 @@
# Стартовые пороги из литературы, финальные утверждает психиатр (issue #8).
# AUDIT-C (Bush et al., 1998; Bradley et al., 2007). Положительный скрининг: ≥ 4 у мужчин, ≥ 3 у женщин.
# Если пол неизвестен — используется более чувствительный порог (≥ 3), чтобы не пропустить риск.
id: auditc
title: "Тест на употребление алкоголя (AUDIT-C)"
recall_period: "последний год"
instruction: "Ответьте, пожалуйста, на вопросы об употреблении алкоголя. Одна стандартная порция — 10 г чистого спирта."
license_note: "AUDIT/AUDIT-C (ВОЗ) — свободное использование."
translation_note: >-
Рабочий перевод команды на основе русской версии AUDIT (ВОЗ). Финальный перевод выбирает психиатр.
items:
- id: auditc_1
text: "Как часто Вы употребляете алкогольные напитки?"
options:
- {value: 0, label: "Никогда"}
- {value: 1, label: "Раз в месяц или реже"}
- {value: 2, label: "2–4 раза в месяц"}
- {value: 3, label: "2–3 раза в неделю"}
- {value: 4, label: "4 раза в неделю или чаще"}
- id: auditc_2
text: "Сколько стандартных порций алкоголя Вы обычно выпиваете в день, когда пьёте?"
options:
- {value: 0, label: "1–2"}
- {value: 1, label: "3–4"}
- {value: 2, label: "5–6"}
- {value: 3, label: "7–9"}
- {value: 4, label: "10 и более"}
- id: auditc_3
text: "Как часто Вы выпиваете 6 и более стандартных порций за один раз?"
options:
- {value: 0, label: "Никогда"}
- {value: 1, label: "Реже раза в месяц"}
- {value: 2, label: "Ежемесячно"}
- {value: 3, label: "Еженедельно"}
- {value: 4, label: "Ежедневно или почти ежедневно"}
range: {min: 0, max: 12}
# Пол неизвестен — чувствительный порог (≥ 3).
severity:
- {min: 0, max: 2, code: negative, label: "Скрининг отрицательный"}
- {min: 3, max: 12, code: positive, label: "Скрининг положительный"}
by_sex:
m:
- {min: 0, max: 3, code: negative, label: "Скрининг отрицательный"}
- {min: 4, max: 12, code: positive, label: "Скрининг положительный"}
f:
- {min: 0, max: 2, code: negative, label: "Скрининг отрицательный"}
- {min: 3, max: 12, code: positive, label: "Скрининг положительный"}
flags:
- code: auditc_positive
when: {severity_in: [positive]}
+58
View File
@@ -0,0 +1,58 @@
# Стартовые пороги из литературы, финальные утверждает психиатр (issue #8).
# C-SSRS, скрининговая версия (Posner et al., 2011; Columbia Lighthouse Project), триаж:
# q1–q2 (желание умереть, суицидальные мысли) → low
# q3 (мысли о способе, без намерения) → moderate
# q4 (намерение), q5 (план), q6 (поведение за 3 месяца) → high
# Ответы «да/нет» хранятся как 1/0. Порядок правил важен: первое совпавшее — уровень риска.
# В оригинале q6 задаётся «за всю жизнь» с уточнением «в последние 3 месяца»; для триажа нужен
# только второй вариант, поэтому q6 сформулирован сразу про 3 месяца (утверждает психиатр).
id: cssrs_screen
title: "Колумбийская шкала оценки тяжести суицидального риска (C-SSRS), скрининг"
recall_period: "последний месяц (вопрос 6 — последние 3 месяца)"
instruction: "Ответьте, пожалуйста, на несколько вопросов о том, как Вы себя чувствовали за последний месяц."
license_note: "C-SSRS — бесплатно для клинического и некоммерческого использования (Columbia Lighthouse Project)."
translation_note: >-
Рабочий перевод команды (демо-текст). Финальный текст — официальная русская скрининговая версия,
выбирает психиатр.
scoring: rules
options:
- {value: 0, label: "Нет"}
- {value: 1, label: "Да"}
items:
- {id: cssrs_1, text: "Хотелось ли Вам умереть или заснуть и не проснуться?", buttons_only: true}
- {id: cssrs_2, text: "Были ли у Вас мысли о том, чтобы покончить с собой?", buttons_only: true}
- {id: cssrs_3, text: "Думали ли Вы о том, как могли бы это сделать?", buttons_only: true}
- id: cssrs_4
text: "Были ли у Вас такие мысли и хоть какое-то намерение действовать в соответствии с ними?"
buttons_only: true
- id: cssrs_5
text: "Начинали ли Вы продумывать или уже продумали детали того, как покончить с собой? Собираетесь ли Вы осуществить этот план?"
buttons_only: true
- id: cssrs_6
text: "За последние 3 месяца делали ли Вы что-нибудь, начинали делать или готовились сделать, чтобы покончить с собой?"
buttons_only: true
range: {min: 0, max: 6}
severity_rules:
- code: high
label: "Высокий риск"
when:
any_of:
- {item: cssrs_4, op: ">=", value: 1}
- {item: cssrs_5, op: ">=", value: 1}
- {item: cssrs_6, op: ">=", value: 1}
- code: moderate
label: "Умеренный риск"
when: {item: cssrs_3, op: ">=", value: 1}
- code: low
label: "Низкий риск"
when:
any_of:
- {item: cssrs_1, op: ">=", value: 1}
- {item: cssrs_2, op: ">=", value: 1}
default_severity: {code: none, label: "Риск не выявлен"}
flags:
# high → acute-алерт и кризисный протокол (ADR-006); moderate → врачу ≤ 24 ч (config/rules.yaml).
- code: cssrs_high
when: {severity_in: [high]}
- code: cssrs_moderate
when: {severity_in: [moderate]}
+28
View File
@@ -0,0 +1,28 @@
# Стартовые пороги из литературы, финальные утверждает психиатр (issue #8).
# GAD-7 (Spitzer et al., 2006). Градации: 0–4, 5–9, 10–14, 15–21; ≥ 10 — вероятное ГТР.
id: gad7
title: "Шкала генерализованного тревожного расстройства (GAD-7)"
recall_period: "последние 2 недели"
instruction: "Как часто за последние 2 недели Вас беспокоили следующие проблемы?"
license_note: "GAD-7 (Pfizer) — свободное использование, разрешения не требуется."
translation_note: "Стандартная русская версия GAD-7 (MAPI). Финальную формулировку утверждает психиатр."
options:
- {value: 0, label: "Ни разу"}
- {value: 1, label: "Несколько дней"}
- {value: 2, label: "Более половины дней"}
- {value: 3, label: "Почти каждый день"}
items:
- {id: gad7_1, text: "Вы нервничали, тревожились или были на грани срыва"}
- {id: gad7_2, text: "Вы были не в состоянии остановить или контролировать своё беспокойство"}
- {id: gad7_3, text: "Вы слишком сильно беспокоились по различным поводам"}
- {id: gad7_4, text: "Вам было трудно расслабиться"}
- {id: gad7_5, text: "Вы были настолько беспокойны, что Вам было тяжело усидеть на месте"}
- {id: gad7_6, text: "Вы легко раздражались или злились"}
- {id: gad7_7, text: "Вы испытывали страх, как будто может случиться что-то ужасное"}
range: {min: 0, max: 21}
severity:
- {min: 0, max: 4, code: minimal, label: "Минимальная"}
- {min: 5, max: 9, code: mild, label: "Лёгкая"}
- {min: 10, max: 14, code: moderate, label: "Умеренная"}
- {min: 15, max: 21, code: severe, label: "Тяжёлая"}
flags: []
+60
View File
@@ -0,0 +1,60 @@
# Стартовые пороги из литературы, финальные утверждает психиатр (issue #8).
# PCL-5 (Weathers et al., 2013). Порог вероятного ПТСР в литературе 31–33; берём 33 (VA National Center for PTSD).
# Градация ≥ 50 «высокая» — рабочее решение команды под правило SPEC «PCL-5 ≥ 50 → high», не часть оригинальной шкалы.
# Клинически значимое изменение — 10 баллов (правило «рост PCL-5 на ≥ 10 → врач» считает delta()).
id: pcl5
title: "Контрольный список симптомов ПТСР (PCL-5)"
recall_period: "последний месяц"
instruction: >-
Ниже перечислены проблемы, которые иногда возникают у людей в ответ на очень стрессовое событие.
Отметьте, насколько каждая из них беспокоила Вас за последний месяц.
license_note: "PCL-5 — общественное достояние (VA National Center for PTSD); русская версия — лицензия на проверке."
translation_note: >-
Рабочий перевод команды (демо-текст). Финальный русский перевод выбирает психиатр (SPEC §9).
Для еженедельного мониторинга используется та же шкала с периодом «последняя неделя».
options:
- {value: 0, label: "Совсем нет"}
- {value: 1, label: "Немного"}
- {value: 2, label: "Умеренно"}
- {value: 3, label: "Сильно"}
- {value: 4, label: "Крайне сильно"}
items:
- {id: pcl5_1, text: "Повторяющиеся, тревожащие и нежелательные воспоминания о стрессовом событии"}
- {id: pcl5_2, text: "Повторяющиеся тревожащие сны о стрессовом событии"}
- id: pcl5_3
text: "Внезапное ощущение или поведение, как будто стрессовое событие происходит снова (как будто Вы снова в нём)"
- {id: pcl5_4, text: "Сильное расстройство, когда что-то напоминало Вам о стрессовом событии"}
- id: pcl5_5
text: >-
Сильные телесные реакции, когда что-то напоминало Вам о стрессовом событии
(например, сердцебиение, затруднённое дыхание, потливость)
- {id: pcl5_6, text: "Избегание воспоминаний, мыслей или чувств, связанных со стрессовым событием"}
- id: pcl5_7
text: >-
Избегание внешних напоминаний о стрессовом событии (людей, мест, разговоров, занятий,
предметов или ситуаций)
- {id: pcl5_8, text: "Трудности с воспоминанием важных деталей стрессового события"}
- id: pcl5_9
text: >-
Сильные негативные убеждения о себе, других людях или мире (например: «я плохой»,
«со мной что-то серьёзно не так», «никому нельзя доверять», «мир совершенно опасен»)
- {id: pcl5_10, text: "Обвинение себя или кого-то другого в стрессовом событии или в том, что произошло после него"}
- {id: pcl5_11, text: "Сильные негативные чувства: страх, ужас, гнев, вина или стыд"}
- {id: pcl5_12, text: "Потеря интереса к занятиям, которые раньше приносили Вам удовольствие"}
- {id: pcl5_13, text: "Ощущение отстранённости или отчуждённости от других людей"}
- id: pcl5_14
text: "Трудности с переживанием положительных чувств (например, неспособность испытывать счастье или любовь к близким)"
- {id: pcl5_15, text: "Раздражительность, вспышки гнева или агрессивное поведение"}
- {id: pcl5_16, text: "Чрезмерный риск или действия, которые могут причинить Вам вред"}
- {id: pcl5_17, text: "Состояние «сверхнастороженности»: постоянная бдительность, готовность к опасности"}
- {id: pcl5_18, text: "Пугливость, Вы легко вздрагиваете"}
- {id: pcl5_19, text: "Трудности с концентрацией внимания"}
- {id: pcl5_20, text: "Трудности с засыпанием или прерывистый сон"}
range: {min: 0, max: 80}
severity:
- {min: 0, max: 32, code: below_threshold, label: "Ниже порога ПТСР"}
- {min: 33, max: 49, code: probable, label: "Вероятное ПТСР"}
- {min: 50, max: 80, code: high, label: "Выраженная симптоматика ПТСР"}
flags:
- code: pcl5_probable_ptsd
when: {total: true, op: ">=", value: 33}
+30
View File
@@ -0,0 +1,30 @@
# Стартовые пороги из литературы, финальные утверждает психиатр (issue #8).
# PC-PTSD-5 (Prins et al., 2016). Положительный скрининг: ≥ 3 ответов «да».
id: pcptsd5
title: "Скрининг ПТСР для первичного звена (PC-PTSD-5)"
recall_period: "последний месяц"
instruction: >-
Иногда с людьми случаются необычно или особенно пугающие, ужасные или травмирующие события.
Если с Вами случалось такое событие, ответьте: было ли с Вами за последний месяц следующее?
notes: >-
Задаётся только если на вопрос о травматическом опыте пациент ответил «да».
Если травматического опыта не было, шкала не проводится (результат = 0).
license_note: "PC-PTSD-5 — общественное достояние (VA National Center for PTSD)."
translation_note: "Рабочий перевод команды (демо-текст). Финальный перевод выбирает психиатр."
options:
- {value: 0, label: "Нет"}
- {value: 1, label: "Да"}
items:
- {id: pcptsd5_1, text: "Вам снились кошмары о событии или Вы думали о нём, когда не хотели?"}
- {id: pcptsd5_2, text: "Вы старались не думать о событии или избегали ситуаций, которые о нём напоминали?"}
- {id: pcptsd5_3, text: "Вы были постоянно настороже, бдительны или легко вздрагивали?"}
- {id: pcptsd5_4, text: "Вы чувствовали оцепенение или отстранённость от людей, занятий или окружающего?"}
- id: pcptsd5_5
text: "Вы чувствовали вину или не могли перестать обвинять себя или других в событии или в проблемах, которые оно вызвало?"
range: {min: 0, max: 5}
severity:
- {min: 0, max: 2, code: negative, label: "Скрининг отрицательный"}
- {min: 3, max: 5, code: positive, label: "Скрининг положительный"}
flags:
- code: pcptsd5_positive
when: {severity_in: [positive]}
+31
View File
@@ -0,0 +1,31 @@
# Стартовые пороги из литературы, финальные утверждает психиатр (issue #8).
# PHQ-4 (Kroenke et al., 2009) = GAD-2 (п.1–2) + PHQ-2 (п.3–4). Модуль family.
# Общий балл: 0–2 норма, 3–5 лёгкая, 6–8 умеренная, 9–12 тяжёлая.
# Подшкалы: GAD-2 ≥ 3 и PHQ-2 ≥ 3 — положительный скрининг → полный GAD-7 / PHQ-9.
id: phq4
title: "Опросник здоровья пациента (PHQ-4)"
recall_period: "последние 2 недели"
instruction: "Как часто за последние 2 недели Вас беспокоили следующие проблемы?"
license_note: "PHQ-4 (Pfizer) — свободное использование, разрешения не требуется."
translation_note: "Пункты из стандартных русских версий GAD-7 и PHQ-9. Финальную формулировку утверждает психиатр."
options:
- {value: 0, label: "Ни разу"}
- {value: 1, label: "Несколько дней"}
- {value: 2, label: "Более половины дней"}
- {value: 3, label: "Почти каждый день"}
items:
- {id: phq4_1, text: "Вы нервничали, тревожились или были на грани срыва"}
- {id: phq4_2, text: "Вы были не в состоянии остановить или контролировать своё беспокойство"}
- {id: phq4_3, text: "Вам не хотелось ничего делать"}
- {id: phq4_4, text: "У Вас было плохое настроение, Вы были подавлены или испытывали чувство безысходности"}
range: {min: 0, max: 12}
severity:
- {min: 0, max: 2, code: normal, label: "Норма"}
- {min: 3, max: 5, code: mild, label: "Лёгкая"}
- {min: 6, max: 8, code: moderate, label: "Умеренная"}
- {min: 9, max: 12, code: severe, label: "Тяжёлая"}
flags:
- code: phq4_anxiety_positive
when: {sum_of: [phq4_1, phq4_2], op: ">=", value: 3}
- code: phq4_depression_positive
when: {sum_of: [phq4_3, phq4_4], op: ">=", value: 3}
+49
View File
@@ -0,0 +1,49 @@
# Стартовые пороги из литературы, финальные утверждает психиатр (issue #8).
# PHQ-9 (Kroenke, Spitzer, Williams, 2001). Градации: 0–4, 5–9, 10–14, 15–19, 20–27.
id: phq9
title: "Опросник здоровья пациента (PHQ-9)"
recall_period: "последние 2 недели"
instruction: "Как часто за последние 2 недели Вас беспокоили следующие проблемы?"
license_note: "PHQ-9 (Pfizer) — свободное использование, разрешения не требуется."
translation_note: "Стандартная русская версия PHQ-9 (MAPI). Финальную формулировку утверждает психиатр."
options:
- {value: 0, label: "Ни разу"}
- {value: 1, label: "Несколько дней"}
- {value: 2, label: "Более половины дней"}
- {value: 3, label: "Почти каждый день"}
items:
- id: phq9_1
text: "Вам не хотелось ничего делать"
- id: phq9_2
text: "У Вас было плохое настроение, Вы были подавлены или испытывали чувство безысходности"
- id: phq9_3
text: "Вам было трудно заснуть, у Вас был прерывистый сон, или Вы слишком много спали"
- id: phq9_4
text: "Вы были утомлены, или у Вас было мало сил"
- id: phq9_5
text: "У Вас был плохой аппетит, или Вы переедали"
- id: phq9_6
text: >-
Вы плохо о себе думали: считали себя неудачником (неудачницей), или были в себе разочарованы,
или считали, что подвели свою семью
- id: phq9_7
text: "Вам было трудно сосредоточиться (например, на чтении газеты или просмотре телепередач)"
- id: phq9_8
text: >-
Вы двигались или говорили настолько медленно, что окружающие это замечали? Или, наоборот,
Вы были настолько суетливы или взбудоражены, что двигались гораздо больше обычного
- id: phq9_9
text: "Вас посещали мысли о том, что Вам лучше было бы умереть, или о том, чтобы причинить себе какой-нибудь вред"
# Пункт о суицидальных мыслях — только кнопками, без интерпретации LLM (SPEC, ADR-002).
buttons_only: true
range: {min: 0, max: 27}
severity:
- {min: 0, max: 4, code: minimal, label: "Минимальная"}
- {min: 5, max: 9, code: mild, label: "Лёгкая"}
- {min: 10, max: 14, code: moderate, label: "Умеренная"}
- {min: 15, max: 19, code: moderately_severe, label: "Умеренно-тяжёлая"}
- {min: 20, max: 27, code: severe, label: "Тяжёлая"}
flags:
# П.9 ≥ 1 → C-SSRS + кризисная ветка (обрабатывают правила, config/rules.yaml).
- code: phq9_item9_positive
when: {item: phq9_9, op: ">=", value: 1}
+323
View File
@@ -0,0 +1,323 @@
from pathlib import Path
import pytest
import yaml
from pydantic import ValidationError
from app.core.scales import ScaleDef, ScaleResult, delta, load_all, load_scale, score
def answers_for(scale_id: str, total: int) -> dict[str, int]:
"""Fill items greedily (each up to its max option) so the plain sum equals `total`."""
scale = load_scale(scale_id)
out: dict[str, int] = {}
left = total
for item in scale.items:
v = min(left, item.max_value)
out[item.id] = v
left -= v
assert left == 0, f"{scale_id}: total {total} unreachable"
return out
def severity(scale_id: str, total: int, **kw: object) -> str:
return score(scale_id, answers_for(scale_id, total), **kw).severity_code
# ---------------------------------------------------------------- loading
def test_all_scales_load_with_expected_item_counts() -> None:
counts = {k: len(v.items) for k, v in load_all().items()}
expected = {"phq9": 9, "gad7": 7, "pcl5": 20, "auditc": 3, "pcptsd5": 5, "cssrs_screen": 6, "phq4": 4}
assert counts == expected
def test_unknown_scale() -> None:
with pytest.raises(KeyError):
load_scale("nope")
def test_phq9_item9_buttons_only() -> None:
phq9 = load_scale("phq9")
assert [i.id for i in phq9.items if i.buttons_only] == ["phq9_9"]
assert all(i.buttons_only for i in load_scale("cssrs_screen").items)
def test_clinical_notes_present() -> None:
for scale in load_all().values():
assert scale.translation_note, scale.id
assert scale.license_note, scale.id
# ---------------------------------------------------------------- band boundaries
@pytest.mark.parametrize(
("scale_id", "boundaries"),
[
(
"phq9",
[
(0, "minimal"),
(4, "minimal"),
(5, "mild"),
(9, "mild"),
(10, "moderate"),
(14, "moderate"),
(15, "moderately_severe"),
(19, "moderately_severe"),
(20, "severe"),
(27, "severe"),
],
),
(
"gad7",
[
(0, "minimal"),
(4, "minimal"),
(5, "mild"),
(9, "mild"),
(10, "moderate"),
(14, "moderate"),
(15, "severe"),
(21, "severe"),
],
),
(
"pcl5",
[
(0, "below_threshold"),
(32, "below_threshold"),
(33, "probable"),
(49, "probable"),
(50, "high"),
(80, "high"),
],
),
("pcptsd5", [(0, "negative"), (2, "negative"), (3, "positive"), (5, "positive")]),
("phq4", [(0, "normal"), (2, "normal"), (3, "mild"), (5, "mild"), (6, "moderate"), (9, "severe")]),
],
)
def test_band_boundaries(scale_id: str, boundaries: list[tuple[int, str]]) -> None:
for total, code in boundaries:
assert severity(scale_id, total) == code, (scale_id, total)
def test_result_fields() -> None:
r = score("phq9", answers_for("phq9", 12))
assert r == ScaleResult(
scale="phq9",
total=12,
max_total=27,
severity_code="moderate",
severity_label="Умеренная",
complete=True,
answered=9,
flags=[],
)
# ---------------------------------------------------------------- flags
def test_phq9_item9_flag() -> None:
base = {f"phq9_{i}": 0 for i in range(1, 10)}
assert score("phq9", base).flags == []
r = score("phq9", base | {"phq9_9": 1})
assert r.flags == ["phq9_item9_positive"]
# Item 9 flag fires regardless of the low total.
assert r.severity_code == "minimal"
def test_phq9_item9_flag_in_incomplete_questionnaire() -> None:
r = score("phq9", {"phq9_9": 2})
assert not r.complete
assert "phq9_item9_positive" in r.flags
def test_pcl5_probable_flag() -> None:
assert score("pcl5", answers_for("pcl5", 32)).flags == []
assert score("pcl5", answers_for("pcl5", 33)).flags == ["pcl5_probable_ptsd"]
def test_pcptsd5_flag() -> None:
assert score("pcptsd5", answers_for("pcptsd5", 2)).flags == []
assert score("pcptsd5", answers_for("pcptsd5", 3)).flags == ["pcptsd5_positive"]
def test_phq4_subscale_flags() -> None:
zero = {"phq4_1": 0, "phq4_2": 0, "phq4_3": 0, "phq4_4": 0}
assert score("phq4", zero | {"phq4_1": 1, "phq4_2": 1, "phq4_3": 1, "phq4_4": 1}).flags == []
assert score("phq4", zero | {"phq4_1": 2, "phq4_2": 1}).flags == ["phq4_anxiety_positive"]
assert score("phq4", zero | {"phq4_3": 3}).flags == ["phq4_depression_positive"]
both = score("phq4", {"phq4_1": 3, "phq4_2": 0, "phq4_3": 0, "phq4_4": 3})
assert both.flags == ["phq4_anxiety_positive", "phq4_depression_positive"]
# ---------------------------------------------------------------- AUDIT-C by sex
@pytest.mark.parametrize(
("total", "sex", "code"),
[
(3, "m", "negative"),
(4, "m", "positive"),
(2, "f", "negative"),
(3, "f", "positive"),
# Unknown sex: the more sensitive (female) cut-off.
(2, None, "negative"),
(3, None, "positive"),
(0, "m", "negative"),
(12, "f", "positive"),
],
)
def test_auditc_by_sex(total: int, sex: str | None, code: str) -> None:
r = score("auditc", answers_for("auditc", total), sex=sex) # type: ignore[arg-type]
assert r.severity_code == code
assert (r.flags == ["auditc_positive"]) == (code == "positive")
# ---------------------------------------------------------------- C-SSRS triage
def cssrs(*yes: int) -> dict[str, int]:
return {f"cssrs_{i}": int(i in yes) for i in range(1, 7)}
@pytest.mark.parametrize(
("yes", "level"),
[
((), "none"),
((1,), "low"),
((2,), "low"),
((1, 2), "low"),
((1, 2, 3), "moderate"),
((3,), "moderate"),
((1, 2, 3, 4), "high"),
((5,), "high"),
((6,), "high"),
((1, 6), "high"),
],
)
def test_cssrs_triage(yes: tuple[int, ...], level: str) -> None:
r = score("cssrs_screen", cssrs(*yes))
assert r.severity_code == level
assert r.total == len(yes)
assert r.max_total == 6
def test_cssrs_flags() -> None:
assert score("cssrs_screen", cssrs(4)).flags == ["cssrs_high"]
assert score("cssrs_screen", cssrs(3)).flags == ["cssrs_moderate"]
assert score("cssrs_screen", cssrs(1)).flags == []
def test_cssrs_partial_high_answer_still_high() -> None:
# Screen version skips q3–q5 when q2 = no, so partial answers are normal; risk must still be detected.
r = score("cssrs_screen", {"cssrs_1": 1, "cssrs_2": 0, "cssrs_6": 1})
assert r.severity_code == "high"
assert not r.complete
# ---------------------------------------------------------------- incomplete / invalid
def test_incomplete_no_imputation() -> None:
r = score("phq9", {"phq9_1": 3, "phq9_2": 3})
assert r.total == 6
assert r.answered == 2
assert r.complete is False
assert r.severity_code == "mild" # lower bound, not prorated (prorating would give 27)
def test_empty_answers() -> None:
r = score("gad7", {})
assert (r.total, r.answered, r.complete, r.severity_code) == (0, 0, False, "minimal")
@pytest.mark.parametrize(
"answers",
[
{"phq9_1": 4},
{"phq9_1": -1},
{"phq9_10": 1},
{"gad7_1": 1},
{"phq9_1": True},
{"phq9_1": "2"},
],
)
def test_invalid_answers(answers: dict[str, int]) -> None:
with pytest.raises(ValueError):
score("phq9", answers)
def test_cssrs_rejects_non_binary() -> None:
with pytest.raises(ValueError):
score("cssrs_screen", {"cssrs_1": 2})
def test_pcl5_accepts_4_phq9_does_not() -> None:
assert score("pcl5", {"pcl5_1": 4}).total == 4
with pytest.raises(ValueError):
score("phq9", {"phq9_1": 4})
# ---------------------------------------------------------------- delta
def test_delta() -> None:
prev = score("phq9", answers_for("phq9", 8))
cur = score("phq9", answers_for("phq9", 14))
assert delta(prev, cur) == 6
assert delta(cur, prev) == -6
def test_delta_rejects_mismatch_and_incomplete() -> None:
phq = score("phq9", answers_for("phq9", 8))
with pytest.raises(ValueError):
delta(phq, score("gad7", answers_for("gad7", 8)))
with pytest.raises(ValueError):
delta(phq, score("phq9", {"phq9_1": 1}))
# ---------------------------------------------------------------- schema validation
def _raw(scale_id: str) -> dict:
path = Path(__file__).resolve().parent.parent / "config" / "scales" / f"{scale_id}.yaml"
return yaml.safe_load(path.read_text(encoding="utf-8"))
def test_schema_rejects_gap_in_bands() -> None:
raw = _raw("phq9")
raw["severity"][1]["min"] = 6
with pytest.raises(ValidationError):
ScaleDef.model_validate(raw)
def test_schema_rejects_wrong_range() -> None:
raw = _raw("gad7")
raw["range"]["max"] = 27
with pytest.raises(ValidationError):
ScaleDef.model_validate(raw)
def test_schema_rejects_unknown_item_in_flag() -> None:
raw = _raw("phq9")
raw["flags"][0]["when"]["item"] = "phq9_99"
with pytest.raises(ValidationError):
ScaleDef.model_validate(raw)
def test_schema_rejects_typo_key() -> None:
raw = _raw("phq9")
raw["flagz"] = raw.pop("flags")
with pytest.raises(ValidationError):
ScaleDef.model_validate(raw)
def test_load_all_from_custom_dir(tmp_path: Path) -> None:
(tmp_path / "scales").mkdir()
(tmp_path / "scales" / "gad7.yaml").write_text(yaml.safe_dump(_raw("gad7"), allow_unicode=True), "utf-8")
assert list(load_all(tmp_path)) == ["gad7"]