Normalizing Site and Investigator Identifiers Across Systems

This walkthrough builds the identity half of the model described in Clinical Metadata Normalization Across Trial Systems, inside Automated Document Ingestion & Validation Workflows: resolving a site number and a principal investigator to one stable canonical identifier when the CTMS, the EDC, and the IRB each hold a slightly different representation of the same entity. Where the parent guide covers field mapping and coercion broadly, this page focuses narrowly on the matching problem itself — deciding when “0042”, “SITE-0042-US”, and the institution name in an IRB letter really do refer to the same site, and when “Dr. Robert J. Smith” and “R. Smith” really are the same investigator. Every resolved match, and every match the resolver refuses to make, is written to the append-only audit log.

Why naive approaches fail

Identity resolution invites shortcuts that look reasonable until a real dataset breaks them:

  • Exact string equality on the raw field. "0042" == "SITE-0042-US" is False, so a naive equality check treats every source system’s own numbering convention as a different site, fragmenting one real site into three canonical records.
  • Case- and punctuation-insensitive string matching alone. Stripping case and whitespace closes the gap for some pairs but does nothing for “Smith, Robert” versus “Robert J. Smith” or for a site number embedded inside a longer identifier string. It also produces false positives — two different investigators who happen to share a common surname and lowercased initial.
  • A trained similarity model as the final decision. A model can rank “R. Smith” as 94% likely to match “Robert J. Smith” — but a percentage is not a reason a data manager can cite when an inspector asks why two records were merged. The decision has to be traceable to a specific rule: an exact NPI match, or an explicit, confirmed alias.
  • Silently picking whichever value looks “more complete.” Preferring the longer or more detailed-looking string over the shorter one is not a documented rule; it is an implementation accident that happens to work on the current dataset and breaks the moment a new source system’s export style changes.

Architecture overview

The resolver is a short pipeline: extract a matching key for each candidate record, group records that share a key, apply a fixed precedence rule to pick the representative value for each field, and emit the resolved identity with its full source trail.

Identity resolution pipeline Five stages left to right: extract matching key, group by key, apply precedence, resolve identity, and emit audit record. Each stage feeds the next with a single arrow. Matching key Group by key Precedence Resolve ID Audit record

Setup and configuration

No third-party library is required for the matching logic itself — deterministic string normalization and dictionary grouping cover it entirely. Country and date coercion, if a caller pulls it in from the parent guide, uses pycountry, but the resolver below only depends on the standard library.

pip install "pycountry>=23.12"

The alias table that maps known name and site-number variants onto a canonical spelling is external configuration, not inline code, so regulatory operations can extend it without a Python change:

"""Environment-driven configuration for the identity resolver."""
from __future__ import annotations

import os
from dataclasses import dataclass


@dataclass(frozen=True)
class ResolverConfig:
    alias_table_path: str
    require_npi_for_auto_merge: bool

    @classmethod
    def from_env(cls) -> "ResolverConfig":
        def required(name: str) -> str:
            value = os.environ.get(name)
            if not value:
                raise RuntimeError(f"Missing required environment variable: {name}")
            return value

        return cls(
            alias_table_path=required("IDENTITY_ALIAS_TABLE_PATH"),
            require_npi_for_auto_merge=os.environ.get("REQUIRE_NPI_FOR_AUTO_MERGE", "true") == "true",
        )

Full working implementation

Site numbers are normalized by stripping non-alphanumeric characters and leading zeros, so "0042", "SITE-0042-US", and "Site 42" all collapse to the same comparison key once the protocol prefix is separated out.

"""Normalize a raw site-number string into a stable matching key."""
from __future__ import annotations

import re


def site_matching_key(protocol_no: str, raw_site_number: str) -> str:
    """Strip separators and leading zeros so equivalent spellings converge.

    'SITE-0042-US' and '0042' both become 'PROTO123:42' when protocol_no
    is 'PROTO123', regardless of which source system supplied the string.
    """
    digits_only = re.sub(r"[^0-9]", "", raw_site_number)
    normalized = digits_only.lstrip("0") or "0"
    return f"{protocol_no}:{normalized}"

