implement data ingestion: CCLF loader, BCDA loader, seed loader, unified staging

- CCLF: fixed-width parser using IP-derived field positions (cclf_layout.py),
  ZIP extraction, file discovery by CMS naming convention, DuckDB loading,
  optional pipeline execution to produce input_layer tables
- BCDA: flatten ndjson → Parquet, load into bcda schema in DuckDB
- Seeds: auto-discover CSV/Excel/Parquet files, load into reference_data schema
- Unified staging: orchestrate all three loaders with optional Iceberg promotion
- CLI: stack load cclf/bcda/seed with full options (--path, --database, etc.)
- 57 tests covering parsers, loaders, CLI commands, and staging pipeline

fixes #5 fixes #6 fixes #7 fixes #8

src/aco/load/bcda.py

@@ -0,0 +1,108 @@
+"""Load BCDA FHIR exports into DuckDB.
+
+Flattens raw ndjson files (Patient, ExplanationOfBenefit, Coverage) into
+Parquet via the existing ``bcda.express.flatten`` module, then loads the
+resulting Parquet files into the ``bcda`` schema in DuckDB.
+
+Usage::
+
+    from aco.load.bcda import load_bcda
+
+    stats = load_bcda()
+    # {"bcda.patient": 50, "bcda.explanation_of_benefit": 2021, ...}
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def load_bcda(
+    ndjson_dir: str | Path | None = None,
+    database: str | None = None,
+    *,
+    skip_flatten: bool = False,
+) -> dict[str, int]:
+    """Flatten BCDA ndjson and load into DuckDB.
+
+    Parameters
+    ----------
+    ndjson_dir : str or Path, optional
+        Directory containing raw ndjson files.  Defaults to the latest
+        export directory under ``conf.path("storage.bcda")``.
+    database : str, optional
+        DuckDB database path.  Defaults to ``conf.path("db.aco")``.
+    skip_flatten : bool
+        If True, skip the flatten step and load existing Parquet files.
+
+    Returns
+    -------
+    dict[str, int]
+        Row counts per table loaded.
+    """
+    import duckdb
+
+    from conf import path
+
+    db_path = database or str(path("db.aco"))
+    store_path = path("storage.bcda")
+
+    if ndjson_dir is None:
+        ndjson_dir = _find_latest_export(store_path)
+
+    if not skip_flatten:
+        from bcda.express.flatten import flatten_export
+
+        flatten_export(ndjson_dir, store_path)
+
+    # Load Parquet files into DuckDB
+    flat_dir = store_path / "flat"
+    if not flat_dir.exists():
+        raise FileNotFoundError(f"No flattened data at {flat_dir}")
+
+    con = duckdb.connect(str(db_path), read_only=False)
+    con.execute('CREATE SCHEMA IF NOT EXISTS "bcda"')
+
+    stats: dict[str, int] = {}
+    parquet_map = {
+        "patient": "bcda.patient",
+        "explanation_of_benefit": "bcda.explanation_of_benefit",
+        "eob_item": "bcda.eob_item",
+        "coverage": "bcda.coverage",
+    }
+
+    for file_stem, table_ref in parquet_map.items():
+        pq_path = flat_dir / f"{file_stem}.parquet"
+        if not pq_path.exists():
+            continue
+
+        schema, table = table_ref.split(".")
+        qualified = f'"{schema}"."{table}"'
+        con.execute(f"DROP TABLE IF EXISTS {qualified}")
+        con.execute(
+            f"CREATE TABLE {qualified} AS SELECT * FROM read_parquet('{pq_path}')"
+        )
+        count = con.execute(f"SELECT COUNT(*) FROM {qualified}").fetchone()[0]
+        stats[table_ref] = count
+
+    con.close()
+    return stats
+
+
+def _find_latest_export(store_path: Path) -> Path:
+    """Find the most recent export directory under the BCDA store."""
+    exports_dir = store_path / "exports"
+    if not exports_dir.exists():
+        raise FileNotFoundError(
+            f"No exports directory at {exports_dir}. "
+            "Run a BCDA export first or specify --ndjson-dir."
+        )
+    # Export directories are named by job ID; pick the most recent
+    dirs = sorted(
+        [d for d in exports_dir.iterdir() if d.is_dir()],
+        key=lambda d: d.stat().st_mtime,
+        reverse=True,
+    )
+    if not dirs:
+        raise FileNotFoundError(f"No export directories in {exports_dir}")
+    return dirs[0]

src/aco/load/cclf.py

@@ -0,0 +1,255 @@
+"""Load CCLF fixed-width files into DuckDB and run the CCLF pipeline.
+
+Reads CMS CCLF files (delivered as fixed-width text inside ZIPs),
+parses them using the IP-defined field positions, loads into the
+``cclf`` schema in DuckDB, then runs the 19-step CCLF pipeline
+to produce ``input_layer.medical_claim``, ``input_layer.pharmacy_claim``,
+and ``input_layer.eligibility``.
+
+Usage::
+
+    from aco.load.cclf import load_cclf_directory
+
+    stats = load_cclf_directory(Path("data/cclf"))
+    # stats = {"cclf1": 45000, "cclf5": 120000, ...}
+"""
+
+from __future__ import annotations
+
+import zipfile
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+import polars as pl
+
+from aco.table.cclf_filenames import classify
+from aco.table.cclf_layout import LAYOUTS
+
+
+def _parse_value(raw: str, fmt: str) -> Any:
+    """Convert a raw fixed-width string to a typed Python value.
+
+    Parameters
+    ----------
+    raw : str
+        Stripped string extracted from the fixed-width line.
+    fmt : str
+        CCLF IP format code (e.g. ``X(11)``, ``9(13)``, ``YYYY-MM-DD``).
+    """
+    if not raw or raw.isspace():
+        return None
+
+    if "YYYY" in fmt or "MM-DD" in fmt:
+        try:
+            return date.fromisoformat(raw)
+        except ValueError:
+            return None
+
+    if "V9" in fmt or ".99" in fmt or ".9999" in fmt:
+        try:
+            return float(raw)
+        except ValueError:
+            return None
+
+    return raw
+
+
+def parse_cclf_file(
+    lines: list[str],
+    cclf_table: str,
+) -> pl.DataFrame:
+    """Parse fixed-width CCLF lines into a polars DataFrame.
+
+    Parameters
+    ----------
+    lines : list[str]
+        Raw text lines from a single CCLF file.
+    cclf_table : str
+        Table identifier (e.g. ``cclf1``, ``cclf8``).
+
+    Returns
+    -------
+    pl.DataFrame
+        Parsed DataFrame with columns matching the CCLF IP layout.
+    """
+    layout = LAYOUTS[cclf_table]
+    records: list[dict[str, Any]] = []
+
+    for line in lines:
+        if not line.strip():
+            continue
+        row: dict[str, Any] = {}
+        for field_name, start, end, fmt in layout:
+            if field_name in ("blank", "delimiter", "filler"):
+                continue
+            # Convert 1-based inclusive positions to 0-based Python slice
+            raw = line[start - 1 : end].strip()
+            row[field_name] = _parse_value(raw, fmt)
+        records.append(row)
+
+    if not records:
+        # Return empty DataFrame with correct schema
+        schema = {
+            name: _polars_type(fmt)
+            for name, _, _, fmt in layout
+            if name not in ("blank", "delimiter", "filler")
+        }
+        return pl.DataFrame(schema=schema)
+
+    return pl.DataFrame(records)
+
+
+def _polars_type(fmt: str) -> type:
+    """Map CCLF format to polars-compatible Python type for empty schemas."""
+    if "YYYY" in fmt or "MM-DD" in fmt:
+        return date
+    if "V9" in fmt or ".99" in fmt or ".9999" in fmt:
+        return float
+    return str
+
+
+def _extract_zip(zip_path: Path, dest_dir: Path) -> list[Path]:
+    """Extract CCLF files from a ZIP archive.
+
+    Returns paths to extracted files that match CCLF naming conventions.
+    """
+    extracted = []
+    with zipfile.ZipFile(zip_path) as zf:
+        for name in zf.namelist():
+            info = classify(name)
+            if info and not info.is_zip and info.cclf_table:
+                target = dest_dir / Path(name).name
+                with zf.open(name) as src, open(target, "wb") as dst:
+                    dst.write(src.read())
+                extracted.append(target)
+    return extracted
+
+
+def discover_cclf_files(directory: Path) -> dict[str, list[Path]]:
+    """Find all CCLF files in a directory, optionally extracting ZIPs.
+
+    Returns a dict mapping cclf_table (e.g. ``cclf1``) to a list of
+    file paths.  ZIP files are extracted to a temporary subdirectory.
+    """
+    result: dict[str, list[Path]] = {}
+    extracted_dir = directory / ".extracted"
+
+    for p in sorted(directory.iterdir()):
+        if p.name.startswith("."):
+            continue
+
+        # Handle ZIP files
+        if p.suffix.lower() == ".zip" or (p.suffix == "" and zipfile.is_zipfile(p)):
+            info = classify(p.name)
+            if info and info.is_zip:
+                extracted_dir.mkdir(exist_ok=True)
+                for extracted in _extract_zip(p, extracted_dir):
+                    einfo = classify(extracted.name)
+                    if einfo and einfo.cclf_table:
+                        result.setdefault(einfo.cclf_table, []).append(extracted)
+                continue
+
+        # Handle plain CCLF files
+        info = classify(p.name)
+        if info and not info.is_zip and info.cclf_table:
+            result.setdefault(info.cclf_table, []).append(p)
+
+    return result
+
+
+def load_cclf_directory(
+    directory: Path,
+    database: str | None = None,
+    *,
+    run_pipeline: bool = True,
+) -> dict[str, int]:
+    """Load all CCLF files from a directory into DuckDB.
+
+    Parameters
+    ----------
+    directory : Path
+        Directory containing CCLF files (plain or ZIP).
+    database : str, optional
+        DuckDB database path.  Defaults to ``conf.path("db.aco")``.
+    run_pipeline : bool
+        If True (default), run the CCLF pipeline after loading raw
+        tables to produce input_layer outputs.
+
+    Returns
+    -------
+    dict[str, int]
+        Row counts per CCLF table loaded.
+    """
+    import duckdb
+
+    from conf import path
+
+    db_path = database or str(path("db.aco"))
+
+    files = discover_cclf_files(directory)
+    if not files:
+        raise FileNotFoundError(f"No CCLF files found in {directory}")
+
+    con = duckdb.connect(db_path, read_only=False)
+    con.execute('CREATE SCHEMA IF NOT EXISTS "cclf"')
+
+    stats: dict[str, int] = {}
+
+    for cclf_table, paths in sorted(files.items()):
+        if cclf_table not in LAYOUTS:
+            continue
+
+        all_lines: list[str] = []
+        file_names: list[str] = []
+        for p in paths:
+            text = p.read_text(encoding="latin-1")
+            file_lines = text.splitlines()
+            all_lines.extend(file_lines)
+            file_names.append(p.name)
+
+        if not all_lines:
+            continue
+
+        df = parse_cclf_file(all_lines, cclf_table)
+        if df.is_empty():
+            continue
+
+        # Drop and recreate — each load is a fresh import of the monthly feed
+        tbl = f'"cclf"."{cclf_table}"'
+        con.execute(f"DROP TABLE IF EXISTS {tbl}")
+        con.execute(f"CREATE TABLE {tbl} AS SELECT * FROM df")  # noqa: S608
+
+        stats[cclf_table] = len(df)
+
+    con.close()
+
+    if run_pipeline and stats:
+        _run_cclf_pipeline(db_path)
+
+    return stats
+
+
+def _run_cclf_pipeline(database: str) -> dict[str, int]:
+    """Execute the CCLF pipeline and save outputs to input_layer."""
+    from aco.lake.context import DuckDBContext
+    from aco.pipe.cclf import pipeline
+
+    ctx = DuckDBContext(database=database, read_only=False)
+    cache = pipeline.run(ctx.load)
+
+    # Save the three output tables
+    output_stats: dict[str, int] = {}
+    output_tables = [
+        "cclf.medical_claim",
+        "cclf.pharmacy_claim",
+        "cclf.eligibility",
+    ]
+    for table_ref in output_tables:
+        if table_ref in cache:
+            # Map cclf.X -> input_layer.X
+            target = table_ref.replace("cclf.", "input_layer.")
+            ctx.save(target, cache[table_ref], mode="replace")
+            output_stats[target] = len(cache[table_ref])
+
+    return output_stats

src/aco/load/seed.py

@@ -0,0 +1,114 @@
+"""Load seed/reference data into DuckDB.
+
+Seeds are CSV or Excel files in ``dev/seeds/`` that populate reference
+tables (FIPS codes, calendars, value sets, etc.).  This module
+discovers seed files, reads them with polars, and loads them into
+the ``reference_data`` or ``terminology`` schemas.
+
+Usage::
+
+    from aco.load.seed import load_seeds
+
+    stats = load_seeds()
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import polars as pl
+
+
+def _read_tabular(path: Path) -> pl.DataFrame:
+    """Read a CSV or Excel file into a polars DataFrame."""
+    suffix = path.suffix.lower()
+    if suffix == ".csv":
+        return pl.read_csv(path, infer_schema_length=10000)
+    if suffix in (".xlsx", ".xls"):
+        return pl.read_excel(path)
+    if suffix == ".parquet":
+        return pl.read_parquet(path)
+    raise ValueError(f"Unsupported file type: {suffix}")
+
+
+def _table_name(file_path: Path) -> str:
+    """Derive a table name from the file path stem.
+
+    Converts the filename to lowercase snake_case, stripping
+    common suffixes and parenthetical versions.
+    """
+    import re
+
+    name = file_path.stem
+    # Remove parenthetical bits like "(1)"
+    name = re.sub(r"\s*\(\d+\)\s*", "", name)
+    # Replace non-alphanumeric with underscores
+    name = re.sub(r"[^a-zA-Z0-9]+", "_", name)
+    # Collapse runs of underscores
+    name = re.sub(r"_+", "_", name).strip("_")
+    return name.lower()
+
+
+def load_seeds(
+    seed_dir: str | Path | None = None,
+    database: str | None = None,
+    *,
+    schema: str = "reference_data",
+) -> dict[str, int]:
+    """Load all seed files into DuckDB.
+
+    Parameters
+    ----------
+    seed_dir : str or Path, optional
+        Directory containing seed files.  Defaults to ``dev/seeds/``.
+    database : str, optional
+        DuckDB database path.  Defaults to ``conf.path("db.aco")``.
+    schema : str
+        Target DuckDB schema.  Defaults to ``reference_data``.
+
+    Returns
+    -------
+    dict[str, int]
+        Row counts per table loaded.
+    """
+    import duckdb
+
+    from conf import ROOT, path
+
+    db_path = database or str(path("db.aco"))
+    if seed_dir is None:
+        seed_dir = ROOT / "dev" / "seeds"
+    else:
+        seed_dir = Path(seed_dir)
+
+    if not seed_dir.exists():
+        raise FileNotFoundError(f"Seed directory not found: {seed_dir}")
+
+    con = duckdb.connect(db_path, read_only=False)
+    con.execute(f'CREATE SCHEMA IF NOT EXISTS "{schema}"')
+
+    stats: dict[str, int] = {}
+
+    for p in sorted(seed_dir.iterdir()):
+        if p.name.startswith(".") or p.is_dir():
+            continue
+        if p.suffix.lower() not in (".csv", ".xlsx", ".xls", ".parquet"):
+            continue
+
+        try:
+            df = _read_tabular(p)
+        except Exception:
+            continue
+
+        if df.is_empty():
+            continue
+
+        table = _table_name(p)
+        qualified = f'"{schema}"."{table}"'
+        con.execute(f"DROP TABLE IF EXISTS {qualified}")
+        con.execute(f"CREATE TABLE {qualified} AS SELECT * FROM df")
+
+        stats[f"{schema}.{table}"] = len(df)
+
+    con.close()
+    return stats

