De-identification

openbtk.deid is cross-modal by design and has zero heavy dependencies in its default configuration (its built-in RuleRecognizer is stdlib re only) — importing it eagerly is safe.

from openbtk.deid import DeidEngine, DeidMode

engine = DeidEngine(mode=DeidMode.SURROGATE, recognizers=["rule"])
result = engine.deidentify(text, patient_id="hashed-123")

openbtk.deid.engine.DeidEngine

DeidEngine(*, mode: DeidMode = DeidMode.SURROGATE, recall_bias: RecallBias = 'high', recognizers: Sequence[str] = _DEFAULT_RECOGNIZERS, consistency_key: bytes | None = None, consistency_store: ConsistencyStore | None = None)

See module docstring.

Source code in src/openbtk/deid/engine.py

def __init__(
    self,
    *,
    mode: DeidMode = DeidMode.SURROGATE,
    recall_bias: RecallBias = "high",
    recognizers: Sequence[str] = _DEFAULT_RECOGNIZERS,
    consistency_key: bytes | None = None,
    consistency_store: ConsistencyStore | None = None,
) -> None:
    # DeidMode(mode) is idempotent for an already-real DeidMode member
    # (enum construction from an existing member returns that member),
    # and coerces a plain string -- which is exactly what every
    # registry/config-driven construction supplies (StepConfig.params
    # is JSON-safe only, never a live enum member; docs/03_ARCHITECTURE.md
    # section 7.3's own worked YAML example passes `mode: surrogate` as
    # a bare string). Without this, Transform's `self._mode is
    # DeidMode.REDACT`-style identity checks silently never match a
    # plain string, and _compute_config_hash's `self._mode.value`
    # crashes outright -- both confirmed by direct reproduction, not
    # assumed.
    try:
        mode = DeidMode(mode)
    except ValueError as e:
        raise DeidError(
            f"Unknown mode {mode!r}.",
            context={"mode": str(mode), "valid": [m.value for m in DeidMode]},
        ) from e
    if recall_bias not in _RECALL_BIAS_THRESHOLDS:
        raise DeidError(
            f"Unknown recall_bias {recall_bias!r}.",
            context={
                "recall_bias": recall_bias,
                "valid": list(_RECALL_BIAS_THRESHOLDS),
            },
        )
    self._mode = mode
    self._recall_bias = recall_bias
    self._recognizer_names = list(recognizers)
    self._recognizers = [self._resolve_recognizer(name) for name in recognizers]
    self._merger = SpanMerger()
    self._transform = Transform(
        mode, key=consistency_key, consistency_store=consistency_store
    )
    self._config_hash = self._compute_config_hash()

deidentify

deidentify(text: str, *, patient_id: str, document_id: str | None = None) -> DeidResult

Detect, merge, filter, transform, and report -- one document.

Parameters:

Name	Type	Description	Default
text	str	The document text to de-identify.	required
patient_id	str	Stable identifier used for SURROGATE/DATE_SHIFT consistency. Should already be a hashed/pseudonymous value by the time it reaches here -- this engine has no way to enforce that.	required
document_id	str | None	Report identifier. Auto-generated (a random UUID) if not given -- never derived from the text itself, which would risk encoding content into an identifier.	None

Returns:

Type	Description
DeidResult	The de-identified text plus its DeidReport.

Source code in src/openbtk/deid/engine.py

