"""Build the CD4 TCR atlas from public sources only (sources/). One parser per source, one schema. Never add unpublished sequences here (see README). Usage: python -I scripts/build_atlas.py (from the atlas folder, with the analysis venv) Writes atlas/cd4_tcr_atlas_records.csv.gz (one row per published record), atlas/cd4_tcr_atlas_clonotypes.csv.gz (one row per unique species + CDR3alpha + CDR3beta) and atlas/summary.csv. The records table is also written as parquet.""" import importlib.util import re import sys from pathlib import Path import pandas as pd ROOT = Path(__file__).resolve().parent.parent SRC, OUT = ROOT / "sources", ROOT / "atlas" COLS = ["source", "study", "species", "cd4_evidence", "restriction", "TRAV", "TRAJ", "CDR3a", "TRBV", "TRBJ", "CDR3b", "epitope", "antigen", "organism", "context", "annotation_level", "specificity_evidence", "donor", "cells"] AA = re.compile(r"^[ACDEFGHIKLMNPQRSTVWY]+$") def cdr3(s): """Upper-case amino acid CDR3 or None.""" if not isinstance(s, str): return None s = s.strip().upper() return s if 4 <= len(s) <= 40 and AA.match(s) else None def core(s): """CDR3 without the conserved C and F/W anchors, so junction and CDR3 conventions agree.""" if not isinstance(s, str) or len(s) < 6: return None return (s[1:] if s[0] == "C" else s)[: -1 if s[-1] in "FW" else None] def gene(s): return None if not isinstance(s, str) or not s.strip() else s.split("*")[0].split(",")[0].strip() def level(d): """epitope > antigen > condition > none.""" e, a, c = (d[k].notna() & d[k].astype(str).str.strip().ne("") for k in ("epitope", "antigen", "context")) return e.map({True: "epitope"}).fillna(a.map({True: "antigen"})).fillna(c.map({True: "condition"})).fillna("none") CLASS2 = re.compile(r"HLA-D[RPQ]|class II|H-?2-?I[AE]|\bI-?[AE][a-z0-9]\b|DRB|DQ[AB]|DPB", re.I) # ---------------------------------------------------------------- annotated databases def vdjdb(): v = pd.read_csv(SRC / "_databases/vdjdb/vdjdb_full.txt", sep="\t", low_memory=False) v = v[v["mhc.class"] == "MHCII"] return pd.DataFrame({"source": "VDJdb", "study": v["reference.id"], "species": v.species.map({"HomoSapiens": "human", "MusMusculus": "mouse"}).fillna(v.species), "cd4_evidence": "MHC class II restriction", "restriction": v["mhc.a"].astype(str) + "/" + v["mhc.b"].astype(str), "TRAV": v["v.alpha"], "TRAJ": v["j.alpha"], "CDR3a": v["cdr3.alpha"], "TRBV": v["v.beta"], "TRBJ": v["j.beta"], "CDR3b": v["cdr3.beta"], "epitope": v["antigen.epitope"], "antigen": v["antigen.gene"], "organism": v["antigen.species"], "context": None, "donor": v["meta.subject.id"], "cells": None}) def iedb(): i = pd.read_csv(SRC / "_databases/iedb/tcr_full_v3.csv", header=[0, 1], low_memory=False) i = i[(i[("Receptor", "Type")] == "alphabeta") & i[("Assay", "MHC Allele Names")].astype(str).str.contains(CLASS2)] assert set(i[("Chain 1", "Type")].dropna()) <= {"alpha"} and set(i[("Chain 2", "Type")].dropna()) <= {"beta"} c = lambda ch, f: i[(ch, f)] cd = lambda ch: c(ch, "Junction Calculated").fillna(c(ch, "CDR3 Curated")) g = lambda ch, x: c(ch, f"Curated {x} Gene").fillna(c(ch, f"Calculated {x} Gene")) org = c("Chain 2", "Organism IRI").fillna(c("Chain 1", "Organism IRI")).astype(str).str.extract(r"(\d+)$")[0] return pd.DataFrame({"source": "IEDB", "study": i[("Reference", "IEDB IRI")], "species": org.map({"9606": "human", "10090": "mouse"}).fillna("taxon:" + org.astype(str)), "cd4_evidence": "MHC class II restriction", "restriction": i[("Assay", "MHC Allele Names")], "TRAV": g("Chain 1", "V"), "TRAJ": g("Chain 1", "J"), "CDR3a": cd("Chain 1"), "TRBV": g("Chain 2", "V"), "TRBJ": g("Chain 2", "J"), "CDR3b": cd("Chain 2"), "epitope": i[("Epitope", "Name")], "antigen": i[("Epitope", "Source Molecule")], "organism": i[("Epitope", "Source Organism")], "context": None, "donor": None, "cells": None}) def mcpas(): m = pd.read_csv(SRC / "_databases/mcpas/mcpas.csv", encoding="utf-8-sig", low_memory=False) cd4 = m["T.Cell.Type"].astype(str).str.strip().isin(["CD4", "FOXP3"]) m = m[cd4 | (m["T.Cell.Type"].isna() & m.MHC.astype(str).str.contains(CLASS2))] return pd.DataFrame({"source": "McPAS-TCR", "study": "PMID:" + m["PubMed.ID"].astype(str), "species": m.Species.str.lower(), "cd4_evidence": m["T.Cell.Type"].map({"CD4": "annotated CD4", "FOXP3": "annotated FOXP3"}).fillna("MHC class II restriction"), "restriction": m.MHC, "TRAV": m.TRAV, "TRAJ": m.TRAJ, "CDR3a": m["CDR3.alpha.aa"], "TRBV": m.TRBV, "TRBJ": m.TRBJ, "CDR3b": m["CDR3.beta.aa"], "epitope": m["Epitope.peptide"], "antigen": m["Antigen.protein"], "organism": None, "context": m.Pathology, "donor": None, "cells": None}) # ---------------------------------------------------------------- published studies def pair_contigs(c, keys): """10x / scRepertoire contig rows -> one row per cell (keys identify the cell), top-UMI TRA and TRB.""" c = c[c.chain.isin(["TRA", "TRB"]) & c.cdr3.map(cdr3).notna()].sort_values("umis", ascending=False).drop_duplicates(keys + ["chain"]) a, b = (c[c.chain == ch].set_index(keys)[["v_gene", "j_gene", "cdr3"]] for ch in ("TRA", "TRB")) d = a.join(b, how="outer", lsuffix="_a", rsuffix="_b").reset_index() return d.rename(columns={"v_gene_a": "TRAV", "j_gene_a": "TRAJ", "cdr3_a": "CDR3a", "v_gene_b": "TRBV", "j_gene_b": "TRBJ", "cdr3_b": "CDR3b"}) def clones(d, by, **const): """Cells -> one record per (by + TCR), with the number of cells.""" tcr = ["TRAV", "TRAJ", "CDR3a", "TRBV", "TRBJ", "CDR3b"] for k in tcr: if k not in d: d[k] = None d = d.fillna({k: "" for k in by + tcr}).groupby(by + tcr).size().rename("cells").reset_index().replace("", None) return d.assign(**const) def lu2024(): """eLife 2024, doi 10.7554/eLife.89999: Tables 4 and 6, clonotypes reconstituted in reporter cells, epitope and HLA mapped.""" out = [] for n, ep in ((4, "S epitope*"), (6, "S epitope")): t = pd.read_csv(SRC / f"Lu2024/derived/elife-89999-table{n}.csv") fix = lambda s, p: p + s.astype(str).str.replace("–", "-").str.strip() out.append(pd.DataFrame({"TRAV": fix(t.TRAV, "TRAV"), "TRAJ": fix(t.TRAJ, "TRAJ"), "CDR3a": t["CDR3α"], "TRBV": fix(t.TRBV, "TRBV"), "TRBJ": fix(t.TRBJ, "TRBJ"), "CDR3b": t["CDR3β"], "epitope": t[ep].str.extract(r"^([A-Z]+)")[0], "restriction": t["Restricting HLA"], "donor": t.Donor, "context": "BNT162b2 vaccinee, " + ("post-vaccination expanded" if n == 4 else "pre-vaccination cross-reactive")})) return pd.concat(out).assign(source="Lu 2024 (eLife)", study="doi:10.7554/eLife.89999", species="human", cd4_evidence="MHC class II restriction", antigen="Spike", organism="SARS-CoV-2", specificity_evidence="TCR reconstitution, reporter assay", annotation_level="epitope") def lu2021(): """J Exp Med 2021, doi 10.1084/jem.20211327: Table S3 (cTfh cells by stimulation) and Table S4 (S864-882 public clonotypes).""" s3 = pd.read_excel(SRC / "Lu2021/JEM_20211327_TableS3.xlsx", sheet_name="Table S3") s3 = s3.rename(columns={"CDR3.alpha": "CDR3a", "CDR3.beta": "CDR3b", "Patient": "donor", "Stimulation": "antigen"}) a = clones(s3, ["donor", "antigen"], context="circulating Tfh, COVID-19 convalescent or healthy donor", epitope=None, restriction=None, cd4_evidence="circulating Tfh cluster (single cell)", specificity_evidence="expanded after in vitro stimulation", annotation_level="antigen") s4 = pd.read_excel(SRC / "Lu2021/JEM_20211327_TableS4.xlsx", sheet_name="Table S4", header=None, skiprows=3).iloc[:, :7] s4.columns = ["donor", "TRAV", "CDR3a", "TRAJ", "TRBV", "CDR3b", "TRBJ"] s4 = s4[s4.CDR3b.map(cdr3).notna()].assign(donor=lambda d: d.donor.ffill()) for k, p in (("TRAV", "TRAV"), ("TRAJ", "TRAJ"), ("TRBV", "TRBV"), ("TRBJ", "TRBJ")): s4[k] = p + s4[k].astype(str).str.strip() b = s4.assign(antigen="Spike", epitope="S864-882 (residue range, sequence not in the table)", context="public circulating Tfh clonotype", cells=None, restriction=None, cd4_evidence="circulating Tfh cluster (single cell)", specificity_evidence="expanded after peptide stimulation", annotation_level="epitope") return pd.concat([a, b]).assign(source="Lu 2021 (J Exp Med)", study="doi:10.1084/jem.20211327", species="human", organism="SARS-CoV-2") def sakai2025(): """J Clin Invest 2025, doi 10.1172/JCI185443: clone Y-50, the one clonotype of Supplemental Table 1 confirmed TMM/CD1b-reactive. The 51 other candidates are not loaded: antigen not confirmed, lineage not given, several carry the MAIT TRAV1-2/TRAJ33 chain.""" return pd.DataFrame([{"source": "Sakai 2025 (J Clin Invest)", "study": "doi:10.1172/JCI185443", "species": "human", "cd4_evidence": "CD4+ T cell clone (paper)", "restriction": "CD1b", "TRAV": "TRAV14/DV4", "TRAJ": "TRAJ5", "CDR3a": "CAMRRNTGRRALTF", "TRBV": "TRBV4-1", "TRBJ": "TRBJ2-3", "CDR3b": "CASSHLGLAGGIDGTDTQYF", "epitope": "trehalose monomycolate (lipid)", "antigen": "trehalose monomycolate", "organism": "Mycobacterium tuberculosis", "context": "clone Y-50", "specificity_evidence": "TCR reconstitution, reporter assay; crystal structure", "annotation_level": "epitope"}]) def morita2023(): """iScience 2023, doi 10.1016/j.isci.2023.106685: CD4+ clone-111 (Table S9).""" return pd.DataFrame([{"source": "Morita 2023 (iScience)", "study": "doi:10.1016/j.isci.2023.106685", "species": "human", "cd4_evidence": "CD4+ T cell clone (paper)", "TRAV": "TRAV13-2", "TRAJ": "TRAJ4", "CDR3a": "CAENPRFSGGYNKLIF", "TRBV": "TRBV6-1", "TRBJ": "TRBJ2-3", "CDR3b": "CASSGTGVDTQYF", "antigen": "SARS-CoV-2 peptide (epitope in the paper, Figures 4 and 5, not in Table S9)", "organism": "SARS-CoV-2", "context": "clone-111, agammaglobulinemia patient", "specificity_evidence": "peptide-expanded clone", "annotation_level": "antigen"}]) def immgent(): """ImmGenT (GSE297097), mouse: tetramer-positive CD4 and Treg cells. IGT52 (GSE347941) is the Der p 1 model of the Pepper lab (Bangs 2025 preprint, doi 10.1101/2025.10.03.680350; link inferred from the GEO record). I-Ab is stated for the Der p 1 tetramer only.""" b = pd.read_csv(SRC / "Bangs2025/GSE297097_All.IGTs.Enhanced.Summary.Table.2026-02-12.csv.gz", low_memory=False, usecols=["IGT.ID", "organ", "sample.name", "IGT.cellID", "alpha.functionality", "alpha.vgene", "alpha.jgene", "alpha.junction", "beta.functionality", "beta.vgene", "beta.jgene", "beta.junction", "level1"]) a = pd.read_csv(SRC / "Bangs2025/GSE297097_antigen_specific_cells.tsv.gz", sep="\t") ag = {"Derp1Tetp": ("Der p 1", "Dermatophagoides pteronyssinus", "H2-IAb"), "2WTetp": ("2W peptide (tetramer '2WTetp')", None, None), "AS15Tetp": ("AS15 (tetramer 'AS15Tetp')", None, None)} t = b.merge(a[a.Ag_spe.isin(ag)], left_on="IGT.cellID", right_on="cellID") t = t[t.level1.isin(["CD4", "Treg"])] for ch, s in (("alpha", "a"), ("beta", "b")): ok = t[f"{ch}.functionality"].eq("productive") t["TR%sV" % s.upper()], t["TR%sJ" % s.upper()], t["CDR3" + s] = t[f"{ch}.vgene"].where(ok), t[f"{ch}.jgene"].where(ok), t[f"{ch}.junction"].where(ok) t["antigen"], t["organism"], t["restriction"] = (t.Ag_spe.map(lambda k: ag[k][j]) for j in range(3)) t["donor"] = t["IGT.ID"] + " " + t["sample.name"] t["context"] = t["IGT.ID"].map({"IGT52": "Der p 1 intranasal allergy model (GSE347941, Pepper lab)", "IGT81": "Der p 1 tetramer-positive cells, run IGT81"}).fillna("ImmGenT run " + t["IGT.ID"]) + ", " + t.organ + ", " + t.level1 d = clones(t, ["donor", "context", "antigen", "organism", "restriction"]) return d.assign(source="ImmGenT (GSE297097)", study="GEO:GSE297097", species="mouse", cd4_evidence="CD4 or Treg cluster (single cell)", specificity_evidence="peptide:MHC tetramer", annotation_level="antigen") def han2025(): """GSE288452 (PMID 41188408), peanut oral immunotherapy trial POISED, BD Rhapsody: sorted CD4 T cells after 18 h peanut stimulation. Cells of the 'prcd4' table are peanut-reactive (CD154+ and/or CD137+); the other cells of the 'cd4' table are loaded without antigen.""" cd4 = pd.read_csv(SRC / "Han2025/GSE288452_POISED_all_samples_TCRab_cd4_scRepertoire.csv.gz", low_memory=False) pr = set(pd.read_csv(SRC / "Han2025/GSE288452_POISED_all_samples_TCRab_prcd4_scRepertoire.csv.gz", usecols=["barcode"]).barcode) meta = cd4.drop_duplicates("barcode").set_index("barcode")[["PPID", "Group2"]] d = pair_contigs(cd4, ["barcode"]).join(meta, on="barcode") d["donor"], d["pr"] = d.PPID, d.barcode.isin(pr) d["context"] = "peanut allergy trial, arm " + d.Group2.astype(str) d = clones(d, ["donor", "context", "pr"]) d["cdr3a_core"], d["cdr3b_core"] = d.CDR3a, d.CDR3b # deposited without the C and F/W anchors d["antigen"] = d.pr.map({True: "peanut extract"}) d["specificity_evidence"] = d.pr.map({True: "CD154+ and/or CD137+ after 18 h stimulation", False: "none (CD4 cell outside the peanut-reactive table)"}) d["annotation_level"] = d.pr.map({True: "antigen", False: "none"}) return d.assign(source="Han 2025 (GSE288452)", study="GEO:GSE288452", species="human", cd4_evidence="sorted CD4+") def gazzinelli2024(): """GSE240865 (PMID 38587077): sorted memory CD4 subsets (CRTH2 / CCR6), filarial-infected and control subjects. No antigen.""" out = [] for f in sorted((SRC / "GazzinelliGuimaraes2024").glob("*contig_annotations.csv.gz")): lab = re.search(r"Tcells_(.+?)_(control|filarial)_", f.name) c = pd.read_csv(f, low_memory=False) c = c[c.is_cell.astype(str).str.lower().eq("true") & c.high_confidence.astype(str).str.lower().eq("true") & c.productive.astype(str).str.lower().eq("true")] d = pair_contigs(c, ["barcode"]).assign(donor=f"pool {lab.group(2)} {lab.group(1)}", context=f"memory CD4 subset {lab.group(1)}, {lab.group(2)} subjects") out.append(clones(d, ["donor", "context"])) return pd.concat(out).assign(source="Gazzinelli-Guimaraes 2024 (GSE240865)", study="GEO:GSE240865", species="human", cd4_evidence="sorted memory CD4+", specificity_evidence="none", annotation_level="none") def lewis2023(): """GSE231590 (PMID 37604312): CD4 memory T cells of cow milk allergic and non-allergic children, with the deposited ARTE flag (column M111a). No V or J genes in the deposited metadata.""" m = pd.read_csv(SRC / "Lewis2023/GSE231590_milk_allergy_1_metadata.txt.gz", sep="\t", low_memory=False) m = m[m.t_cdr3s_aa.notna()] m["CDR3a"], m["CDR3b"] = m.t_cdr3s_aa.str.extract(r"TRA:([A-Z]+)")[0], m.t_cdr3s_aa.str.extract(r"TRB:([A-Z]+)")[0] m["donor"], m["pos"] = m.MULTI_ID.astype(str), m.M111a.eq("ARTEpos") m["context"] = "cow milk allergy study, group " + m.Group.map({"zAllergic": "allergic", "NonA": "non-allergic", "yBakedTol": "baked milk tolerant"}).fillna(m.Group) d = clones(m, ["donor", "context", "pos"]) d["antigen"] = d.pos.map({True: "cow milk peptide pool"}) d["specificity_evidence"] = d.pos.map({True: "flagged ARTEpos in the deposited metadata (antigen-reactive T cell enrichment)", False: "none (flagged ARTEneg)"}) d["annotation_level"] = d.pos.map({True: "antigen", False: "none"}) return d.assign(source="Lewis 2023 (GSE231590)", study="GEO:GSE231590", species="human", cd4_evidence="CD4 memory T cells (GEO design)") def wedderburn1993(): """PNAS 1993, doi 10.1073/pnas.90.17.8214: house dust mite specific CD4 clones of one atopic donor. Junctions rebuilt from the segments transcribed from Fig. 1 (last C of the V end + N region + J start up to the F/W of the FGXG motif); V and J names are the 1993 nomenclature, not IMGT. Clone DH15 alpha has no C in the printed V end, so its alpha chain is left empty.""" j = pd.read_csv(SRC / "Wedderburn1993/derived/wedderburn1993_fig1_junctions_aa.csv") t = pd.read_csv(SRC / "Wedderburn1993/derived/wedderburn1993_tables1_2.csv").set_index("clone") v = j.V_end_aa.str.extract(r"(C[A-Z]*)$")[0] j["junction"] = v + j.N_region_aa + j.J_start_aa.str.extract(r"^(.*?[FW])G.G")[0] w = j.pivot(index="clone", columns="chain", values="junction").join(t) same = t[t.identical_to.notna()].groupby("identical_to").size() pter = w.prolif_D_pteronyssinus.eq("+") return pd.DataFrame({"source": "Wedderburn 1993 (PNAS)", "study": "doi:10.1073/pnas.90.17.8214", "species": "human", "cd4_evidence": "CD4+ T cell clone (paper)", "restriction": "HLA-" + w.MHC_restriction.where(w.MHC_restriction != "ND"), "TRAV": "Va" + w.TRAV.astype(str) + " (1993 name)", "TRAJ": w.TRAJ.astype(str) + " (1993 name)", "CDR3a": w.TRA, "TRBV": "Vb" + w.TRBV.astype(str) + " (1993 name)", "TRBJ": "Jb" + w.TRBJ.astype(str) + " (1993 name)", "CDR3b": w.TRB, "antigen": pter.map({True: "house dust mite extract (D. farinae and D. pteronyssinus)", False: "house dust mite extract (D. farinae only)"}), "organism": "Dermatophagoides farinae", "context": "clone " + w.index + ", atopic donor", "specificity_evidence": "clone proliferation to mite extract", "annotation_level": "antigen", "donor": "one atopic donor", "cells": None}).assign(cells=lambda d: 1 + same.reindex(w.index).fillna(0).values) def morgan2021(): """Sci Immunol 2021, doi 10.1126/sciimmunol.abi5586 (eosinophilic oesophagitis): blood table only, sorted memory CD4 by CRTH2 and CD161. The tissue tables hold all T cells without a CD4 column and are not loaded. No antigen.""" d = pd.read_csv(SRC / "Morgan2021/pb_metadata.csv", low_memory=False).rename(columns={"TRB_CDR3": "CDR3b", "TRA_CDR3": "CDR3a", "Patient": "donor"}) d["context"] = "eosinophilic oesophagitis (" + d.Diagnosis.str.lower() + "), blood memory CD4 " + d.Fraction + ", " + d.phenotype return clones(d, ["donor", "context"]).assign(source="Morgan 2021 (Sci Immunol)", study="doi:10.1126/sciimmunol.abi5586", species="human", cd4_evidence="sorted memory CD4+", specificity_evidence="none", annotation_level="none") PARSERS = [vdjdb, iedb, mcpas, lu2024, lu2021, sakai2025, morita2023, immgent, han2025, gazzinelli2024, lewis2023, wedderburn1993, morgan2021] def extra_parsers(): """One file per study in scripts/parsers/, each with parse() returning the common schema. They import helpers with `from atlas import ...`.""" sys.modules["atlas"] = sys.modules[__name__] out = [] for f in sorted((ROOT / "scripts/parsers").glob("*.py")): spec = importlib.util.spec_from_file_location(f.stem, f) m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m) m.parse.__name__ = f.stem out.append(m.parse) return out def main(): parts = [] for p in PARSERS + extra_parsers(): d = p().reindex(columns=COLS + ["cdr3a_core", "cdr3b_core"]) d["CDR3a"], d["CDR3b"] = d.CDR3a.map(cdr3), d.CDR3b.map(cdr3) for k in ("TRAV", "TRAJ", "TRBV", "TRBJ"): d[k] = d[k].map(gene) d = d[d.CDR3a.notna() | d.CDR3b.notna()] d["annotation_level"] = d.annotation_level.fillna(level(d)) d["specificity_evidence"] = d.specificity_evidence.fillna("curated database record") d["cdr3a_core"], d["cdr3b_core"] = d.cdr3a_core.fillna(d.CDR3a.map(core)), d.cdr3b_core.fillna(d.CDR3b.map(core)) print(f"{p.__name__:28s} {len(d):8d} records") parts.append(d) R = pd.concat(parts, ignore_index=True).drop_duplicates() R["paired"] = R.CDR3a.notna() & R.CDR3b.notna() OUT.mkdir(exist_ok=True) R.to_csv(OUT / "cd4_tcr_atlas_records.csv.gz", index=False) R.to_parquet(OUT / "cd4_tcr_atlas_records.parquet", index=False) keys = ["species", "cdr3a_core", "cdr3b_core"] K = R[keys].fillna("") g = K.groupby(keys, sort=False).ngroup() C = K.groupby(g).first().assign(records=g.value_counts().sort_index(), studies=R.groupby(g).study.nunique()) def joined(col): # distinct values of a column per clonotype, joined t = pd.DataFrame({"g": g, "v": R[col].astype("string").str.strip()}).dropna() t = t[t.v != ""].drop_duplicates().sort_values(["g", "v"]) return t.groupby("g").v.agg("; ".join).str[:500] for name, col in [("sources", "source"), ("TRAV", "TRAV"), ("TRBV", "TRBV"), ("restriction", "restriction"), ("epitopes", "epitope"), ("antigens", "antigen"), ("organisms", "organism"), ("contexts", "context"), ("cd4_evidence", "cd4_evidence")]: C[name] = joined(col) order = ["epitope", "antigen", "condition", "none"] C["best_annotation"] = R.annotation_level.map(order.index).groupby(g).min().map(dict(enumerate(order))) C.to_csv(OUT / "cd4_tcr_atlas_clonotypes.csv.gz", index=False) S = R.groupby(["source", "species"], sort=False).agg(records=("source", "size"), studies=("study", "nunique"), unique_CDR3b=("cdr3b_core", "nunique"), unique_CDR3a=("cdr3a_core", "nunique"), paired_records=("paired", "sum"), epitopes=("epitope", "nunique"), with_antigen=("annotation_level", lambda s: int(s.isin(["epitope", "antigen"]).sum()))).reset_index() S.to_csv(OUT / "summary.csv", index=False) pd.set_option("display.width", 220); pd.set_option("display.max_columns", 20) print(S.to_string(index=False)) print(f"\ntotal {len(R)} records, {len(C)} unique clonotypes (species + CDR3a + CDR3b), {R.cdr3b_core.nunique()} unique CDR3beta, {int(R.paired.sum())} paired records") print(R.groupby(["species", "annotation_level"]).size().unstack(fill_value=0)) if __name__ == "__main__": main()