Building CDISC datasets in Python

Lesson 9 — Python

Lesson 9 of 20 Advanced ~100 min

Learning objectives

Build an SDTM DM domain from raw source data end to end
Derive the standard variables: USUBJID, --SEQ, --DY, RFSTDTC
Build a small ADaM dataset with population flags and grouping variables
Apply a specification to enforce types, lengths, labels and order
Write a conformant XPT file and verify it
Judge honestly when Python is the right tool for this and when it is not

The honest framing first

Lesson 8 set out the ecosystem gap. For output it has largely closed — rtflite gives Python submission-quality RTF, covered in lesson 12. For derivation it has not: there is no admiral, no metacore, no xportr. Everything in this lesson is code you write and test yourself, where the R equivalent is a function call into a package that thousands of people have exercised.

That is a real cost and you should weigh it. Python is nonetheless the right choice when:

The study is already a Python shop and the toolchain is fixed
The source data arrives through a Python data pipeline
You need one or two domains, not a full submission
The work feeds a model or an application rather than a CSR

If you are producing a full submission package, the honest recommendation remains R and the pharmaverse. Read this lesson to understand what that ecosystem does for you.

Source data

A realistic raw extract — four files that do not resemble SDTM:

import pandas as pd
import numpy as np
from pathlib import Path

RAW = Path("data/raw")

demographics = pd.read_csv(RAW / "demog.csv", dtype=str)
randomisation = pd.read_csv(RAW / "randomisation.csv", dtype=str)
exposure = pd.read_csv(RAW / "exposure.csv", dtype=str)
disposition = pd.read_csv(RAW / "disposition.csv", dtype=str)

demographics.head(3)
#>   SUBJECT SITE BIRTHDT      SEX RACE_CD ETHNIC_CD COUNTRY
#> 0    0042  001  1954-08-12   F       1         2      USA
#> 1    0107  002  1948-03-25   M       2         1      DEU
#> 2    0203  001  1961-11-03   F       1         2      USA

Reading everything as dtype=str is deliberate — see lesson 7. Type conversion is a derivation, done consciously, not something the CSV parser guesses.

Building SDTM DM

Identifiers

STUDYID = "ABC-101"

dm = demographics.assign(
    STUDYID=STUDYID,
    DOMAIN="DM",
    SUBJID=lambda d: d["SUBJECT"].str.strip(),
    SITEID=lambda d: d["SITE"].str.strip().str.zfill(3),
)
dm["USUBJID"] = dm["STUDYID"] + "-" + dm["SITEID"] + "-" + dm["SUBJID"]

str.zfill(3) restores leading zeros. This is why the source was read as text — had SITE been parsed as an integer, "007" would already be 7 and the zero-padding would be guesswork.

Controlled terminology, with a failure on unmapped values

SEX_MAP    = {"F": "F", "M": "M", "FEMALE": "F", "MALE": "M", "U": "U"}
RACE_MAP   = {"1": "WHITE", "2": "BLACK OR AFRICAN AMERICAN",
              "3": "ASIAN", "4": "AMERICAN INDIAN OR ALASKA NATIVE",
              "5": "NATIVE HAWAIIAN OR OTHER PACIFIC ISLANDER", "6": "OTHER"}
ETHNIC_MAP = {"1": "HISPANIC OR LATINO", "2": "NOT HISPANIC OR LATINO",
              "3": "NOT REPORTED", "4": "UNKNOWN"}


def map_ct(series: pd.Series, mapping: dict, variable: str) -> pd.Series:
    """Map to controlled terminology, raising on any unmapped value."""
    key = series.str.strip().str.upper()
    out = key.map(mapping)

    unmapped = key[out.isna() & key.notna() & (key != "")].unique()
    if len(unmapped):
        raise ValueError(
            f"{variable}: {len(unmapped)} unmapped value(s): {sorted(unmapped)}. "
            "Extend the mapping or raise a data query."
        )
    return out


dm = dm.assign(
    SEX=lambda d: map_ct(d["SEX"], SEX_MAP, "SEX"),
    RACE=lambda d: map_ct(d["RACE_CD"], RACE_MAP, "RACE"),
    ETHNIC=lambda d: map_ct(d["ETHNIC_CD"], ETHNIC_MAP, "ETHNIC"),
)

Raise, do not default to missing

out = key.map(mapping)          # unmapped -> NaN, silently

A severity of "Grade 3" that the mapping does not anticipate becomes missing and flows all the way to a results table. The unmapped check turns a silent data-loss bug into an immediate, named failure.