def deidentify(
    self, text: str, *, patient_id: str, document_id: str | None = None
) -> DeidResult:
    """Detect, merge, filter, transform, and report -- one document.

    Args:
        text: The document text to de-identify.
        patient_id: Stable identifier used for SURROGATE/DATE_SHIFT
            consistency. Should already be a hashed/pseudonymous value
            by the time it reaches here -- this engine has no way to
            enforce that.
        document_id: Report identifier. Auto-generated (a random UUID)
            if not given -- never derived from the text itself, which
            would risk encoding content into an identifier.

    Returns:
        The de-identified text plus its ``DeidReport``.
    """
    doc_id = document_id if document_id is not None else uuid.uuid4().hex
    raw_detections: list[Detection] = []
    for recognizer in self._recognizers:
        raw_detections.extend(recognizer.detect(text))
    raw_detections = self._shield_rule_detections_from_ner(raw_detections)

    merged = self._merger.merge(raw_detections)
    threshold = _RECALL_BIAS_THRESHOLDS[self._recall_bias]
    surviving = [d for d in merged if d.confidence >= threshold]

    deidentified_text = self._transform.apply(
        text, surviving, patient_id=patient_id
    )

    entity_counts: dict[PHICategory, int] = {}
    for detection in surviving:
        entity_counts[detection.category] = (
            entity_counts.get(detection.category, 0) + 1
        )

    report = DeidReport(
        document_id=doc_id,
        entity_counts=entity_counts,
        detections=surviving,
        residual_risk=self._estimate_risk(surviving),
        engine_config_hash=self._config_hash,
    )
    return DeidResult(text=deidentified_text, report=report)

openbtk.deid.schemas.DeidMode

Bases: StrEnum

How a detected span is transformed. See openbtk.deid.transforms.

openbtk.deid.schemas.DeidReport

Bases: BaseModel

Per-document audit output of one DeidEngine.deidentify() call.

Safe to archive, forward to a privacy officer, or attach to a run manifest -- see this module's docstring for why no field here can ever hold document content.

Example

report = DeidReport( ... document_id="doc-1", ... entity_counts={PHICategory.SSN: 1}, ... detections=[], ... residual_risk=RiskEstimate(level="low", rationale="clean"), ... engine_config_hash="abc123", ... ) report.entity_counts[PHICategory.SSN] 1

openbtk.deid.schemas.DeidResult

Bases: BaseModel

The return value of DeidEngine.deidentify(): transformed text plus its audit report.

Example

report = DeidReport( ... document_id="doc-1", ... entity_counts={}, ... detections=[], ... residual_risk=RiskEstimate(level="low", rationale="clean"), ... engine_config_hash="abc123", ... ) DeidResult(text="hello", report=report).text 'hello'

openbtk.deid.schemas.DeidStatus

Bases: StrEnum

What state a record's PHI is in -- lives here, not in any one modality's own schemas module, because both clinical_text and ehr need it (docs/05_DATA_MODALITY_SPEC.md sections 1.1 and 2.1) and the layering rule (docs/03_ARCHITECTURE.md section 2) keeps the two modalities independent of each other -- a shared concept belongs in a module both already depend on downward, not in either one's own package.

Making PHI state part of the type (docs/05_DATA_MODALITY_SPEC.md section 1.1) is cheap and prevents a whole class of accident: a pipeline step can refuse to send a RAW record to an off-site provider by checking one field, rather than trusting that de-id happened somewhere upstream.

openbtk.deid.schemas.Detection

Bases: BaseModel

One PHI span found by a recognizer or the merger.

Example

from openbtk.core.schemas import TextSpan d = Detection( ... category=PHICategory.SSN, ... span=TextSpan(start=10, end=21, label="ssn", confidence=0.99), ... confidence=0.99, ... method="rule", ... ) d.category

Note

span.confidence and confidence are kept equal by every producer in this codebase -- TextSpan is a shared generic type (also used by guardrails and entity linking) that always carries its own confidence field, not a second, independent estimate. Keeping one authoritative value and mirroring it into the span avoids two numbers silently drifting apart.

openbtk.deid.schemas.PHICategory

Bases: StrEnum

The 18 HIPAA Safe Harbor identifier categories (45 CFR 164.514(b)(2)) -- a public, regulatory enumeration, not something specific to any patient or institution.

Two categories cannot appear in text at all and are included here only for completeness of the legal enumeration, not because a text recognizer can ever populate them:

• FULL_FACE_PHOTO: an imaging artefact. Detecting it is out of scope for openbtk.deid on text input, and out of scope for Openbtk entirely under the scope gate (CLAUDE.md rule 13) until an imaging modality ships.
• BIOMETRIC_IDENTIFIER: fingerprints, retinal scans, voiceprints -- likewise not a text-detectable category.