src/aco/load/stage.py

@@ -0,0 +1,120 @@
+"""Unified staging pipeline: raw files -> DuckDB -> (optional) Iceberg.
+
+Orchestrates the three loaders (CCLF, BCDA, seeds) in sequence,
+with optional promotion to Iceberg for lakehouse contexts.
+
+Usage::
+
+    from aco.load.stage import stage_all
+
+    stats = stage_all(cclf_dir=Path("data/cclf"))
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+log = logging.getLogger(__name__)
+
+
+def stage_all(
+    *,
+    cclf_dir: Path | None = None,
+    bcda_ndjson_dir: Path | None = None,
+    seed_dir: Path | None = None,
+    database: str | None = None,
+    run_pipelines: bool = True,
+    promote_to_iceberg: bool = False,
+) -> dict[str, dict[str, int]]:
+    """Run the full staging pipeline.
+
+    Parameters
+    ----------
+    cclf_dir : Path, optional
+        Directory containing CCLF files.  Skipped if None.
+    bcda_ndjson_dir : Path, optional
+        Directory containing BCDA ndjson files.  Skipped if None.
+    seed_dir : Path, optional
+        Directory containing seed files.  Skipped if None.
+    database : str, optional
+        DuckDB database path.  Defaults to config.
+    run_pipelines : bool
+        Run transformation pipelines after loading (default True).
+    promote_to_iceberg : bool
+        Push loaded data to Iceberg after DuckDB staging (default False).
+
+    Returns
+    -------
+    dict[str, dict[str, int]]
+        Nested dict: ``{"cclf": {...}, "bcda": {...}, "seed": {...}}``.
+    """
+    results: dict[str, dict[str, int]] = {}
+
+    if cclf_dir is not None:
+        from aco.load.cclf import load_cclf_directory
+
+        log.info("Staging CCLF from %s", cclf_dir)
+        results["cclf"] = load_cclf_directory(
+            cclf_dir,
+            database=database,
+            run_pipeline=run_pipelines,
+        )
+
+    if bcda_ndjson_dir is not None:
+        from aco.load.bcda import load_bcda
+
+        log.info("Staging BCDA from %s", bcda_ndjson_dir)
+        results["bcda"] = load_bcda(
+            ndjson_dir=bcda_ndjson_dir,
+            database=database,
+        )
+
+    if seed_dir is not None:
+        from aco.load.seed import load_seeds
+
+        log.info("Staging seeds from %s", seed_dir)
+        results["seed"] = load_seeds(
+            seed_dir=seed_dir,
+            database=database,
+        )
+
+    if promote_to_iceberg and results:
+        _promote_to_iceberg(results, database=database)
+
+    return results
+
+
+def _promote_to_iceberg(
+    stage_results: dict[str, dict[str, int]],
+    *,
+    database: str | None = None,
+) -> None:
+    """Push staged DuckDB tables to Iceberg via IcebergContext.
+
+    Reads each staged table from DuckDB and writes it to the
+    Iceberg catalog configured in stack.toml.
+    """
+    from aco.lake.context import DuckDBContext, IcebergContext
+    from conf import cfg, path
+
+    db_path = database or str(path("db.aco"))
+    src = DuckDBContext(database=db_path)
+
+    lake = cfg.lake
+    ice = IcebergContext(
+        catalog_uri=lake.nessie.catalog_uri,
+        warehouse=lake.warehouse,
+        properties={
+            "s3.endpoint": cfg.lake.get("s3_endpoint", "http://rustfs:9000"),
+            "s3.path-style-access": "true",
+        },
+    )
+
+    for source, tables in stage_results.items():
+        for table_ref in tables:
+            if "." not in table_ref:
+                continue
+            log.info("Promoting %s to Iceberg", table_ref)
+            df = src.load(table_ref)
+            ice.save(table_ref, df, mode="replace")

src/aco/table/cclf_layout.py

@@ -0,0 +1,333 @@
+"""CCLF fixed-width field positions for all 12 file types.
+
+Auto-generated from the CCLF Information Packet (Version 41.0, 07/16/2025).
+Positions are 1-based (matching the IP); the reader converts to 0-based slices.
+
+Each entry is ``(field_name, start, end, format)`` where start/end are
+inclusive 1-based byte positions from the CCLF IP appendix.
+"""
+
+from __future__ import annotations
+
+# {cclf_table: [(field_name, start_1based, end_1based, format), ...]}
+LAYOUTS: dict[str, list[tuple[str, int, int, str]]] = {
+    "cclf0": [
+        ("file_number_label", 1, 13, "X(13)"),
+        ("file_description_label", 15, 34, "X(20)"),
+        ("total_records_count_label", 36, 55, "X(20)"),
+        ("record_length_label", 57, 69, "X(13)"),
+        ("file_type", 1, 7, "X(7)"),
+        ("file_name", 9, 51, "X(43)"),
+        ("number_of_records", 53, 63, "X(11)"),
+        ("length_of_record", 65, 69, "X(5)"),
+    ],
+    "cclf1": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("prvdr_oscar_num", 14, 19, "X(06)"),
+        ("bene_mbi_id", 20, 30, "X(11)"),
+        ("bene_hic_num", 31, 41, "X(11)"),
+        ("clm_type_cd", 42, 43, "9(02)"),
+        ("clm_from_dt", 44, 53, "YYYY-MM-DD"),
+        ("clm_thru_dt", 54, 63, "YYYY-MM-DD"),
+        ("clm_bill_fac_type_cd", 64, 64, "X(01)"),
+        ("clm_bill_clsfctn_cd", 65, 65, "X(01)"),
+        ("prncpl_dgns_cd", 66, 72, "X(07)"),
+        ("admtg_dgns_cd", 73, 79, "X(07)"),
+        ("clm_mdcr_npmt_rsn_cd", 80, 81, "X(02)"),
+        ("clm_pmt_amt", 82, 98, "-9(13).99"),
+        ("clm_nch_prmry_pyr_cd", 99, 99, "X(01)"),
+        ("prvdr_fac_fips_st_cd", 100, 101, "X(02)"),
+        ("bene_ptnt_stus_cd", 102, 103, "X(02)"),
+        ("dgns_drg_cd", 104, 107, "X(04)"),
+        ("clm_op_srvc_type_cd", 108, 108, "X(01)"),
+        ("fac_prvdr_npi_num", 109, 118, "X(10)"),
+        ("oprtg_prvdr_npi_num", 119, 128, "X(10)"),
+        ("atndg_prvdr_npi_num", 129, 138, "X(10)"),
+        ("othr_prvdr_npi_num", 139, 148, "X(10)"),
+        ("clm_adjsmt_type_cd", 149, 150, "X(02)"),
+        ("clm_efctv_dt", 151, 160, "YYYY-MM-DD"),
+        ("clm_idr_ld_dt", 161, 170, "YYYY-MM-DD"),
+        ("bene_eqtbl_bic_hicn_num", 171, 181, "X(11)"),
+        ("clm_admsn_type_cd", 182, 183, "X(2)"),
+        ("clm_admsn_src_cd", 184, 185, "X(2)"),
+        ("clm_bill_freq_cd", 186, 186, "X(1)"),
+        ("clm_query_cd", 187, 187, "X(1)"),
+        ("dgns_prcdr_icd_ind", 188, 188, "X(1)"),
+        ("clm_mdcr_instnl_tot_chrg_amt", 189, 203, "-9(11).99"),
+        ("clm_mdcr_ip_pps_cptl_ime_amt", 204, 218, "-9(11).99"),
+        ("clm_oprtnl_ime_amt", 219, 240, "-9(18).99"),
+        ("clm_mdcr_ip_pps_dsprprtnt_amt", 241, 255, "-9(11).99"),
+        ("clm_hipps_uncompd_care_amt", 256, 270, "-9(11).99"),
+        ("clm_oprtnl_dsprprtnt_amt", 271, 292, "-9(18).99"),
+        ("clm_blg_prvdr_oscar_num", 293, 312, "x(20)"),
+        ("clm_blg_prvdr_npi_num", 313, 322, "x(10)"),
+        ("clm_oprtg_prvdr_npi_num", 323, 332, "x(10)"),
+        ("clm_atndg_prvdr_npi_num", 333, 342, "x(10)"),
+        ("clm_othr_prvdr_npi_num", 343, 352, "x(10)"),
+        ("clm_cntl_num", 353, 392, "x(40)"),
+        ("clm_org_cntl_num", 393, 432, "x(40)"),
+        ("clm_cntrctr_num", 433, 437, "x(5)"),
+    ],
+    "cclf2": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("clm_line_num", 14, 23, "9(10)"),
+        ("bene_mbi_id", 24, 34, "X(11)"),
+        ("bene_hic_num", 35, 45, "X(11)"),
+        ("clm_type_cd", 46, 47, "9(02)"),
+        ("clm_line_from_dt", 48, 57, "YYYY-MM-DD"),
+        ("clm_line_thru_dt", 58, 67, "YYYY-MM-DD"),
+        ("clm_line_prod_rev_ctr_cd", 68, 71, "X(04)"),
+        ("clm_line_instnl_rev_ctr_dt", 72, 81, "YYYY-MM-DD"),
+        ("clm_line_hcpcs_cd", 82, 86, "X(05)"),
+        ("bene_eqtbl_bic_hicn_num", 87, 97, "X(11)"),
+        ("prvdr_oscar_num", 98, 103, "X(6)"),
+        ("clm_from_dt", 104, 113, "YYYY-MM-DD"),
+        ("clm_thru_dt", 114, 123, "YYYY-MM-DD"),
+        ("clm_line_srvc_unit_qty", 124, 147, "-9(18).9999"),
+        ("clm_line_cvrd_pd_amt", 148, 164, "-9(13).99"),
+        ("hcpcs_1_mdfr_cd", 165, 166, "X(2)"),
+        ("hcpcs_2_mdfr_cd", 167, 168, "X(2)"),
+        ("hcpcs_3_mdfr_cd", 169, 170, "X(2)"),
+        ("hcpcs_4_mdfr_cd", 171, 172, "X(2)"),
+        ("hcpcs_5_mdfr_cd", 173, 174, "X(2)"),
+        ("clm_rev_apc_hipps_cd", 175, 179, "X(5)"),
+        ("clm_fac_prvdr_oscar_num", 180, 199, "X(20)"),
+    ],
+    "cclf3": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("bene_mbi_id", 14, 24, "X(11)"),
+        ("bene_hic_num", 25, 35, "X(11)"),
+        ("clm_type_cd", 36, 37, "9(02)"),
+        ("clm_val_sqnc_num", 38, 39, "9(2)"),
+        ("clm_prcdr_cd", 40, 46, "X(07)"),
+        ("clm_prcdr_prfrm_dt", 47, 56, "YYYY-MM-DD"),
+        ("bene_eqtbl_bic_hicn_num", 57, 67, "X(11)"),
+        ("prvdr_oscar_num", 68, 73, "X(6)"),
+        ("clm_from_dt", 74, 83, "YYYY-MM-DD"),
+        ("clm_thru_dt", 84, 93, "YYYY-MM-DD"),
+        ("dgns_prcdr_icd_ind", 94, 94, "X(1)"),
+        ("clm_blg_prvdr_oscar_num", 95, 114, "X(20)"),
+    ],
+    "cclf4": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("bene_mbi_id", 14, 24, "X(11)"),
+        ("bene_hic_num", 25, 35, "X(11)"),
+        ("clm_type_cd", 36, 37, "9(02)"),
+        ("clm_prod_type_cd", 38, 38, "X(01)"),
+        ("clm_val_sqnc_num", 39, 40, "9(2)"),
+        ("clm_dgns_cd", 41, 47, "X(07)"),
+        ("bene_eqtbl_bic_hicn_num", 48, 58, "X(11)"),
+        ("prvdr_oscar_num", 59, 64, "X(6)"),
+        ("clm_from_dt", 65, 74, "YYYY-MM-DD"),
+        ("clm_thru_dt", 75, 84, "YYYY-MM-DD"),
+        ("clm_poa_ind", 85, 91, "X(7)"),
+        ("dgns_prcdr_icd_ind", 92, 92, "X(1)"),
+        ("clm_blg_prvdr_oscar_num", 93, 112, "X(20)"),
+    ],
+    "cclf5": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("clm_line_num", 14, 23, "9(10)"),
+        ("bene_mbi_id", 24, 34, "X(11)"),
+        ("bene_hic_num", 35, 45, "X(11)"),
+        ("clm_type_cd", 46, 47, "9(02)"),
+        ("clm_from_dt", 48, 57, "YYYY-MM-DD"),
+        ("clm_thru_dt", 58, 67, "YYYY-MM-DD"),
+        ("rndrg_prvdr_type_cd", 68, 70, "X(03)"),
+        ("rndrg_prvdr_fips_st_cd", 71, 72, "X(02)"),
+        ("clm_prvdr_spclty_cd", 73, 74, "X(02)"),
+        ("clm_fed_type_srvc_cd", 75, 75, "X(01)"),
+        ("clm_pos_cd", 76, 77, "X(02)"),
+        ("clm_line_from_dt", 78, 87, "YYYY-MM-DD"),
+        ("clm_line_thru_dt", 88, 97, "YYYY-MM-DD"),
+        ("clm_line_hcpcs_cd", 98, 102, "X(05)"),
+        ("clm_line_cvrd_pd_amt", 103, 117, "X(15)"),
+        ("clm_line_prmry_pyr_cd", 118, 118, "X(01)"),
+        ("clm_line_dgns_cd", 119, 125, "X(07)"),
+        ("clm_rndrg_prvdr_tax_num", 126, 135, "X(10)"),
+        ("rndrg_prvdr_npi_num", 136, 145, "X(10)"),
+        ("clm_carr_pmt_dnl_cd", 146, 147, "X(02)"),
+        ("clm_prcsg_ind_cd", 148, 149, "X(02)"),
+        ("clm_adjsmt_type_cd", 150, 151, "X(02)"),
+        ("clm_efctv_dt", 152, 161, "YYYY-MM-DD"),
+        ("clm_idr_ld_dt", 162, 171, "YYYY-MM-DD"),
+        ("clm_cntl_num", 172, 211, "X(40)"),
+        ("bene_eqtbl_bic_hicn_num", 212, 222, "X(11)"),
+        ("clm_line_alowd_chrg_amt", 223, 239, "X(17)"),
+        ("clm_line_srvc_unit_qty", 240, 263, "-9(18).9999"),
+        ("hcpcs_1_mdfr_cd", 264, 265, "X(2)"),
+        ("hcpcs_2_mdfr_cd", 266, 267, "X(2)"),
+        ("hcpcs_3_mdfr_cd", 268, 269, "X(2)"),
+        ("hcpcs_4_mdfr_cd", 270, 271, "X(2)"),
+        ("hcpcs_5_mdfr_cd", 272, 273, "X(2)"),
+        ("clm_disp_cd", 274, 275, "X(2)"),
+        ("clm_dgns_1_cd", 276, 282, "X(7)"),
+        ("clm_dgns_2_cd", 283, 289, "X(7)"),
+        ("clm_dgns_3_cd", 290, 296, "X(7)"),
+        ("clm_dgns_4_cd", 297, 303, "X(7)"),
+        ("clm_dgns_5_cd", 304, 310, "X(7)"),
+        ("clm_dgns_6_cd", 311, 317, "X(7)"),
+        ("clm_dgns_7_cd", 318, 324, "X(7)"),
+        ("clm_dgns_8_cd", 325, 331, "X(7)"),
+        ("dgns_prcdr_icd_ind", 332, 332, "X(1)"),
+        ("clm_dgns_9_cd", 333, 339, "X(7)"),
+        ("clm_dgns_10_cd", 340, 346, "X(7)"),
+        ("clm_dgns_11_cd", 347, 353, "X(7)"),
+        ("clm_dgns_12_cd", 354, 360, "X(7)"),
+        ("hcpcs_betos_cd", 361, 363, "X(3)"),
+        ("clm_rndrg_prvdr_npi_num", 364, 373, "X(10)"),
+        ("clm_rfrg_prvdr_npi_num", 374, 383, "X(10)"),
+    ],
+    "cclf6": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("clm_line_num", 14, 23, "9(10)"),
+        ("bene_mbi_id", 24, 34, "X(11)"),
+        ("bene_hic_num", 35, 45, "X(11)"),
+        ("clm_type_cd", 46, 47, "9(02)"),
+        ("clm_from_dt", 48, 57, "YYYY-MM-DD"),
+        ("clm_thru_dt", 58, 67, "YYYY-MM-DD"),
+        ("clm_fed_type_srvc_cd", 68, 68, "X(01)"),
+        ("clm_pos_cd", 69, 70, "X(02)"),
+        ("clm_line_from_dt", 71, 80, "YYYY-MM-DD"),
+        ("clm_line_thru_dt", 81, 90, "YYYY-MM-DD"),
+        ("clm_line_hcpcs_cd", 91, 95, "X(05)"),
+        ("clm_line_cvrd_pd_amt", 96, 110, "-9(11).99"),
+        ("clm_prmry_pyr_cd", 111, 111, "X(01)"),
+        ("payto_prvdr_npi_num", 112, 121, "X(10)"),
+        ("ordrg_prvdr_npi_num", 122, 131, "X(10)"),
+        ("clm_carr_pmt_dnl_cd", 132, 133, "X(02)"),
+        ("clm_prcsg_ind_cd", 134, 135, "X(02)"),
+        ("clm_adjsmt_type_cd", 136, 137, "X(02)"),
+        ("clm_efctv_dt", 138, 147, "YYYY-MM-DD"),
+        ("clm_idr_ld_dt", 148, 157, "YYYY-MM-DD"),
+        ("clm_cntl_num", 158, 197, "X(40)"),
+        ("bene_eqtbl_bic_hicn_num", 198, 208, "X(11)"),
+        ("clm_line_alowd_chrg_amt", 209, 225, "-9(14).99"),
+        ("clm_disp_cd", 226, 227, "X(2)"),
+        ("clm_blg_prvdr_npi_num", 228, 237, "X(10)"),
+        ("clm_rfrg_prvdr_npi_num", 238, 247, "X(10)"),
+        ("clm_cntrctr_num", 384, 388, "X(5)"),
+    ],
+    "cclf7": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("bene_mbi_id", 14, 24, "X(11)"),
+        ("bene_hic_num", 25, 35, "X(11)"),
+        ("clm_line_ndc_cd", 36, 46, "X(11)"),
+        ("clm_type_cd", 47, 48, "9(02)"),
+        ("clm_line_from_dt", 49, 58, "YYYY-MM-DD"),
+        ("prvdr_srvc_id_qlfyr_cd", 59, 60, "X(02)"),
+        ("clm_srvc_prvdr_gnrc_id_num", 61, 80, "X(20)"),
+        ("clm_dspnsng_stus_cd", 81, 81, "X(01)"),
+        ("clm_daw_prod_slctn_cd", 82, 82, "X(01)"),
+        ("clm_line_srvc_unit_qty", 83, 106, "-9(18).9999"),
+        ("clm_line_days_suply_qty", 107, 115, "9(09)"),
+        ("prvdr_prsbng_id_qlfyr_cd", 116, 117, "X(02)"),
+        ("blank", 118, 137, "X(20)"),
+        ("clm_line_bene_pmt_amt", 138, 150, "-9(9).99"),
+        ("clm_adjsmt_type_cd", 151, 152, "X(02)"),
+        ("clm_efctv_dt", 153, 162, "YYYY-MM-DD"),
+        ("clm_idr_ld_dt", 163, 172, "YYYY-MM-DD"),
+        ("clm_line_rx_srvc_rfrnc_num", 173, 184, "9(12)"),
+        ("clm_line_rx_fill_num", 185, 193, "X(09)"),
+        ("clm_phrmcy_srvc_type_cd", 194, 195, "X(02)"),
+        ("clm_prsbng_prvdr_gnrc_id_num", 196, 230, "X(35)"),
+    ],
+    "cclf8": [
+        ("bene_mbi_id", 1, 11, "X(11)"),
+        ("bene_hic_num", 12, 22, "X(11)"),
+        ("bene_fips_state_cd", 23, 24, "9(02)"),
+        ("bene_fips_cnty_cd", 25, 27, "9(03)"),
+        ("bene_zip_cd", 28, 32, "X(05)"),
+        ("bene_dob", 33, 42, "YYYY-MM-DD"),
+        ("bene_sex_cd", 43, 43, "X(01)"),
+        ("bene_race_cd", 44, 44, "X(01)"),
+        ("bene_age", 45, 47, "9(03)"),
+        ("bene_mdcr_stus_cd", 48, 49, "X(02)"),
+        ("bene_dual_stus_cd", 50, 51, "X(02)"),
+        ("bene_death_dt", 52, 61, "YYYY-MM-DD"),
+        ("bene_rng_bgn_dt", 62, 71, "YYYY-MM-DD"),
+        ("bene_rng_end_dt", 72, 81, "YYYY-MM-DD"),
+        ("bene_1st_name", 82, 111, "X(30)"),
+        ("bene_midl_name", 112, 126, "X(15)"),
+        ("bene_last_name", 127, 166, "X(40)"),
+        ("bene_orgnl_entlmt_rsn_cd", 167, 167, "X(01)"),
+        ("bene_entlmt_buyin_ind", 168, 168, "X(01)"),
+        ("bene_part_a_enrlmt_bgn_dt", 169, 178, "YYYY-MM-DD"),
+        ("bene_part_b_enrlmt_bgn_dt", 179, 188, "YYYY-MM-DD"),
+        ("bene_line_1_adr", 189, 233, "X(45)"),
+        ("bene_line_2_adr", 234, 278, "X(45)"),
+        ("bene_line_3_adr", 279, 318, "X(40)"),
+        ("bene_line_4_adr", 319, 358, "X(40)"),
+        ("bene_line_5_adr", 359, 398, "X(40)"),
+        ("bene_line_6_adr", 399, 438, "X(40)"),
+        ("geo_zip_plc_name", 439, 538, "X(100)"),
+        ("geo_usps_state_cd", 539, 540, "X(2)"),
+        ("geo_zip5_cd", 541, 545, "X(5)"),
+        ("geo_zip4_cd", 546, 549, "X(4)"),
+    ],
+    "cclf9": [
+        ("hicn_mbi_xref_ind", 1, 1, "X(1)"),
+        ("crnt_num", 2, 12, "X(11)"),
+        ("prvs_num", 13, 23, "X(11)"),
+        ("prvs_id_efctv_dt", 24, 33, "YYYY-MM-DD"),
+        ("prvs_id_obslt_dt", 34, 43, "YYYY-MM-DD"),
+        ("bene_rrb_num", 44, 55, "X(12)"),
+    ],
+    "cclfa": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("bene_mbi_id", 14, 24, "X(11)"),
+        ("bene_hic_num", 25, 35, "X(11)"),
+        ("clm_type_cd", 36, 37, "9(02)"),
+        ("clm_actv_care_from_dt", 38, 47, "YYYY-MM-DD"),
+        ("clm_ngaco_pbpmt_sw", 48, 48, "X(1)"),
+        ("clm_ngaco_pdschrg_hcbs_sw", 49, 49, "X(1)"),
+        ("clm_ngaco_snf_wvr_sw", 50, 50, "X(1)"),
+        ("clm_ngaco_tlhlth_sw", 51, 51, "X(1)"),
+        ("clm_ngaco_cptatn_sw", 52, 52, "X(1)"),
+        ("clm_demo_1st_num", 53, 54, "X(2)"),
+        ("clm_demo_2nd_num", 55, 56, "X(2)"),
+        ("clm_demo_3rd_num", 57, 58, "X(2)"),
+        ("clm_demo_4th_num", 59, 60, "X(2)"),
+        ("clm_demo_5th_num", 61, 62, "X(2)"),
+        ("clm_pbp_inclsn_amt", 63, 81, "-9(15).99"),
+        ("clm_pbp_rdctn_amt", 82, 100, "-9(15).99"),
+        ("clm_ngaco_cmg_wvr_sw", 101, 101, "X(1)"),
+        ("clm_instnl_per_diem_amt", 102, 120, "-9(15).99"),
+        ("clm_mdcr_ip_bene_ddctbl_amt", 121, 135, "-9(11).99"),
+        ("clm_mdcr_coinsrnc_amt", 136, 154, "-9(15).99"),
+        ("clm_blood_lblty_amt", 155, 169, "-9(11).99"),
+        ("clm_instnl_prfnl_amt", 170, 184, "-9(11).99"),
+        ("clm_ncvrd_chrg_amt", 185, 203, "-9(15).99"),
+        ("clm_mdcr_ddctbl_amt", 204, 222, "-9(15).99"),
+        ("clm_rlt_cond_cd", 223, 224, "X(2)"),
+        ("clm_oprtnl_outlr_amt", 225, 243, "-9(15).99"),
+        ("clm_mdcr_new_tech_amt", 244, 262, "-9(15).99"),
+        ("clm_islet_isoln_amt", 263, 281, "-9(15).99"),
+        ("clm_sqstrtn_rdctn_amt", 282, 300, "-9(15).99"),
+        ("clm_1_rev_cntr_ansi_rsn_cd", 301, 303, "X(3)"),
+        ("clm_1_rev_cntr_ansi_grp_cd", 304, 305, "X(2)"),
+        ("clm_mips_pmt_amt", 306, 324, "-9(15).99"),
+    ],
+    "cclfb": [
+        ("cur_clm_uniq_id", 1, 13, "9(13)"),
+        ("clm_line_num", 14, 23, "9(10)"),
+        ("bene_mbi_id", 24, 34, "X(11)"),
+        ("bene_hic_num", 35, 45, "X(11)"),
+        ("clm_type_cd", 46, 47, "9(02)"),
+        ("clm_line_ngaco_pbpmt_sw", 48, 48, "X(1)"),
+        ("clm_line_ngaco_pdschrg_hcbs_sw", 49, 49, "X(1)"),
+        ("clm_line_ngaco_snf_wvr_sw", 50, 50, "X(1)"),
+        ("clm_line_ngaco_tlhlth_sw", 51, 51, "X(1)"),
+        ("clm_line_ngaco_cptatn_sw", 52, 52, "X(1)"),
+        ("clm_demo_1st_num", 53, 54, "X(2)"),
+        ("clm_demo_2nd_num", 55, 56, "X(2)"),
+        ("clm_demo_3rd_num", 57, 58, "X(2)"),
+        ("clm_demo_4th_num", 59, 60, "X(2)"),
+        ("clm_demo_5th_num", 61, 62, "X(2)"),
+        ("clm_pbp_inclsn_amt", 63, 77, "-9(11).99"),
+        ("clm_pbp_rdctn_amt", 78, 92, "-9(11).99"),
+        ("clm_ngaco_cmg_wvr_sw", 93, 93, "X(1)"),
+        ("clm_mdcr_ddctbl_amt", 94, 112, "-9(15).99"),
+        ("clm_sqstrtn_rdctn_amt", 113, 127, "-9(11).99"),
+        ("clm_line_carr_hpsa_scrcty_cd", 128, 128, "X(1)"),
+    ],
+}

src/cli/load.py

@@ -11,19 +11,93 @@ app = typer.Typer(no_args_is_help=True)
 
 @app.command()
 def cclf(
-    path: Path = typer.Option(None, help="Directory containing CCLF files."),
+    path: Path = typer.Option(..., help="Directory containing CCLF files."),
+    database: str = typer.Option(None, help="DuckDB database path (default: config)."),
+    no_pipeline: bool = typer.Option(
+        False, "--no-pipeline", help="Load raw tables only, skip CCLF pipeline."
+    ),
 ) -> None:
-    """Ingest CCLF claim files into DuckDB."""
+    """Ingest CCLF claim files into DuckDB.
-    typer.echo(f"load cclf: path={path} [stub]")
+
+    Reads fixed-width CCLF files (plain or ZIP) from the given directory,
+    parses them into the ``cclf`` schema, then runs the 19-step CCLF
+    pipeline to produce ``input_layer`` tables.
+    """
+    from aco.load.cclf import load_cclf_directory
+
+    typer.echo(f"Loading CCLF files from {path} ...")
+    try:
+        stats = load_cclf_directory(
+            path,
+            database=database,
+            run_pipeline=not no_pipeline,
+        )
+    except FileNotFoundError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1) from None
+
+    for table, count in sorted(stats.items()):
+        typer.echo(f"  {table}: {count:,} rows")
+    typer.echo("Done.")
 
 
 @app.command()
-def bcda() -> None:
+def bcda(
-    """Run a BCDA bulk FHIR export and ingest into DuckDB."""
+    ndjson_dir: Path = typer.Option(
-    typer.echo("load bcda [stub]")
+        None, help="Directory containing FHIR ndjson files."
+    ),
+    database: str = typer.Option(None, help="DuckDB database path (default: config)."),
+    skip_flatten: bool = typer.Option(
+        False,
+        "--skip-flatten",
+        help="Skip flattening, load existing Parquet.",
+    ),
+) -> None:
+    """Flatten BCDA FHIR exports and load into DuckDB.
+
+    Reads Patient/EOB/Coverage ndjson files, flattens them to Parquet,
+    and loads into the ``bcda`` schema.
+    """
+    from aco.load.bcda import load_bcda
+
+    typer.echo("Loading BCDA data ...")
+    try:
+        stats = load_bcda(
+            ndjson_dir=ndjson_dir,
+            database=database,
+            skip_flatten=skip_flatten,
+        )
+    except FileNotFoundError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1) from None
+
+    for table, count in sorted(stats.items()):
+        typer.echo(f"  {table}: {count:,} rows")
+    typer.echo("Done.")
 
 
 @app.command()
-def seed() -> None:
+def seed(
-    """Load seed/reference data into DuckDB."""
+    seed_dir: Path = typer.Option(None, help="Directory containing seed files."),
-    typer.echo("load seed [stub]")
+    database: str = typer.Option(None, help="DuckDB database path (default: config)."),
+    schema: str = typer.Option(
+        "reference_data", help="Target DuckDB schema for seed tables."
+    ),
+) -> None:
+    """Load seed/reference data (CSV, Excel, Parquet) into DuckDB."""
+    from aco.load.seed import load_seeds
+
+    typer.echo("Loading seed data ...")
+    try:
+        stats = load_seeds(
+            seed_dir=seed_dir,
+            database=database,
+            schema=schema,
+        )
+    except FileNotFoundError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1) from None
+
+    for table, count in sorted(stats.items()):
+        typer.echo(f"  {table}: {count:,} rows")
+    typer.echo("Done.")

tests/aco/test_load_bcda.py

@@ -0,0 +1,32 @@
+"""Tests for BCDA data loading."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from aco.load.bcda import _find_latest_export
+
+
+class TestFindLatestExport:
+    def test_no_exports_dir_raises(self, tmp_path: Path) -> None:
+        with pytest.raises(FileNotFoundError, match="No exports directory"):
+            _find_latest_export(tmp_path)
+
+    def test_empty_exports_dir_raises(self, tmp_path: Path) -> None:
+        (tmp_path / "exports").mkdir()
+        with pytest.raises(FileNotFoundError, match="No export directories"):
+            _find_latest_export(tmp_path)
+
+    def test_finds_latest(self, tmp_path: Path) -> None:
+        import os
+
+        exports = tmp_path / "exports"
+        exports.mkdir()
+        (exports / "old_job").mkdir()
+        (exports / "new_job").mkdir()
+        # Set old_job mtime to past so new_job is clearly newer
+        os.utime(exports / "old_job", (1000000, 1000000))
+        result = _find_latest_export(tmp_path)
+        assert result.name == "new_job"