This is the single highest-value check in SDTM mapping, in any language — the same argument as in SDTM programming in R.

Dates and reference dates

ex = exposure.assign(
    USUBJID=lambda d: STUDYID + "-" + d["SITE"].str.zfill(3) + "-" + d["SUBJECT"],
    EXSTDTC=lambda d: pd.to_datetime(d["DOSE_START"], errors="coerce"),
    EXENDTC=lambda d: pd.to_datetime(d["DOSE_END"], errors="coerce"),
    EXDOSE=lambda d: pd.to_numeric(d["DOSE_MG"], errors="coerce"),
)

# First and last dose of actual treatment
dosing = ex[(ex["EXDOSE"] > 0) | (ex["EXTRT"].str.contains("PLACEBO", na=False))]

ref_dates = dosing.groupby("USUBJID", as_index=False).agg(
    RFXSTDTC=("EXSTDTC", "min"),
    RFXENDTC=("EXENDTC", "max"),
)

dm = dm.merge(ref_dates, on="USUBJID", how="left", validate="one_to_one")

The EXDOSE > 0 or PLACEBO filter is the standard idiom: a zero dose is real exposure on a placebo arm and a non-dose on an active arm.

validate="one_to_one" is doing real work — if ref_dates somehow had two rows for a subject, DM would silently gain a row.

ISO 8601 output

SDTM --DTC variables are character strings, not dates, and partial dates must stay partial:

def to_dtc(dates: pd.Series) -> pd.Series:
    """Format a datetime Series as ISO 8601 date strings, blank where missing."""
    return dates.dt.strftime("%Y-%m-%d").fillna("")


dm = dm.assign(
    BRTHDTC=lambda d: to_dtc(pd.to_datetime(d["BIRTHDT"], errors="coerce")),
    RFSTDTC=lambda d: to_dtc(d["RFXSTDTC"]),
    RFENDTC=lambda d: to_dtc(d["RFXENDTC"]),
    RFXSTDTC=lambda d: to_dtc(d["RFXSTDTC"]),
    RFXENDTC=lambda d: to_dtc(d["RFXENDTC"]),
)

Never impute a partial date in SDTM

If only the year and month were collected, AESTDTC is "2026-03" and that is correct and complete. Imputing to "2026-03-01" destroys the information that the day was unknown, and is a conformance finding.

Imputation belongs in ADaM, with an imputation flag recording that it happened.

Age

def compute_age(birth: pd.Series, reference: pd.Series) -> pd.Series:
    """Age in completed years at the reference date."""
    b, r = pd.to_datetime(birth), pd.to_datetime(reference)
    years = r.dt.year - b.dt.year
    # Subtract one if the birthday has not occurred by the reference date
    before_birthday = (r.dt.month, r.dt.day) < (b.dt.month, b.dt.day)
    return (years - ((r.dt.month * 100 + r.dt.day)
                     < (b.dt.month * 100 + b.dt.day)).astype(int)).astype("Int64")


dm = dm.assign(
    AGE=lambda d: compute_age(d["BRTHDTC"], d["RFSTDTC"]),
    AGEU="YEARS",
)

The month * 100 + day comparison avoids a tuple comparison that does not vectorise. Int64 — the nullable integer from lesson 5 — keeps age an integer when the birth date is missing, rather than silently becoming a float.

Treatment arms and disposition

rand = randomisation.assign(
    USUBJID=lambda d: STUDYID + "-" + d["SITE"].str.zfill(3) + "-" + d["SUBJECT"],
)

ARM_MAP = {"1": ("Placebo", "PBO"),
           "2": ("Drug A 50mg", "A50"),
           "3": ("Drug A 100mg", "A100")}

rand = rand.assign(
    ARM=lambda d: d["ARM_CD"].map(lambda x: ARM_MAP.get(x, (None, None))[0]),
    ARMCD=lambda d: d["ARM_CD"].map(lambda x: ARM_MAP.get(x, (None, None))[1]),
)

dm = dm.merge(
    rand[["USUBJID", "ARM", "ARMCD", "RANDDT"]],
    on="USUBJID", how="left", validate="one_to_one",
)

# Subjects who were randomised but never dosed
dm["ACTARM"]  = np.where(dm["RFSTDTC"] != "", dm["ARM"], "")
dm["ACTARMCD"] = np.where(dm["RFSTDTC"] != "", dm["ARMCD"], "")