A recognizer that claims to detect either of these on text input should be treated as a bug, not a feature.

openbtk.deid.schemas.RiskEstimate

Bases: BaseModel

A coarse, explainable residual-risk judgement for one document.

Not specified beyond a bare type reference in docs/04_API_DESIGN.md section 5 -- designed from scratch. Deliberately a bounded category plus a human-readable reason rather than a numeric score: a privacy officer reading a DeidReport needs "why", and a single float invites false precision this milestone has no calibration data to back up.

Example

RiskEstimate(level="low", rationale="No low-confidence spans.").level 'low'

Structured data: k-anonymity

openbtk.deid.kanonymity.k_anonymity_report

k_anonymity_report(rows: Iterable[Mapping[str, object]], quasi_identifiers: Sequence[str], *, k: int) -> KAnonymityReport

Measure k-anonymity of rows over quasi_identifiers.

Streams the rows: memory is one counter entry per distinct combination, not one per record. A missing value counts as its own value ("unknown"), so records with the same gaps are in the same class.

Parameters:

Name	Type	Description	Default
rows	Iterable[Mapping[str, object]]	The table.	required
quasi_identifiers	Sequence[str]	The columns an outsider could link on.	required
k	int	The threshold to report against.	required

Example

rows = [{"sex": "f", "yob": 1980}] * 3 + [{"sex": "m", "yob": 1975}] report = k_anonymity_report(rows, ["sex", "yob"], k=3) report.k_achieved, report.n_below_k, report.satisfied (1, 1, False)

Source code in src/openbtk/deid/kanonymity.py

def k_anonymity_report(
    rows: Iterable[Mapping[str, object]], quasi_identifiers: Sequence[str], *, k: int
) -> KAnonymityReport:
    """Measure k-anonymity of ``rows`` over ``quasi_identifiers``.

    Streams the rows: memory is one counter entry per distinct combination, not one per
    record. A missing value counts as its own value ("unknown"), so records with the
    same gaps are in the same class.

    Args:
        rows: The table.
        quasi_identifiers: The columns an outsider could link on.
        k: The threshold to report against.

    Example:
        >>> rows = [{"sex": "f", "yob": 1980}] * 3 + [{"sex": "m", "yob": 1975}]
        >>> report = k_anonymity_report(rows, ["sex", "yob"], k=3)
        >>> report.k_achieved, report.n_below_k, report.satisfied
        (1, 1, False)
    """
    if k < 1:
        raise ValueError("k must be at least 1")
    names = list(quasi_identifiers)
    if not names:
        raise ValueError("name at least one quasi-identifier")
    return _report(Counter(_key(row, names) for row in rows), names, k)

openbtk.deid.kanonymity.KAnonymityReport

Bases: BaseModel

How k-anonymous a table is. Counts only: no value from the data appears here.

Example

rows = [{"sex": "f"}, {"sex": "f"}, {"sex": "m"}] report = k_anonymity_report(rows, ["sex"], k=2) (report.k_achieved, report.n_unique, report.satisfied) (1, 1, False)

openbtk.deid.kanonymity.anonymise_to_k

anonymise_to_k(rows: Iterable[Mapping[str, object]], quasi_identifiers: Sequence[str], *, k: int, ladders: Mapping[str, Sequence[Generaliser]] | None = None, max_suppression: float = 0.05, keep: Sequence[str] = ()) -> AnonymisationResult

Generalise (and if need be suppress) rows until they are k-anonymous.

Parameters:

Name	Type	Description	Default
rows	Iterable[Mapping[str, object]]	The table. Materialised once, as the algorithm needs every record.	required
quasi_identifiers	Sequence[str]	The columns to generalise.	required
k	int	The class size to reach.	required
ladders	Mapping[str, Sequence[Generaliser]] | None	Per column, a sequence of levels: level 0 first (usually the value as it is), each next one coarser, ending in * if the column may vanish. Columns not given here use :data:DEFAULT_LADDERS, then value -> *.	None
max_suppression	float	The fraction of records that may be removed instead of generalising further. 0 never suppresses until the ladders run out.	0.05
keep	Sequence[str]	Extra columns copied through untouched (a pseudonymous id, say). Nothing else is exported.	()