tests/aco/test_load_cclf.py

@@ -0,0 +1,169 @@
+"""Tests for CCLF file loading."""
+
+from __future__ import annotations
+
+from datetime import date
+from pathlib import Path
+
+import pytest
+
+from aco.load.cclf import (
+    _parse_value,
+    discover_cclf_files,
+    load_cclf_directory,
+    parse_cclf_file,
+)
+from aco.table.cclf_layout import LAYOUTS
+
+
+class TestParseValue:
+    def test_empty_string(self) -> None:
+        assert _parse_value("", "X(11)") is None
+
+    def test_whitespace_only(self) -> None:
+        assert _parse_value("   ", "X(11)") is None
+
+    def test_string_field(self) -> None:
+        assert _parse_value("1234567890", "X(10)") == "1234567890"
+
+    def test_date_field(self) -> None:
+        assert _parse_value("2025-07-16", "YYYY-MM-DD") == date(2025, 7, 16)
+
+    def test_date_invalid(self) -> None:
+        assert _parse_value("0000-00-00", "YYYY-MM-DD") is None
+
+    def test_numeric_field(self) -> None:
+        assert _parse_value("123.45", "-9(13).99") == 123.45
+
+    def test_numeric_invalid(self) -> None:
+        assert _parse_value("N/A", "-9(13).99") is None
+
+    def test_numeric_9_format(self) -> None:
+        assert _parse_value("42", "9(13)") == "42"
+
+
+class TestParseCclfFile:
+    def test_parse_cclf9(self) -> None:
+        """CCLF9 is the simplest file: 6 fields, 55 bytes per line."""
+        # Build a realistic CCLF9 line: positions 1-55
+        line = (
+            "N"  # hicn_mbi_xref_ind (1-1)
+            "1AN0Y00AA04"  # crnt_num (2-12)
+            "2AN0Y00AA05"  # prvs_num (13-23)
+            "2024-01-15"  # prvs_id_efctv_dt (24-33)
+            "2025-06-30"  # prvs_id_obslt_dt (34-43)
+            "RRB123456789"  # bene_rrb_num (44-55)
+        )
+        df = parse_cclf_file([line], "cclf9")
+        assert len(df) == 1
+        assert df["hicn_mbi_xref_ind"][0] == "N"
+        assert df["crnt_num"][0] == "1AN0Y00AA04"
+        assert df["prvs_num"][0] == "2AN0Y00AA05"
+        assert df["prvs_id_efctv_dt"][0] == date(2024, 1, 15)
+        assert df["prvs_id_obslt_dt"][0] == date(2025, 6, 30)
+        assert df["bene_rrb_num"][0] == "RRB123456789"
+
+    def test_parse_empty_lines(self) -> None:
+        df = parse_cclf_file(["", "  ", ""], "cclf9")
+        assert df.is_empty()
+
+    def test_parse_cclf8_date_and_string(self) -> None:
+        """Check a CCLF8 line with date, numeric, and string fields."""
+        # Build CCLF8 line (549 bytes)
+        line = " " * 549
+        parts = list(line)
+        # bene_mbi_id (1-11)
+        for i, c in enumerate("1AN0Y00AA04"):
+            parts[i] = c
+        # bene_dob (33-42)
+        for i, c in enumerate("1945-03-22"):
+            parts[32 + i] = c
+        # bene_sex_cd (43)
+        parts[42] = "1"
+        # bene_race_cd (44)
+        parts[43] = "2"
+        line = "".join(parts)
+        df = parse_cclf_file([line], "cclf8")
+        assert len(df) == 1
+        assert df["bene_mbi_id"][0] == "1AN0Y00AA04"
+        assert df["bene_dob"][0] == date(1945, 3, 22)
+        assert df["bene_sex_cd"][0] == "1"
+        assert df["bene_race_cd"][0] == "2"
+
+    def test_empty_file_returns_schema(self) -> None:
+        df = parse_cclf_file([], "cclf9")
+        assert df.is_empty()
+        assert "crnt_num" in df.columns
+
+
+class TestDiscoverCclfFiles:
+    def test_discover_plain_files(self, tmp_path: Path) -> None:
+        # Create a couple of fake CCLF files
+        (tmp_path / "P.A1234.ACO.ZC1Y25.D250716.T1234567").write_text("data\n")
+        (tmp_path / "P.A1234.ACO.ZC8Y25.D250716.T1234567").write_text("data\n")
+        (tmp_path / "random.txt").write_text("noise\n")
+
+        found = discover_cclf_files(tmp_path)
+        assert "cclf1" in found
+        assert "cclf8" in found
+        assert len(found) == 2
+
+    def test_discover_skips_hidden(self, tmp_path: Path) -> None:
+        (tmp_path / ".extracted").mkdir()
+        (tmp_path / "P.A1234.ACO.ZC1Y25.D250716.T1234567").write_text("data\n")
+        found = discover_cclf_files(tmp_path)
+        assert "cclf1" in found
+
+
+class TestLayouts:
+    def test_all_pipeline_inputs_have_layouts(self) -> None:
+        """Every CCLF table referenced by the pipeline must have a layout."""
+        needed = {
+            "cclf1",
+            "cclf2",
+            "cclf3",
+            "cclf4",
+            "cclf5",
+            "cclf6",
+            "cclf7",
+            "cclf8",
+            "cclf9",
+        }
+        assert needed.issubset(LAYOUTS.keys())
+
+    def test_positions_are_sequential(self) -> None:
+        """Field positions should be non-overlapping and ordered."""
+        for table, fields in LAYOUTS.items():
+            if table == "cclf0":
+                continue  # CCLF0 has two record types sharing positions
+            for name, start, end, fmt in fields:
+                if name in ("blank", "delimiter", "filler"):
+                    continue
+                assert start <= end, f"{table}.{name}: start {start} > end {end}"
+                assert start >= 1, f"{table}.{name}: start must be >= 1"
+
+
+class TestLoadCclfDirectory:
+    def test_no_files_raises(self, tmp_path: Path) -> None:
+        with pytest.raises(FileNotFoundError):
+            load_cclf_directory(tmp_path, run_pipeline=False)
+
+    def test_load_raw_no_pipeline(self, tmp_path: Path) -> None:
+        """Load a single CCLF9 file into a temp DuckDB (no pipeline)."""
+        import duckdb
+
+        # Create a fake CCLF9 file
+        line = "N1AN0Y00AA042AN0Y00AA052024-01-152025-06-30RRB123456789"
+        cclf_file = tmp_path / "P.A1234.ACO.ZC9Y25.D250716.T1234567"
+        cclf_file.write_text(line + "\n")
+
+        db_path = str(tmp_path / "test.duckdb")
+        stats = load_cclf_directory(tmp_path, database=db_path, run_pipeline=False)
+        assert stats["cclf9"] == 1
+
+        # Verify it's in DuckDB
+        con = duckdb.connect(db_path, read_only=True)
+        rows = con.execute("SELECT * FROM cclf.cclf9").fetchall()
+        assert len(rows) == 1
+        assert rows[0][1] == "1AN0Y00AA04"  # crnt_num
+        con.close()