# Screen failures
screen_fail = dm["ARMCD"].isna()
dm.loc[screen_fail, ["ARM", "ACTARM"]]     = "Screen Failure"
dm.loc[screen_fail, ["ARMCD", "ACTARMCD"]] = "SCRNFAIL"

Assembling and validating

DM_VARIABLES = [
    "STUDYID", "DOMAIN", "USUBJID", "SUBJID", "RFSTDTC", "RFENDTC",
    "RFXSTDTC", "RFXENDTC", "SITEID", "BRTHDTC", "AGE", "AGEU",
    "SEX", "RACE", "ETHNIC", "ARMCD", "ARM", "ACTARMCD", "ACTARM", "COUNTRY",
]

dm_final = dm[DM_VARIABLES].sort_values("USUBJID").reset_index(drop=True)


def validate_dm(df: pd.DataFrame) -> None:
    """Contract checks for the DM domain."""
    problems = []

    if df["USUBJID"].duplicated().any():
        n = df["USUBJID"].duplicated().sum()
        problems.append(f"DM must be one record per subject; {n} duplicate USUBJID")

    for var in ("STUDYID", "DOMAIN", "USUBJID", "SUBJID", "SITEID"):
        if df[var].isna().any() or (df[var].astype(str).str.strip() == "").any():
            problems.append(f"{var} is required and must be populated")

    if not (df["DOMAIN"] == "DM").all():
        problems.append("DOMAIN must be 'DM' on every record")

    bad_sex = set(df["SEX"].dropna()) - {"F", "M", "U", "UNDIFFERENTIATED"}
    if bad_sex:
        problems.append(f"SEX has values outside controlled terminology: {bad_sex}")

    iso = r"^$|^\d{4}(-\d{2}(-\d{2})?)?$"
    for var in [c for c in df.columns if c.endswith("DTC")]:
        invalid = df.loc[~df[var].fillna("").str.match(iso), var].unique()
        if len(invalid):
            problems.append(f"{var} has non-ISO8601 values: {list(invalid)[:3]}")

    age_bad = df["AGE"].dropna()
    if ((age_bad < 0) | (age_bad > 120)).any():
        problems.append("AGE outside the plausible range 0-120")

    if problems:
        raise ValueError("DM validation failed:\n  " + "\n  ".join(problems))
    print(f"DM validated: {len(df)} subjects, {len(df.columns)} variables")


validate_dm(dm_final)

A minimal ADSL

adsl = (
    dm_final
    .assign(
        TRTSDT=lambda d: pd.to_datetime(d["RFXSTDTC"], errors="coerce"),
        TRTEDT=lambda d: pd.to_datetime(d["RFXENDTC"], errors="coerce"),
    )
    .assign(
        TRT01P=lambda d: d["ARM"],
        TRT01A=lambda d: d["ACTARM"],
        TRTDURD=lambda d: (d["TRTEDT"] - d["TRTSDT"]).dt.days.add(1).astype("Int64"),
        SAFFL=lambda d: np.where(d["TRTSDT"].notna(), "Y", "N"),
        ITTFL=lambda d: np.where(d["ARMCD"] != "SCRNFAIL", "Y", "N"),
        AGEGR1=lambda d: pd.cut(
            d["AGE"].astype("Float64").astype(float),
            bins=[-np.inf, 65, 80, np.inf],
            labels=["<65", "65-80", ">80"],
            right=False,
        ).astype(str),
    )
    .assign(
        AGEGR1N=lambda d: d["AGEGR1"].map({"<65": 1, "65-80": 2, ">80": 3}).astype("Int64"),
        TRT01PN=lambda d: d["ARMCD"].map({"PBO": 0, "A50": 1, "A100": 2}).astype("Int64"),
    )
)

TRTDURD is inclusive of both endpoints

.dt.days.add(1) — ADaM defines treatment duration as TRTEDT - TRTSDT + 1. A subject dosed once on a single day has a duration of 1 day, not 0.

This is exactly the off-by-one that double programming finds, and it deserves a test:

def test_trtdurd_is_inclusive():
    d = pd.DataFrame({"TRTSDT": pd.to_datetime(["2026-03-15"]),
                      "TRTEDT": pd.to_datetime(["2026-03-15"])})
    assert derive_trtdurd(d).iloc[0] == 1

Applying a specification

Without metacore and xportr, write the equivalent yourself:

from dataclasses import dataclass


@dataclass
class VariableSpec:
    name: str
    label: str
    dtype: str          # "text" or "numeric"
    length: int
    order: int
    format: str | None = None