Returns:

Type	Description
AnonymisationResult	The exported rows and an account of the levels reached and records suppressed.

Example

rows = [{"yob": 1980 + i % 3, "sex": "f"} for i in range(9)] result = anonymise_to_k(rows, ["yob", "sex"], k=3) result.report.satisfied, result.n_suppressed (True, 0)

Source code in src/openbtk/deid/kanonymity.py

def anonymise_to_k(
    rows: Iterable[Mapping[str, object]],
    quasi_identifiers: Sequence[str],
    *,
    k: int,
    ladders: Mapping[str, Sequence[Generaliser]] | None = None,
    max_suppression: float = 0.05,
    keep: Sequence[str] = (),
) -> AnonymisationResult:
    """Generalise (and if need be suppress) ``rows`` until they are k-anonymous.

    Args:
        rows: The table. Materialised once, as the algorithm needs every record.
        quasi_identifiers: The columns to generalise.
        k: The class size to reach.
        ladders: Per column, a sequence of levels: level 0 first (usually the value as
            it is), each next one coarser, ending in ``*`` if the column may vanish.
            Columns not given here use :data:`DEFAULT_LADDERS`, then ``value -> *``.
        max_suppression: The fraction of records that may be removed instead of
            generalising further. ``0`` never suppresses until the ladders run out.
        keep: Extra columns copied through untouched (a pseudonymous id, say). Nothing
            else is exported.

    Returns:
        The exported rows and an account of the levels reached and records suppressed.

    Example:
        >>> rows = [{"yob": 1980 + i % 3, "sex": "f"} for i in range(9)]
        >>> result = anonymise_to_k(rows, ["yob", "sex"], k=3)
        >>> result.report.satisfied, result.n_suppressed
        (True, 0)
    """
    if k < 1:
        raise ValueError("k must be at least 1")
    if not 0 <= max_suppression <= 1:
        raise ValueError("max_suppression must be between 0 and 1")
    names = list(quasi_identifiers)
    if not names:
        raise ValueError("name at least one quasi-identifier")

    table = [dict(row) for row in rows]
    chosen: dict[str, Sequence[Generaliser]] = {}
    for name in names:
        given = (ladders or {}).get(name)
        ladder = given if given is not None else DEFAULT_LADDERS.get(name)
        ladder = _FALLBACK_LADDER if ladder is None else ladder
        if not 1 <= len(ladder) <= _MAX_LEVELS:
            raise ValueError(
                f"the ladder for {name!r} must have 1-{_MAX_LEVELS} levels"
            )
        chosen[name] = ladder
    level = dict.fromkeys(names, 0)

    def keys() -> list[tuple[Hashable, ...]]:
        return [tuple(chosen[n][level[n]](row.get(n)) for n in names) for row in table]

    current = keys()
    while table:
        classes = Counter(current)
        below = sum(size for size in classes.values() if size < k)
        if below <= max_suppression * len(table):
            break
        movable = [n for n in names if level[n] < len(chosen[n]) - 1]
        if not movable:
            break
        distinct = {
            n: len({chosen[n][level[n]](row.get(n)) for row in table}) for n in movable
        }
        level[max(movable, key=lambda n: (distinct[n], -names.index(n)))] += 1
        current = keys()

    classes = Counter(current)
    kept = [i for i, key in enumerate(current) if classes[key] >= k]
    out: list[dict[str, JsonValue]] = []
    for i in kept:
        exported: dict[str, JsonValue] = {
            n: _plain(chosen[n][level[n]](table[i].get(n))) for n in names
        }
        for extra in keep:
            exported[extra] = _plain(table[i].get(extra))
        out.append(exported)

    return AnonymisationResult(
        rows=out,
        report=_report(Counter(current[i] for i in kept), names, k),
        levels=dict(level),
        n_input=len(table),
        n_suppressed=len(table) - len(kept),
    )