Investigator matching prefers an exact National Provider Identifier match when one is available on both sides, and only falls back to normalized-name matching — never a fuzzy score — when no NPI exists to compare.

"""Typed source record and canonical investigator, plus deterministic matching."""
from __future__ import annotations

import re
import unicodedata
from dataclasses import dataclass, field
from enum import Enum


class SourceSystem(str, Enum):
    CTMS = "ctms"
    EDC = "edc"
    IRB = "irb"


@dataclass(frozen=True)
class InvestigatorRecord:
    system: SourceSystem
    entity_key: str
    family_name: str
    given_name: str
    npi: str | None


@dataclass(frozen=True)
class CanonicalInvestigator:
    canonical_pi_id: str
    family_name: str
    given_name: str
    npi: str | None
    source_ids: dict[SourceSystem, str] = field(default_factory=dict)


def _normalize_name(value: str) -> str:
    """Fold case, strip titles/suffixes and diacritics for name comparison.

    This is deterministic normalization, not fuzzy matching: it removes
    known noise (Dr., MD, punctuation, accents) but never guesses at
    whether two different-looking names might be the same person.
    """
    stripped = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
    stripped = re.sub(r"\b(dr|md|phd|jr|sr|ii|iii)\b\.?", "", stripped, flags=re.IGNORECASE)
    return re.sub(r"[^a-z]", "", stripped.lower())


def investigator_matching_key(record: InvestigatorRecord) -> str:
    """Return the NPI if present; otherwise a normalized family+given name key.

    An NPI match is authoritative on its own. Without one, two records
    only match if their normalized names are identical — a near-miss like
    'R. Smith' vs 'Robert Smith' is deliberately NOT collapsed here; that
    kind of ambiguous case is exactly what the alias table below exists
    to resolve explicitly, one confirmed pair at a time.
    """
    if record.npi:
        return f"npi:{record.npi}"
    return f"name:{_normalize_name(record.family_name)}:{_normalize_name(record.given_name)}"

Grouping and precedence-based resolution mirror the site logic: records that share a matching key are merged, and a fixed system order breaks any field-level disagreement within the group.

"""Group investigator records by matching key and resolve each group."""
from __future__ import annotations

from collections import defaultdict

_PRECEDENCE = (SourceSystem.CTMS, SourceSystem.EDC, SourceSystem.IRB)


def resolve_investigators(
    records: list[InvestigatorRecord],
) -> tuple[list[CanonicalInvestigator], list[list[InvestigatorRecord]]]:
    """Return (resolved canonical investigators, unresolved conflict groups).

    A conflict group is any set of records whose keys did not converge —
    for example, an explicit alias-table entry is required but missing.
    Nothing in a conflict group is merged automatically.
    """
    groups: dict[str, list[InvestigatorRecord]] = defaultdict(list)
    for record in records:
        groups[investigator_matching_key(record)].append(record)

    resolved: list[CanonicalInvestigator] = []
    conflicts: list[list[InvestigatorRecord]] = []
    for key, group in groups.items():
        systems_present = {r.system for r in group}
        if len(systems_present) < len(group):
            # Same system contributed more than one record under this key —
            # a genuine duplicate within one source, not a cross-system match.
            conflicts.append(group)
            continue

        by_system = {r.system: r for r in group}
        winner = next(by_system[s] for s in _PRECEDENCE if s in by_system)
        resolved.append(
            CanonicalInvestigator(
                canonical_pi_id=key,
                family_name=winner.family_name,
                given_name=winner.given_name,
                npi=next((r.npi for r in group if r.npi), None),
                source_ids={r.system: r.entity_key for r in group},
            )
        )
    return resolved, conflicts

Finally, the resolution decision is written to the audit trail with the full contributing set, so a reviewer can see every source record that fed a canonical investigator without re-querying the source systems.