DM_SPEC = [
    VariableSpec("STUDYID", "Study Identifier",           "text",    20, 1),
    VariableSpec("DOMAIN",  "Domain Abbreviation",        "text",     2, 2),
    VariableSpec("USUBJID", "Unique Subject Identifier",  "text",    30, 3),
    VariableSpec("SUBJID",  "Subject Identifier",         "text",    10, 4),
    VariableSpec("RFSTDTC", "Subject Reference Start",    "text",    19, 5),
    VariableSpec("SITEID",  "Study Site Identifier",      "text",     6, 6),
    VariableSpec("AGE",     "Age",                        "numeric",  8, 7),
    VariableSpec("AGEU",    "Age Units",                  "text",     6, 8),
    VariableSpec("SEX",     "Sex",                        "text",     1, 9),
    VariableSpec("ARM",     "Description of Planned Arm", "text",    40, 10),
]


def apply_spec(df: pd.DataFrame, spec: list[VariableSpec]) -> tuple[pd.DataFrame, dict]:
    """Enforce a specification: presence, types, lengths, order and labels."""
    spec_names = [v.name for v in spec]

    missing = set(spec_names) - set(df.columns)
    if missing:
        raise ValueError(f"Dataset is missing specified variable(s): {sorted(missing)}")

    extra = set(df.columns) - set(spec_names)
    if extra:
        print(f"Dropping {len(extra)} unspecified variable(s): {sorted(extra)}")

    out = df[spec_names].copy()
    problems = []

    for v in spec:
        if v.dtype == "text":
            out[v.name] = out[v.name].fillna("").astype(str)
            actual = out[v.name].str.encode("utf-8").str.len().max()
            if actual and actual > v.length:
                worst = out[v.name].loc[out[v.name].str.len().idxmax()]
                problems.append(
                    f"{v.name}: {actual} bytes exceeds specified length {v.length} "
                    f"— values WILL be truncated. Longest: '{worst[:50]}'"
                )
        else:
            out[v.name] = pd.to_numeric(out[v.name], errors="coerce")

    if problems:
        raise ValueError("Specification violations:\n  " + "\n  ".join(problems))

    out = out[[v.name for v in sorted(spec, key=lambda x: x.order)]]
    labels = {v.name: v.label for v in spec}
    return out, labels

The length check runs before anything is written. XPT truncation is silent — the file is produced, no error is raised, and the loss is only discoverable by reading the file back.

Writing XPT

import pyreadstat


def write_xpt(df: pd.DataFrame, labels: dict, path: str,
              table_name: str, file_label: str = "") -> None:
    """Write a SAS transport V5 file after checking the format's constraints."""
    problems = []

    if len(table_name) > 8:
        problems.append(f"table_name '{table_name}' exceeds 8 characters")
    if len(file_label) > 40:
        problems.append(f"file_label exceeds 40 characters")

    long_names = [c for c in df.columns if len(c) > 8]
    if long_names:
        problems.append(f"Variable names over 8 characters: {long_names}")

    stems = pd.Series([c[:8].upper() for c in df.columns])
    collisions = stems[stems.duplicated()].unique().tolist()
    if collisions:
        affected = [c for c in df.columns if c[:8].upper() in collisions]
        problems.append(f"Names collide when truncated to 8: {affected}")

    long_labels = {k: v for k, v in labels.items() if len(v) > 40}
    if long_labels:
        problems.append(f"Labels over 40 characters: {list(long_labels)}")

    for col in df.select_dtypes(include=["object", "string"]).columns:
        s = df[col].dropna().astype(str)
        if s.empty:
            continue
        if s.str.contains(r"[^\x00-\x7F]", regex=True).any():
            examples = s[s.str.contains(r"[^\x00-\x7F]", regex=True)].unique()[:2]
            problems.append(f"{col}: non-ASCII characters, e.g. {list(examples)}")

    if problems:
        raise ValueError("XPT V5 violations:\n  " + "\n  ".join(problems))

    pyreadstat.write_xport(
        df, path,
        file_format_version=5,
        table_name=table_name,
        file_label=file_label,
        column_labels=[labels.get(c, "") for c in df.columns],
    )
    print(f"Wrote {path}: {len(df)} records, {len(df.columns)} variables")


dm_spec_applied, dm_labels = apply_spec(dm_final, DM_SPEC)
write_xpt(dm_spec_applied, dm_labels, "data/submission/dm.xpt",
          table_name="DM", file_label="Demographics")