openbtk.deid.kanonymity.AnonymisationResult

Bases: BaseModel

The exported table and an account of how it was made.

rows holds the generalised quasi-identifiers and any keep columns, and nothing else: columns that were not named are dropped, so an unlisted column cannot leak through.

openbtk.deid.kanonymity.band

band(width: int, *, top: int | None = None) -> Generaliser

Group integers into bands of width (1987 -> "1985-1989" for 5).

Parameters:

Name	Type	Description	Default
width	int	Band width, at least 2.	required
top	int | None	Values at or above this are written "<top>+". None disables it.	None

Non-integers (including None) pass through, so an unknown stays unknown rather than being placed in a band.

Example

band(5)(1987) '1985-1989' band(10, top=90)(93) '90+'

Source code in src/openbtk/deid/kanonymity.py

def band(width: int, *, top: int | None = None) -> Generaliser:
    """Group integers into bands of ``width`` (``1987`` -> ``"1985-1989"`` for 5).

    Args:
        width: Band width, at least 2.
        top: Values at or above this are written ``"<top>+"``. ``None`` disables it.

    Non-integers (including ``None``) pass through, so an unknown stays unknown rather
    than being placed in a band.

    Example:
        >>> band(5)(1987)
        '1985-1989'
        >>> band(10, top=90)(93)
        '90+'
    """
    if width < 2:
        raise ValueError("band width must be at least 2")

    def apply(value: object) -> Hashable:
        if isinstance(value, bool) or not isinstance(value, int):
            return _plain(value)
        if top is not None and value >= top:
            return f"{top}+"
        low = (value // width) * width
        return f"{low}-{low + width - 1}"

    return apply

openbtk.deid.kanonymity.top_code

top_code(limit: int) -> Generaliser

Write every integer at or above limit as "<limit>+" (HIPAA's 90+).

Example

top_code(90)(93), top_code(90)(41) ('90+', 41)

Source code in src/openbtk/deid/kanonymity.py

def top_code(limit: int) -> Generaliser:
    """Write every integer at or above ``limit`` as ``"<limit>+"`` (HIPAA's ``90+``).

    Example:
        >>> top_code(90)(93), top_code(90)(41)
        ('90+', 41)
    """

    def apply(value: object) -> Hashable:
        if isinstance(value, int) and not isinstance(value, bool) and value >= limit:
            return f"{limit}+"
        return _plain(value)

    return apply

openbtk.deid.kanonymity.prefix

prefix(digits: int) -> Generaliser

Keep the first digits characters, mask the rest ("90210" -> "902**").

Example

prefix(3)("90210") '902**'

Source code in src/openbtk/deid/kanonymity.py

def prefix(digits: int) -> Generaliser:
    """Keep the first ``digits`` characters, mask the rest (``"90210"`` -> ``"902**"``).

    Example:
        >>> prefix(3)("90210")
        '902**'
    """
    if digits < 1:
        raise ValueError("digits must be at least 1")

    def apply(value: object) -> Hashable:
        if not isinstance(value, str):
            return _plain(value)
        return value[:digits] + "*" * max(0, len(value) - digits)

    return apply

openbtk.deid.kanonymity.zip_ladder

zip_ladder(length: int = 5) -> tuple[Generaliser, ...]

A ladder for a ZIP code: the full code, then ever shorter prefixes, then *.

Example

levels = [level("90210") for level in zip_ladder(5)] levels ['90210', '9021', '902', '90', '9*', '']

Source code in src/openbtk/deid/kanonymity.py

def zip_ladder(length: int = 5) -> tuple[Generaliser, ...]:
    """A ladder for a ZIP code: the full code, then ever shorter prefixes, then ``*``.

    Example:
        >>> levels = [level("90210") for level in zip_ladder(5)]
        >>> levels
        ['90210', '9021*', '902**', '90***', '9****', '*']
    """
    return (identity, *(prefix(d) for d in range(length - 1, 0, -1)), star)

openbtk.deid.kanonymity.identity

identity(value: object) -> Hashable

Level 0: the value as it is (dates as ISO strings).

Example

identity("female") 'female'

Source code in src/openbtk/deid/kanonymity.py

def identity(value: object) -> Hashable:
    """Level 0: the value as it is (dates as ISO strings).

    Example:
        >>> identity("female")
        'female'
    """
    return _plain(value)

openbtk.deid.kanonymity.star

star(value: object) -> Hashable

The top level: every value becomes *, so the attribute stops distinguishing.

Example

star("anything") '*'

Source code in src/openbtk/deid/kanonymity.py

def star(value: object) -> Hashable:
    """The top level: every value becomes ``*``, so the attribute stops distinguishing.

    Example:
        >>> star("anything")
        '*'
    """
    del value
    return SUPPRESSED

Extending: recognizers

A recognizer finds candidate PHI spans; the engine merges them, filters by confidence and transforms them. Implement BaseRecognizer and name it in DeidEngine(recognizers=[...]) to add your own detector.

openbtk.deid.recognizers.base.BaseRecognizer

Bases: Component

Detect PHI spans in a single document's text.

Unlike BaseLoader/BaseChunker, detect() returns a plain list, not an Iterator: a single document produces a small, bounded number of PHI spans (tens, not millions), so there is no O(corpus) memory hazard here for ADR-0004's streaming rule to guard against -- the corpus-level streaming boundary is the loader's job, one document at a time.

method class-attribute

method: Literal['rule', 'ner', 'llm_verifier']

Stamped onto every Detection this recognizer produces (docs/04_API_DESIGN.md section 5). "ensemble" is reserved for SpanMerger's own output and is never a single recognizer's value.

requires_model_download class-attribute

requires_model_download: bool = False

True for a recognizer whose detect() needs a downloaded model (e.g. NERRecognizer), as opposed to one that's ready the moment it's constructed (RuleRecognizer). Lets test infrastructure (the contract suite, CI) skip exercising detect() for such a recognizer unless OPENBTK_SLOW_TESTS=1 is set, without needing to know about each concrete recognizer by name. Constructing an instance is always cheap regardless -- CLAUDE.md rule 11 still applies; this flag is about detect(), never about __init__.

detect abstractmethod

detect(text: str) -> list[Detection]

Find every PHI span this recognizer can find in text.

Parameters:

Name	Type	Description	Default
text	str	The document text to scan. Never logged, never placed in a raised exception's message or context.	required

Returns:

Type	Description
list[Detection]	Zero or more Detection objects, in no particular order.
list[Detection]	Confidence is this recognizer's own estimate; final confidence
list[Detection]	combination is SpanMerger's job, not this method's.

Raises:

Type	Description
DeidError	On a recognizer-internal failure. Never returns a silently-partial result -- a caller needs to know detection was incomplete, not trust an accidentally-short list.

Source code in src/openbtk/deid/recognizers/base.py

@abstractmethod
def detect(self, text: str) -> list[Detection]:
    """Find every PHI span this recognizer can find in ``text``.

    Args:
        text: The document text to scan. Never logged, never placed in
            a raised exception's message or context.

    Returns:
        Zero or more ``Detection`` objects, in no particular order.
        Confidence is this recognizer's own estimate; final confidence
        combination is ``SpanMerger``'s job, not this method's.

    Raises:
        DeidError: On a recognizer-internal failure. Never returns a
            silently-partial result -- a caller needs to know detection
            was incomplete, not trust an accidentally-short list.
    """

openbtk.deid.recognizers.rule.RuleRecognizer

Bases: BaseRecognizer

High-precision, regex-based PHI detection. See module docstring for exactly which categories this does and does not attempt.

Example

r = RuleRecognizer() hits = r.detect("SSN: 123-45-6789") hits[0].category