"""Emit an audit record for a resolved investigator identity."""
from __future__ import annotations

import json
import logging

logger = logging.getLogger("ingestion.identity_resolution")


def audit_investigator_resolution(investigator: CanonicalInvestigator, group: list[InvestigatorRecord]) -> None:
    logger.info(
        json.dumps(
            {
                "action": "investigator_resolved",
                "canonical_pi_id": investigator.canonical_pi_id,
                "npi": investigator.npi,
                "matched_by": "npi" if investigator.npi else "normalized_name",
                "sources": [
                    {"system": r.system.value, "entity_key": r.entity_key}
                    for r in group
                ],
            },
            sort_keys=True,
        )
    )

Validation and edge-case handling

A few situations account for most of the friction in real identity resolution, and each is handled by an explicit branch above rather than a fallback guess:

  • Two records from the same source system share a matching key. That is flagged as a conflict rather than merged, because it usually means the source system itself has a duplicate or data-entry error — merging silently would hide that defect instead of surfacing it for correction.
  • A near-miss name with no NPI on either side. “R. Smith” and “Robert Smith” produce different normalized-name keys by design; the resolver treats them as two separate candidates and leaves the merge to an explicit alias-table entry a data manager confirms once, rather than a similarity threshold that could just as easily merge two different investigators who happen to share a common name.
  • An NPI present on one record but missing on the other. The record with the NPI cannot key-match a record without one under investigator_matching_key, so this pairing will not auto-resolve; it is a legitimate case for the alias table, where a reviewer confirms the pairing once and it becomes a deterministic rule from then on.
  • A protocol number typo that changes the site’s matching key entirely. Because site_matching_key includes the protocol number, a typo in either field produces a key that matches nothing, which surfaces as an unresolved site rather than a silent merge into the wrong protocol — the safer failure direction for this kind of error.

Testing and verification

The test suite proves the three properties that matter most: NPI match takes precedence, near-miss names are never auto-merged, and same-system duplicates are flagged rather than merged.

"""Tests for the site and investigator identity resolver."""
from __future__ import annotations

import pytest


def test_site_matching_key_normalizes_equivalent_spellings() -> None:
    assert site_matching_key("PROTO123", "0042") == site_matching_key("PROTO123", "SITE-0042-US")


def test_npi_match_takes_precedence_over_name_differences() -> None:
    records = [
        InvestigatorRecord(SourceSystem.CTMS, "ctms-1", "Smith", "Robert", "1234567890"),
        InvestigatorRecord(SourceSystem.EDC, "edc-1", "Smith", "Bob", "1234567890"),
    ]
    resolved, conflicts = resolve_investigators(records)
    assert conflicts == []
    assert len(resolved) == 1
    assert resolved[0].npi == "1234567890"
    assert resolved[0].source_ids == {SourceSystem.CTMS: "ctms-1", SourceSystem.EDC: "edc-1"}


def test_near_miss_names_without_npi_are_not_auto_merged() -> None:
    records = [
        InvestigatorRecord(SourceSystem.CTMS, "ctms-2", "Smith", "Robert", None),
        InvestigatorRecord(SourceSystem.EDC, "edc-2", "Smith", "R.", None),
    ]
    resolved, conflicts = resolve_investigators(records)
    assert len(resolved) == 2
    assert conflicts == []


def test_duplicate_records_from_one_system_are_flagged_as_conflict() -> None:
    records = [
        InvestigatorRecord(SourceSystem.CTMS, "ctms-3", "Lee", "Alice", "9876543210"),
        InvestigatorRecord(SourceSystem.CTMS, "ctms-4", "Lee", "Alice", "9876543210"),
    ]
    resolved, conflicts = resolve_investigators(records)
    assert resolved == []
    assert len(conflicts) == 1
    assert len(conflicts[0]) == 2

The second test is the one worth reading twice: “Robert” and “R.” on otherwise-matching records with no NPI produce two separate canonical investigators rather than one merged guess, which is exactly the deliberately conservative behavior this resolver is built to have.