Verify the round trip

returned, meta = pyreadstat.read_xport("data/submission/dm.xpt")

assert len(returned) == len(dm_spec_applied), "row count changed"
assert list(returned.columns) == list(dm_spec_applied.columns), "column order changed"

for col in dm_spec_applied.select_dtypes(include=["object"]).columns:
    before = dm_spec_applied[col].astype(str)
    after  = returned[col].astype(str).str.rstrip()      # SAS pads to length
    if not before.equals(after):
        n = (before != after).sum()
        raise AssertionError(f"{col}: {n} value(s) changed in the round trip")

print("Round trip verified — no silent truncation")

Run this at least once per study on the first dataset. It is the only way to detect truncation before Pinnacle 21 does.

What you are giving up

`admiral` / `xportr` gives you	Here you write
`derive_vars_merged()`	A validated `merge` with cardinality checks
`derive_var_trtdurd()`	The `+1` convention, and a test for it
`derive_vars_dt(highest_imputation=)`	Imputation logic plus the flag
`restrict_derivation()`	A boolean mask applied to a subset
`xportr_length()` with pre-checks	The byte-width check above
Tested edge cases	Your own test suite
A published specification	Your docstrings

Every row is achievable — none is free. Budget for the tests, because in this layer they are the only thing standing between a convention error and a wrong number in a table.

Common mistakes

Mistake	Consequence	Fix
Reading source with inferred dtypes	Leading zeros lost	`dtype=str`, convert deliberately
Unmapped CT values become `NaN`	Silent data loss	Raise on unmapped
`merge` without `validate=`	Row multiplication	Declare the cardinality
Imputing partial dates in SDTM	Information destroyed	Keep partial; impute in ADaM
`TRTDURD` without `+1`	Off-by-one throughout	Inclusive of both endpoints
Checking lengths after writing	Silent truncation	Check before
Measuring width in characters	XPT violations	`.str.encode("utf-8").str.len()`
Skipping the round-trip check	Truncation found by the regulator	Verify once per study

Exercise 9.1 — Derive a treatment-emergent flag with imputation

Write a function that adds ASTDT, ASTDTF (imputation flag) and TRTEMFL to an adverse events DataFrame, imputing partial start dates to the earliest possible date but never earlier than treatment start.

Show solution

import pandas as pd
import numpy as np


def derive_trtemfl(
    ae: pd.DataFrame,
    adsl: pd.DataFrame,
    window: int = 30,
) -> pd.DataFrame:
    """Derive ASTDT, ASTDTF and TRTEMFL on an adverse events dataset.

    Partial AESTDTC values are imputed to the earliest possible date, then
    constrained to be no earlier than TRTSDT — the standard conservative
    convention, which treats an event whose partial date is consistent with
    the treatment period as treatment emergent.

    Args:
        ae: Must contain USUBJID and AESTDTC (ISO 8601, possibly partial).
        adsl: Must contain USUBJID, TRTSDT, TRTEDT.
        window: Days after last dose still considered emergent.

    Returns:
        `ae` with ASTDT, ASTDTF and TRTEMFL added. Row count is preserved.
    """
    required = {"USUBJID", "AESTDTC"}
    if missing := required - set(ae.columns):
        raise ValueError(f"ae is missing: {sorted(missing)}")

    n_in = len(ae)

    out = ae.merge(
        adsl[["USUBJID", "TRTSDT", "TRTEDT"]],
        on="USUBJID", how="left", validate="many_to_one",
    )
    orphans = out["TRTSDT"].isna() & out["USUBJID"].notna()
    if (unmatched := set(ae["USUBJID"]) - set(adsl["USUBJID"])):
        raise ValueError(f"AE subjects not found in ADSL: {sorted(unmatched)[:5]}")

    dtc = out["AESTDTC"].fillna("").str.slice(0, 10)

    # --- Classify precision -------------------------------------------------
    is_full  = dtc.str.match(r"^\d{4}-\d{2}-\d{2}$")
    is_month = dtc.str.match(r"^\d{4}-\d{2}$")
    is_year  = dtc.str.match(r"^\d{4}$")

    # --- Impute to the earliest possible date -------------------------------
    imputed = pd.Series(pd.NaT, index=out.index, dtype="datetime64[ns]")
    imputed[is_full]  = pd.to_datetime(dtc[is_full],  errors="coerce")
    imputed[is_month] = pd.to_datetime(dtc[is_month] + "-01", errors="coerce")
    imputed[is_year]  = pd.to_datetime(dtc[is_year]  + "-01-01", errors="coerce")

    # --- Imputation flag, BEFORE the min_dates constraint -------------------
    out["ASTDTF"] = np.select(
        [is_full, is_month, is_year],
        [None,    "D",      "M"],
        default="Y",                      # nothing usable
    )
    out.loc[dtc == "", "ASTDTF"] = None   # genuinely missing, not imputed

    # --- Constrain: never earlier than treatment start ----------------------
    # Only applies where a date was actually imputed; a complete date stands.
    constrain = (~is_full) & imputed.notna() & out["TRTSDT"].notna()
    imputed[constrain] = np.maximum(
        imputed[constrain].values, out.loc[constrain, "TRTSDT"].values
    )
    out["ASTDT"] = imputed

    # --- Treatment-emergent flag --------------------------------------------
    upper = out["TRTEDT"] + pd.Timedelta(days=window)

    out["TRTEMFL"] = np.select(
        [
            out["TRTSDT"].isna(),                       # untreated
            out["ASTDT"].isna(),                        # unknown date, treated
            out["ASTDT"] < out["TRTSDT"],               # before first dose
            out["TRTEDT"].isna(),                       # ongoing treatment
            out["ASTDT"] <= upper,                      # within the window
        ],
        ["N", "Y", "N", "Y", "Y"],
        default="N",
    )

    assert len(out) == n_in, "row count changed"
    return out

Verification:

ae = pd.DataFrame({
    "USUBJID": ["001"] * 5,
    "AESTDTC": ["2026-04-01", "2026-03", "2026", "", "2026-03-10"],
})
adsl = pd.DataFrame({
    "USUBJID": ["001"],
    "TRTSDT":  pd.to_datetime(["2026-03-15"]),
    "TRTEDT":  pd.to_datetime(["2026-06-15"]),
})

derive_trtemfl(ae, adsl)[["AESTDTC", "ASTDT", "ASTDTF", "TRTEMFL"]]
#>       AESTDTC      ASTDT ASTDTF TRTEMFL
#> 0  2026-04-01 2026-04-01   None       Y
#> 1     2026-03 2026-03-15      D       Y
#> 2        2026 2026-03-15      M       Y
#> 3                    NaT   None       Y
#> 4  2026-03-10 2026-03-10   None       N

Four design points worth drawing out:

The min_dates constraint is the crux. Row 1 is recorded as “March 2026”. Imputed naively to 1 March, it precedes treatment start on the 15th and is classified not treatment emergent. Constrained to no earlier than TRTSDT, it becomes 15 March and is emergent. Which is correct is an SAP decision — but it must be a decision. admiral exposes this as the min_dates argument precisely because it cannot be assumed.

Row 4 (complete date, 10 March) is correctly "N". The constraint applies only where imputation happened. A complete date that genuinely precedes treatment must not be dragged forward — that would fabricate data.

ASTDTF is set before the constraint, so it records the precision of the source, not the effect of the constraint. A reviewer needs to know the day was unknown regardless of what the imputation rule then did.

Row 3, a missing date in a treated subject, is "Y". Conservative: an event that might be emergent is counted as emergent. Some SAPs specify otherwise, and the np.select ordering makes the convention explicit and easy to change.

Tests for the boundaries:

def test_window_boundary_is_inclusive():
    ae = pd.DataFrame({"USUBJID": ["001", "001"],
                       "AESTDTC": ["2026-07-15", "2026-07-16"]})  # +30, +31
    out = derive_trtemfl(ae, adsl, window=30)
    assert list(out["TRTEMFL"]) == ["Y", "N"]


def test_complete_date_before_treatment_is_not_dragged_forward():
    ae = pd.DataFrame({"USUBJID": ["001"], "AESTDTC": ["2026-01-01"]})
    out = derive_trtemfl(ae, adsl)
    assert out["ASTDT"].iloc[0] == pd.Timestamp("2026-01-01")
    assert out["TRTEMFL"].iloc[0] == "N"

Recap

Read source as text; type conversion is a derivation, not a parser guess
Raise on unmapped controlled terminology — never let it default to missing
validate= on every merge; --SEQ from a deterministic sort
SDTM keeps partial dates; ADaM imputes them with a flag
TRTDURD is inclusive of both endpoints
Check byte widths before writing XPT — truncation is silent
Verify the round trip once per study
Python can do all of this; it cannot give you admiral’s tested edge cases

Next: Statistical analysis.