tests/aco/test_load_seed.py

@@ -0,0 +1,63 @@
+"""Tests for seed data loading."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import duckdb
+import pytest
+
+from aco.load.seed import _table_name, load_seeds
+
+
+class TestTableName:
+    def test_simple_csv(self) -> None:
+        assert _table_name(Path("fips_county.csv")) == "fips_county"
+
+    def test_xlsx_with_parens(self) -> None:
+        result = _table_name(Path("ACO_REACH_bulk_upload (1).xlsx"))
+        assert result == "aco_reach_bulk_upload"
+
+    def test_spaces_become_underscores(self) -> None:
+        result = _table_name(Path("Static Report Crosswalk.xlsx"))
+        assert result == "static_report_crosswalk"
+
+
+class TestLoadSeeds:
+    def test_no_dir_raises(self, tmp_path: Path) -> None:
+        with pytest.raises(FileNotFoundError):
+            load_seeds(seed_dir=tmp_path / "nonexistent")
+
+    def test_load_csv_seed(self, tmp_path: Path) -> None:
+        # Create a CSV seed
+        csv = tmp_path / "test_codes.csv"
+        csv.write_text("code,description\nA01,Test Code A\nB02,Test Code B\n")
+
+        db_path = str(tmp_path / "test.duckdb")
+        stats = load_seeds(seed_dir=tmp_path, database=db_path)
+
+        assert "reference_data.test_codes" in stats
+        assert stats["reference_data.test_codes"] == 2
+
+        # Verify data
+        con = duckdb.connect(db_path, read_only=True)
+        rows = con.execute(
+            "SELECT * FROM reference_data.test_codes ORDER BY code"
+        ).fetchall()
+        assert len(rows) == 2
+        assert rows[0][0] == "A01"
+        con.close()
+
+    def test_skips_non_tabular(self, tmp_path: Path) -> None:
+        (tmp_path / "readme.txt").write_text("not a seed")
+        (tmp_path / "data.pdf").write_text("not tabular")
+        db_path = str(tmp_path / "test.duckdb")
+        stats = load_seeds(seed_dir=tmp_path, database=db_path)
+        assert stats == {}
+
+    def test_custom_schema(self, tmp_path: Path) -> None:
+        csv = tmp_path / "test.csv"
+        csv.write_text("x\n1\n")
+        db_path = str(tmp_path / "test.duckdb")
+        stats = load_seeds(seed_dir=tmp_path, database=db_path, schema="terminology")
+        assert "terminology.test" in stats

tests/aco/test_load_stage.py

@@ -0,0 +1,61 @@
+"""Tests for the unified staging pipeline."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from aco.load.stage import stage_all
+
+
+class TestStageAll:
+    def test_no_sources_returns_empty(self) -> None:
+        result = stage_all()
+        assert result == {}
+
+    def test_cclf_only(self, tmp_path: Path) -> None:
+        # Create a minimal CCLF9 file
+        line = "N1AN0Y00AA042AN0Y00AA052024-01-152025-06-30RRB123456789"
+        cclf_dir = tmp_path / "cclf"
+        cclf_dir.mkdir()
+        (cclf_dir / "P.A1234.ACO.ZC9Y25.D250716.T1234567").write_text(line + "\n")
+
+        db_path = str(tmp_path / "test.duckdb")
+        result = stage_all(
+            cclf_dir=cclf_dir,
+            database=db_path,
+            run_pipelines=False,
+        )
+        assert "cclf" in result
+        assert result["cclf"]["cclf9"] == 1
+
+    def test_seed_only(self, tmp_path: Path) -> None:
+        seed_dir = tmp_path / "seeds"
+        seed_dir.mkdir()
+        (seed_dir / "codes.csv").write_text("code\nA01\nB02\n")
+
+        db_path = str(tmp_path / "test.duckdb")
+        result = stage_all(seed_dir=seed_dir, database=db_path)
+        assert "seed" in result
+        assert result["seed"]["reference_data.codes"] == 2
+
+    def test_combined(self, tmp_path: Path) -> None:
+        # CCLF
+        cclf_dir = tmp_path / "cclf"
+        cclf_dir.mkdir()
+        line = "N1AN0Y00AA042AN0Y00AA052024-01-152025-06-30RRB123456789"
+        (cclf_dir / "P.A1234.ACO.ZC9Y25.D250716.T1234567").write_text(line + "\n")
+
+        # Seeds
+        seed_dir = tmp_path / "seeds"
+        seed_dir.mkdir()
+        (seed_dir / "test.csv").write_text("x\n1\n")
+
+        db_path = str(tmp_path / "test.duckdb")
+        result = stage_all(
+            cclf_dir=cclf_dir,
+            seed_dir=seed_dir,
+            database=db_path,
+            run_pipelines=False,
+        )
+        assert "cclf" in result
+        assert "seed" in result

tests/cli/test_cli.py

@@ -40,20 +40,19 @@ class TestLoad:
         for cmd in ("cclf", "bcda", "seed"):
             assert cmd in result.output
 
-    def test_load_cclf(self) -> None:
+    def test_load_cclf_requires_path(self) -> None:
         result = runner.invoke(app, ["load", "cclf"])
-        assert result.exit_code == 0
+        assert result.exit_code != 0
-        assert "load cclf" in result.output
 
-    def test_load_bcda(self) -> None:
+    def test_load_bcda_help(self) -> None:
-        result = runner.invoke(app, ["load", "bcda"])
+        result = runner.invoke(app, ["load", "bcda", "--help"])
         assert result.exit_code == 0
-        assert "load bcda" in result.output
+        assert "ndjson" in result.output
 
-    def test_load_seed(self) -> None:
+    def test_load_seed_help(self) -> None:
-        result = runner.invoke(app, ["load", "seed"])
+        result = runner.invoke(app, ["load", "seed", "--help"])
         assert result.exit_code == 0
-        assert "load seed" in result.output
+        assert "seed" in result.output.lower()
 
 
 class TestGenerate: