From afdbcbd0621baf5710daab367b21083b696bb9e3 Mon Sep 17 00:00:00 2001 From: Daphne Hansell <128793799+daphnehanse11@users.noreply.github.com> Date: Tue, 14 Jul 2026 10:24:28 -0400 Subject: [PATCH 1/3] Add label-verified CPS ASEC firm-size (NOEMP) reader (#193) Workstream A week-1 piece of the employer-firm plan (#192): a regime-aware reader for the ASEC firm-size label. NOEMP keeps the same 0:6 code domain every year while codes 2-3 change meaning (10-49/50-99 in 2011-2018, 10-24/25-99 in 2019+, verified against all fifteen Census data dictionaries 2011-2025), so the reader hard-codes the per-year map, refuses unverified years and path/year mismatches, and enforces the WKSWORK>0 universe and code domains at read time. Records carry LJCW class of worker, longest-job industry, the I_NOEMP allocation flag, and MARSUPWT; firm_size_tabulation emits the weighted band evidence the C2 banding decision consumes. Staging follows the PSID pattern (~/PolicyEngine/asec-data, POPULACE_DYNAMICS_ASEC_DIR override). Co-Authored-By: Claude Fable 5 --- src/populace_dynamics/data/asec_firm_size.py | 333 +++++++++++++++++++ tests/data/test_asec_firm_size.py | 189 +++++++++++ tests/tier_counts.json | 2 +- 3 files changed, 523 insertions(+), 1 deletion(-) create mode 100644 src/populace_dynamics/data/asec_firm_size.py create mode 100644 tests/data/test_asec_firm_size.py diff --git a/src/populace_dynamics/data/asec_firm_size.py b/src/populace_dynamics/data/asec_firm_size.py new file mode 100644 index 00000000..d938401d --- /dev/null +++ b/src/populace_dynamics/data/asec_firm_size.py @@ -0,0 +1,333 @@ +"""CPS ASEC firm-size (NOEMP) records for the employer extension. + +NOEMP is the worker-reported count of employees "at all locations +where this employer operates" for the **longest job held last +calendar year** (universe ``WKSWORK > 0``), and is the plan's +designated firm-size training label (issue #192; loader scope in +issue #193). The variable keeps the identical code domain ``0:6`` in +every year while codes 2 and 3 silently change meaning across two +band regimes, so a year-blind read mis-bands those codes with no +error. This reader hard-codes the dictionary-adjudicated map per +year and refuses years it has not verified. + +Band regimes, verified against every year's Census public-use data +dictionary at www2.census.gov/programs-surveys/cps/datasets +(read 2026-07-14): + +* 2011-2018 (``asec2011_pubuse.dd.txt`` ... ``08ASEC2018_Data_ + Dict_Full.txt``): 1 under 10, **2 = 10-49, 3 = 50-99**, + 4 = 100-499, 5 = 500-999, 6 = 1000+. +* 2019-2025 (``06_ASEC_2019-Data_Dictionary_Full.pdf`` ... + ``asec2025_ddl_pub_full.pdf``): 1 under 10, **2 = 10-24, + 3 = 25-99**, 4 = 100-499, 5 = 500-999, 6 = 1000+. + +Note the 2019+ regime cannot resolve a 50-employee cut (it falls +inside 25-99), while 2011-2018 can — the C2 banding decision on +issue #192 consumes the tabulations this module emits. + +Alongside the band, each record carries the fields the calibration +side needs to reason about universes: ``LJCW`` (longest-job class of +worker — SUSB/QWI targets exclude government and self-employment), +longest-job industry (detailed ``INDUSTRY`` and major-group +``WEIND``), ``WKSWORK``, the ``I_NOEMP`` allocation flag, and the +ASEC supplement weight ``MARSUPWT``. + +Staging: like the PSID readers, raw microdata stays out of the +repository. Person files are staged as CSVs with original Census +variable names (the published ``asecpub{yy}csv.zip`` person file for +2017+; earlier years converted from the fixed-width ``.dat`` with the +year's ``.dd`` layout) under ``~/PolicyEngine/asec-data`` as +``pppub{yy}.csv`` (or ``.csv.gz``), overridable via the +``POPULACE_DYNAMICS_ASEC_DIR`` environment variable. +""" + +from __future__ import annotations + +import os +import re +from pathlib import Path + +import pandas as pd + +__all__ = [ + "ASEC_FIRM_SIZE_YEARS", + "CLASS_OF_WORKER_LABELS", + "NOEMP_BANDS_2011_2018", + "NOEMP_BANDS_2019_PLUS", + "band_regime", + "firm_size_tabulation", + "noemp_band_map", + "read_asec_firm_size", +] + +#: NOEMP code -> band label, 2011-2018 dictionaries (code 0 is NIU). +NOEMP_BANDS_2011_2018: dict[int, str] = { + 1: "under_10", + 2: "10_49", + 3: "50_99", + 4: "100_499", + 5: "500_999", + 6: "1000_plus", +} + +#: NOEMP code -> band label, 2019-2025 dictionaries (code 0 is NIU). +NOEMP_BANDS_2019_PLUS: dict[int, str] = { + 1: "under_10", + 2: "10_24", + 3: "25_99", + 4: "100_499", + 5: "500_999", + 6: "1000_plus", +} + +#: Survey years whose dictionaries the band maps were verified +#: against; ``noemp_band_map`` refuses anything else. +ASEC_FIRM_SIZE_YEARS: tuple[int, ...] = tuple(range(2011, 2026)) + +#: LJCW code -> label, from the 2024 dictionary ("longest job class +#: of worker"; 5/6 are self-employed incorporated yes/no-or-farm). +CLASS_OF_WORKER_LABELS: dict[int, str] = { + 1: "private", + 2: "federal", + 3: "state", + 4: "local", + 5: "self_employed_incorporated", + 6: "self_employed_unincorporated", + 7: "without_pay", +} + +#: Raw person-file columns the reader requires. +_REQUIRED_COLUMNS = ( + "PERIDNUM", + "NOEMP", + "I_NOEMP", + "LJCW", + "INDUSTRY", + "WEIND", + "WKSWORK", + "MARSUPWT", +) + +_DATA_DIR_ENV = "POPULACE_DYNAMICS_ASEC_DIR" +_DEFAULT_DATA_DIR = Path("~/PolicyEngine/asec-data").expanduser() + +_README_POINTER = ( + "stage Census ASEC person files as pppub{yy}.csv[.gz] under " + "~/PolicyEngine/asec-data (or POPULACE_DYNAMICS_ASEC_DIR)" +) + +#: A pppub filename encodes its survey year as two digits; used to +#: refuse a path/year mismatch instead of silently mis-banding. +_PPPUB_YEAR_RE = re.compile(r"pppub(\d{2})\.csv(\.gz)?$", re.I) + + +def band_regime(year: int) -> str: + """Return the band-regime key (``"2011_2018"``/``"2019_plus"``). + + Raises: + ValueError: If ``year`` has no dictionary-verified band map. + """ + if year not in ASEC_FIRM_SIZE_YEARS: + raise ValueError( + f"No dictionary-verified NOEMP band map for ASEC {year}; " + f"supported years are {ASEC_FIRM_SIZE_YEARS[0]}-" + f"{ASEC_FIRM_SIZE_YEARS[-1]}. Extend the module only " + "with that year's Census data dictionary in hand." + ) + return "2011_2018" if year <= 2018 else "2019_plus" + + +def noemp_band_map(year: int) -> dict[int, str]: + """Return the NOEMP code -> band label map for a survey year.""" + if band_regime(year) == "2011_2018": + return dict(NOEMP_BANDS_2011_2018) + return dict(NOEMP_BANDS_2019_PLUS) + + +def _resolve_data_dir(data_dir: Path | None) -> Path: + """Resolve the ASEC data directory from arg, env var, default.""" + if data_dir is not None: + return Path(data_dir).expanduser() + env_value = os.environ.get(_DATA_DIR_ENV) + if env_value: + return Path(env_value).expanduser() + return _DEFAULT_DATA_DIR + + +def _resolve_person_path(year: int, data_dir: Path) -> Path: + """Locate the staged person file for ``year`` under ``data_dir``.""" + stem = f"pppub{year % 100:02d}" + for suffix in (".csv", ".csv.gz"): + candidate = data_dir / f"{stem}{suffix}" + if candidate.exists(): + return candidate + raise FileNotFoundError( + f"No {stem}.csv[.gz] under {data_dir}; {_README_POINTER}." + ) + + +def _check_path_year(path: Path, year: int) -> None: + """Refuse a pppub filename whose year digits contradict ``year``.""" + match = _PPPUB_YEAR_RE.search(path.name) + if match is None: + return + file_year = 2000 + int(match.group(1)) + if file_year != year: + raise ValueError( + f"{path.name} is an ASEC {file_year} person file but " + f"year={year} was requested; the NOEMP band regimes " + "differ across years, so the mismatch would silently " + "mis-band codes 2-3." + ) + + +def _domain_error(year: int, column: str, bad: pd.Series) -> ValueError: + values = ", ".join(str(v) for v in sorted(bad.unique())[:8]) + return ValueError( + f"ASEC {year} {column} contains out-of-dictionary value(s) " + f"[{values}] on {len(bad)} row(s); refusing to band a file " + "that does not match the year's data dictionary." + ) + + +def read_asec_firm_size( + year: int, + *, + path: str | Path | None = None, + data_dir: str | Path | None = None, +) -> pd.DataFrame: + """Read one ASEC year's longest-job firm-size records. + + Args: + year: ASEC survey year (the job attributes describe the + longest job of calendar year ``year - 1``). + path: Explicit person-file CSV. When the filename carries + pppub year digits they must agree with ``year``. + data_dir: Staging directory (default resolution: explicit + argument, then ``POPULACE_DYNAMICS_ASEC_DIR``, then + ``~/PolicyEngine/asec-data``). + + Returns: + One row per person in the NOEMP universe (``WKSWORK > 0``, + i.e. worked last calendar year), with columns ``person_id``, + ``year``, ``income_year``, ``band_regime``, ``noemp``, + ``firm_size_band``, ``noemp_allocated``, ``ljcw``, + ``class_of_worker``, ``industry_major``, + ``industry_detailed``, ``wkswork``, and ``weight`` + (``MARSUPWT``, persons). + + Raises: + ValueError: On an unsupported year, a path/year mismatch, + missing required columns, or any value outside the + year's dictionary domain (NOEMP outside 0-6, LJCW + outside 0-7, or a nonzero NOEMP off the ``WKSWORK > 0`` + universe). + FileNotFoundError: If no staged person file can be found. + """ + bands = noemp_band_map(year) + if path is not None: + person_path = Path(path).expanduser() + if not person_path.exists(): + raise FileNotFoundError( + f"ASEC person file not found: {person_path}" + ) + else: + person_path = _resolve_person_path(year, _resolve_data_dir(data_dir)) + _check_path_year(person_path, year) + + raw = pd.read_csv(person_path, usecols=None, low_memory=False) + missing = sorted(set(_REQUIRED_COLUMNS) - set(raw.columns)) + if missing: + raise ValueError( + f"{person_path.name} is missing required column(s) " + f"{missing}; expected original Census ASEC person-file " + "variable names." + ) + raw = raw.loc[:, list(_REQUIRED_COLUMNS)].copy() + + for column, low, high in ( + ("NOEMP", 0, 6), + ("LJCW", 0, 7), + ("WKSWORK", 0, 52), + ): + values = pd.to_numeric(raw[column], errors="coerce") + bad = raw[column][values.isna() | (values < low) | (values > high)] + if len(bad): + raise _domain_error(year, column, bad) + raw[column] = values.astype(int) + + off_universe = raw[(raw["NOEMP"] > 0) & (raw["WKSWORK"] == 0)] + if len(off_universe): + raise ValueError( + f"ASEC {year}: {len(off_universe)} row(s) carry a nonzero " + "NOEMP outside the WKSWORK > 0 universe; the file does " + "not match the dictionary's universe statement." + ) + + universe = raw[raw["WKSWORK"] > 0].reset_index(drop=True) + return pd.DataFrame( + { + "person_id": universe["PERIDNUM"].astype(str), + "year": year, + "income_year": year - 1, + "band_regime": band_regime(year), + "noemp": universe["NOEMP"], + "firm_size_band": universe["NOEMP"].map( + lambda code: bands.get(int(code), "niu") + ), + "noemp_allocated": pd.to_numeric(universe["I_NOEMP"]) > 0, + "ljcw": universe["LJCW"], + "class_of_worker": universe["LJCW"].map( + lambda code: CLASS_OF_WORKER_LABELS.get(int(code), "niu") + ), + "industry_major": pd.to_numeric(universe["WEIND"]).astype(int), + "industry_detailed": pd.to_numeric(universe["INDUSTRY"]).astype( + int + ), + "wkswork": universe["WKSWORK"], + "weight": pd.to_numeric(universe["MARSUPWT"]).astype(float), + } + ) + + +def firm_size_tabulation( + records: pd.DataFrame, + by: tuple[str, ...] = ( + "year", + "band_regime", + "firm_size_band", + "class_of_worker", + ), +) -> pd.DataFrame: + """Weighted firm-size tabulation — the C2 evidence artifact. + + Args: + records: Output of :func:`read_asec_firm_size` (one or more + years concatenated). + by: Grouping columns. + + Returns: + One row per group with ``weighted_persons`` (MARSUPWT sum), + ``unweighted_n``, and ``allocated_share`` (weighted share of + the group whose NOEMP was allocated/edited), sorted by the + grouping columns. + """ + missing = sorted( + (set(by) | {"weight", "noemp_allocated"}) - set(records.columns) + ) + if missing: + raise ValueError( + f"records is missing column(s) {missing}; pass the " + "output of read_asec_firm_size." + ) + working = records.assign( + _allocated_weight=records["weight"] * records["noemp_allocated"] + ) + grouped = working.groupby(list(by), sort=True) + out = grouped.agg( + weighted_persons=("weight", "sum"), + unweighted_n=("weight", "size"), + _allocated_weight=("_allocated_weight", "sum"), + ).reset_index() + out["allocated_share"] = out["_allocated_weight"] / out["weighted_persons"] + return out.drop(columns="_allocated_weight") diff --git a/tests/data/test_asec_firm_size.py b/tests/data/test_asec_firm_size.py new file mode 100644 index 00000000..1900115c --- /dev/null +++ b/tests/data/test_asec_firm_size.py @@ -0,0 +1,189 @@ +"""Tests for the ASEC firm-size (NOEMP) reader and tabulation.""" + +from __future__ import annotations + +from pathlib import Path + +import pandas as pd +import pytest + +from populace_dynamics.data import asec_firm_size + +REAL_DATA = Path("~/PolicyEngine/asec-data").expanduser() +needs_real_asec = pytest.mark.skipif( + not REAL_DATA.is_dir(), + reason="ASEC person files not staged", +) + + +def _write_person_file( + directory: Path, + year: int, + rows: list[dict], + filename: str | None = None, +) -> Path: + """Write a fixture Census-style ASEC person CSV. + + Each row dict may override any raw column; defaults describe a + private-sector worker with an unallocated NOEMP of 2 who worked + all year, so tests only state what they exercise. + """ + directory.mkdir(parents=True, exist_ok=True) + defaults = { + "PERIDNUM": 0, + "NOEMP": 2, + "I_NOEMP": 0, + "LJCW": 1, + "INDUSTRY": 770, + "WEIND": 4, + "WKSWORK": 52, + "MARSUPWT": 1000.0, + } + frame = pd.DataFrame( + [ + {**defaults, "PERIDNUM": 10_000 + i, **row} + for i, row in enumerate(rows) + ] + ) + path = directory / (filename or f"pppub{year % 100:02d}.csv") + frame.to_csv(path, index=False) + return path + + +class TestBandMaps: + def test_2016_maps_code_2_to_10_49(self): + assert asec_firm_size.noemp_band_map(2016)[2] == "10_49" + assert asec_firm_size.noemp_band_map(2016)[3] == "50_99" + + def test_2024_maps_code_2_to_10_24(self): + assert asec_firm_size.noemp_band_map(2024)[2] == "10_24" + assert asec_firm_size.noemp_band_map(2024)[3] == "25_99" + + def test_regime_boundaries(self): + assert asec_firm_size.band_regime(2011) == "2011_2018" + assert asec_firm_size.band_regime(2018) == "2011_2018" + assert asec_firm_size.band_regime(2019) == "2019_plus" + assert asec_firm_size.band_regime(2025) == "2019_plus" + + def test_unverified_year_raises(self): + with pytest.raises(ValueError, match="dictionary-verified"): + asec_firm_size.noemp_band_map(2010) + with pytest.raises(ValueError, match="dictionary-verified"): + asec_firm_size.noemp_band_map(2026) + + +class TestReadAsecFirmSize: + def test_bands_by_regime(self, tmp_path): + rows = [{"NOEMP": 2}, {"NOEMP": 3}] + for year, expected in ( + (2016, ["10_49", "50_99"]), + (2024, ["10_24", "25_99"]), + ): + path = _write_person_file(tmp_path / str(year), year, rows) + out = asec_firm_size.read_asec_firm_size(year, path=path) + assert list(out["firm_size_band"]) == expected + assert ( + out["band_regime"] == asec_firm_size.band_regime(year) + ).all() + assert (out["income_year"] == year - 1).all() + + def test_year_and_filename_must_agree(self, tmp_path): + path = _write_person_file(tmp_path, 2016, [{}]) + with pytest.raises(ValueError, match="mis-band"): + asec_firm_size.read_asec_firm_size(2024, path=path) + + def test_resolves_staged_directory(self, tmp_path): + _write_person_file(tmp_path, 2021, [{"NOEMP": 3}]) + out = asec_firm_size.read_asec_firm_size(2021, data_dir=tmp_path) + assert list(out["firm_size_band"]) == ["25_99"] + + def test_missing_staged_file_raises(self, tmp_path): + with pytest.raises(FileNotFoundError, match="pppub21"): + asec_firm_size.read_asec_firm_size(2021, data_dir=tmp_path) + + def test_missing_column_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{}]) + frame = pd.read_csv(path).drop(columns="LJCW") + frame.to_csv(path, index=False) + with pytest.raises(ValueError, match=r"\['LJCW'\]"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_out_of_dictionary_noemp_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"NOEMP": 7}]) + with pytest.raises(ValueError, match="out-of-dictionary"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_noemp_outside_universe_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"NOEMP": 2, "WKSWORK": 0}]) + with pytest.raises(ValueError, match="universe"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_universe_and_flags(self, tmp_path): + rows = [ + {"NOEMP": 1, "I_NOEMP": 1, "LJCW": 6}, + {"NOEMP": 0, "WKSWORK": 0, "LJCW": 0}, # NIU, dropped + {"NOEMP": 4, "LJCW": 3}, + ] + path = _write_person_file(tmp_path, 2021, rows) + out = asec_firm_size.read_asec_firm_size(2021, path=path) + assert len(out) == 2 + assert list(out["noemp_allocated"]) == [True, False] + assert list(out["class_of_worker"]) == [ + "self_employed_unincorporated", + "state", + ] + + def test_env_var_staging(self, tmp_path, monkeypatch): + _write_person_file(tmp_path, 2021, [{}]) + monkeypatch.setenv("POPULACE_DYNAMICS_ASEC_DIR", str(tmp_path)) + out = asec_firm_size.read_asec_firm_size(2021) + assert len(out) == 1 + + +class TestFirmSizeTabulation: + def test_weighted_counts_and_allocated_share(self, tmp_path): + rows = [ + {"NOEMP": 2, "MARSUPWT": 1000.0, "I_NOEMP": 1}, + {"NOEMP": 2, "MARSUPWT": 3000.0, "I_NOEMP": 0}, + {"NOEMP": 5, "MARSUPWT": 500.0, "I_NOEMP": 0, "LJCW": 2}, + ] + path = _write_person_file(tmp_path, 2024, rows) + records = asec_firm_size.read_asec_firm_size(2024, path=path) + out = asec_firm_size.firm_size_tabulation(records) + ten_24 = out[out["firm_size_band"] == "10_24"].iloc[0] + assert ten_24["weighted_persons"] == 4000.0 + assert ten_24["unweighted_n"] == 2 + assert ten_24["allocated_share"] == 0.25 + federal = out[out["class_of_worker"] == "federal"].iloc[0] + assert federal["weighted_persons"] == 500.0 + assert federal["firm_size_band"] == "500_999" + + def test_regime_break_is_visible_across_years(self, tmp_path): + code_2 = [{"NOEMP": 2}] + frames = [ + asec_firm_size.read_asec_firm_size( + year, + path=_write_person_file(tmp_path / str(year), year, code_2), + ) + for year in (2018, 2019) + ] + out = asec_firm_size.firm_size_tabulation(pd.concat(frames)) + assert set(out["firm_size_band"]) == {"10_49", "10_24"} + + def test_wrong_frame_raises(self): + with pytest.raises(ValueError, match="read_asec_firm_size"): + asec_firm_size.firm_size_tabulation(pd.DataFrame({"x": [1]})) + + +@needs_real_asec +class TestRealData: + def test_reads_any_staged_year(self): + staged = sorted(REAL_DATA.glob("pppub*.csv*")) + if not staged: + pytest.skip("no pppub files staged") + year = 2000 + int(staged[-1].name[5:7]) + out = asec_firm_size.read_asec_firm_size(year, path=staged[-1]) + assert len(out) > 10_000 + assert set(out["firm_size_band"]) <= set( + asec_firm_size.noemp_band_map(year).values() + ) diff --git a/tests/tier_counts.json b/tests/tier_counts.json index b298cf6d..3c011a8f 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 420, + "unit": 450, "artifact": 1011, "integration_psid": 800, "reproduction_legacy": 520, From 288bae1ba165b05ab6237386fa86ad548a4be0f1 Mon Sep 17 00:00:00 2001 From: Daphne Hansell <128793799+daphnehanse11@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:08:06 -0400 Subject: [PATCH 2/3] Address review: string PERIDNUM, symmetric universe checks, total error paths - Read PERIDNUM as string: 22-digit ids overflow int64 and a float64 fallback rounds distinct persons together before astype(str); adds a last-digit-apart fixture test and a uniqueness assertion to the real-data pass. - Enforce the shared NOEMP/LJCW universe symmetrically (nonzero exactly where WKSWORK > 0); the 2011-2018 dictionaries state it as WORKYN = 1 and 2019+ as WKSWORK > 0, which coincide. - Make _domain_error total on mixed-type columns (sorted(map(str, ...))). - Validate I_NOEMP (0-9) and MARSUPWT (non-negative) domains. - Read only the eight required columns (usecols) instead of all ~800. Co-Authored-By: Claude Fable 5 --- src/populace_dynamics/data/asec_firm_size.py | 67 +++++++++++++------- tests/data/test_asec_firm_size.py | 37 +++++++++++ 2 files changed, 82 insertions(+), 22 deletions(-) diff --git a/src/populace_dynamics/data/asec_firm_size.py b/src/populace_dynamics/data/asec_firm_size.py index d938401d..a4258c73 100644 --- a/src/populace_dynamics/data/asec_firm_size.py +++ b/src/populace_dynamics/data/asec_firm_size.py @@ -182,7 +182,7 @@ def _check_path_year(path: Path, year: int) -> None: def _domain_error(year: int, column: str, bad: pd.Series) -> ValueError: - values = ", ".join(str(v) for v in sorted(bad.unique())[:8]) + values = ", ".join(sorted(map(str, bad.unique()))[:8]) return ValueError( f"ASEC {year} {column} contains out-of-dictionary value(s) " f"[{values}] on {len(bad)} row(s); refusing to band a file " @@ -218,10 +218,13 @@ def read_asec_firm_size( Raises: ValueError: On an unsupported year, a path/year mismatch, - missing required columns, or any value outside the - year's dictionary domain (NOEMP outside 0-6, LJCW - outside 0-7, or a nonzero NOEMP off the ``WKSWORK > 0`` - universe). + missing required columns, any value outside the year's + dictionary domain (NOEMP outside 0-6, LJCW outside 0-7, + a negative weight), or a universe violation: NOEMP and + LJCW share the longest-job universe in every dictionary + (stated as ``WORKYN = 1`` in 2011-2018 and + ``WKSWORK > 0`` in 2019+, which coincide), so each must + be nonzero exactly on the ``WKSWORK > 0`` rows. FileNotFoundError: If no staged person file can be found. """ bands = noemp_band_map(year) @@ -235,8 +238,16 @@ def read_asec_firm_size( person_path = _resolve_person_path(year, _resolve_data_dir(data_dir)) _check_path_year(person_path, year) - raw = pd.read_csv(person_path, usecols=None, low_memory=False) - missing = sorted(set(_REQUIRED_COLUMNS) - set(raw.columns)) + # PERIDNUM is a 22-digit identifier: wider than int64, and a + # float64 fallback would round distinct persons together, so it + # must never pass through a numeric dtype. + required = set(_REQUIRED_COLUMNS) + raw = pd.read_csv( + person_path, + usecols=lambda column: column in required, + dtype={"PERIDNUM": "string"}, + ) + missing = sorted(required - set(raw.columns)) if missing: raise ValueError( f"{person_path.name} is missing required column(s) " @@ -245,24 +256,36 @@ def read_asec_firm_size( ) raw = raw.loc[:, list(_REQUIRED_COLUMNS)].copy() - for column, low, high in ( - ("NOEMP", 0, 6), - ("LJCW", 0, 7), - ("WKSWORK", 0, 52), + for column, low, high, cast in ( + ("NOEMP", 0, 6, int), + ("LJCW", 0, 7, int), + ("WKSWORK", 0, 52, int), + ("I_NOEMP", 0, 9, int), + ("MARSUPWT", 0, None, float), ): values = pd.to_numeric(raw[column], errors="coerce") - bad = raw[column][values.isna() | (values < low) | (values > high)] + out_of_domain = values.isna() | (values < low) + if high is not None: + out_of_domain |= values > high + bad = raw[column][out_of_domain] if len(bad): raise _domain_error(year, column, bad) - raw[column] = values.astype(int) - - off_universe = raw[(raw["NOEMP"] > 0) & (raw["WKSWORK"] == 0)] - if len(off_universe): - raise ValueError( - f"ASEC {year}: {len(off_universe)} row(s) carry a nonzero " - "NOEMP outside the WKSWORK > 0 universe; the file does " - "not match the dictionary's universe statement." - ) + raw[column] = values.astype(cast) + + # NOEMP and LJCW share the longest-job universe in every + # dictionary year, so a zero inside WKSWORK > 0 (or a nonzero + # outside it) means the file does not match its dictionary. + for column in ("NOEMP", "LJCW"): + mismatch = raw[(raw[column] > 0) != (raw["WKSWORK"] > 0)] + if len(mismatch): + raise ValueError( + f"ASEC {year}: {len(mismatch)} row(s) violate the " + f"{column} universe ({column} must be nonzero " + "exactly where WKSWORK > 0); the file does not " + "match the dictionary's universe statement — " + "re-adjudicate against that year's dictionary " + "before extending the reader." + ) universe = raw[raw["WKSWORK"] > 0].reset_index(drop=True) return pd.DataFrame( @@ -275,7 +298,7 @@ def read_asec_firm_size( "firm_size_band": universe["NOEMP"].map( lambda code: bands.get(int(code), "niu") ), - "noemp_allocated": pd.to_numeric(universe["I_NOEMP"]) > 0, + "noemp_allocated": universe["I_NOEMP"] > 0, "ljcw": universe["LJCW"], "class_of_worker": universe["LJCW"].map( lambda code: CLASS_OF_WORKER_LABELS.get(int(code), "niu") diff --git a/tests/data/test_asec_firm_size.py b/tests/data/test_asec_firm_size.py index 1900115c..09f234ea 100644 --- a/tests/data/test_asec_firm_size.py +++ b/tests/data/test_asec_firm_size.py @@ -118,6 +118,42 @@ def test_noemp_outside_universe_raises(self, tmp_path): with pytest.raises(ValueError, match="universe"): asec_firm_size.read_asec_firm_size(2021, path=path) + def test_zero_noemp_inside_universe_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"NOEMP": 0}]) + with pytest.raises(ValueError, match="NOEMP universe"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_zero_ljcw_inside_universe_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"LJCW": 0}]) + with pytest.raises(ValueError, match="LJCW universe"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_mixed_type_domain_violation_stays_a_value_error(self, tmp_path): + rows = [{"NOEMP": "A"}, {"NOEMP": 7}] + path = _write_person_file(tmp_path, 2021, rows) + with pytest.raises(ValueError, match="out-of-dictionary"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_negative_weight_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"MARSUPWT": -1.0}]) + with pytest.raises(ValueError, match="MARSUPWT"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_wide_person_ids_survive_exactly(self, tmp_path): + # 22-digit PERIDNUMs differing only in the last digit round + # together under any numeric read (int64 tops out at 19 + # digits; float64 keeps ~15-16), so exact string survival + # is the regression test for the dtype pin. + ids = [ + "8812345678901234567891", + "8812345678901234567892", + ] + path = _write_person_file( + tmp_path, 2021, [{"PERIDNUM": i} for i in ids] + ) + out = asec_firm_size.read_asec_firm_size(2021, path=path) + assert list(out["person_id"]) == ids + def test_universe_and_flags(self, tmp_path): rows = [ {"NOEMP": 1, "I_NOEMP": 1, "LJCW": 6}, @@ -184,6 +220,7 @@ def test_reads_any_staged_year(self): year = 2000 + int(staged[-1].name[5:7]) out = asec_firm_size.read_asec_firm_size(year, path=staged[-1]) assert len(out) > 10_000 + assert out["person_id"].is_unique assert set(out["firm_size_band"]) <= set( asec_firm_size.noemp_band_map(year).values() ) From 9e0754e6b13afe86fcd06b053c6a45d8c568c492 Mon Sep 17 00:00:00 2001 From: Daphne Hansell <128793799+daphnehanse11@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:10:10 -0400 Subject: [PATCH 3/3] Harden the loader against silent-wrong-output seams (round-2 review) - Refuse blank or duplicated PERIDNUM (the join key now gets the same refuse-on-mismatch treatment as every coded column; NaN ids vanish silently from downstream groupbys). - Reject non-integral code values (2.9 truncated to band 10_24 with no error) and non-finite weights. - Move WEIND/INDUSTRY into the domain loop: friendly errors instead of raw pandas cast failures, and validation no longer depends on the universe filter. - firm_size_tabulation: keep NaN group keys (dropna=False), NaN allocated_share for zero-weight groups, documented. - Friendly message for an empty staged file. Co-Authored-By: Claude Fable 5 --- src/populace_dynamics/data/asec_firm_size.py | 65 +++++++++++++++----- tests/data/test_asec_firm_size.py | 46 ++++++++++++++ 2 files changed, 96 insertions(+), 15 deletions(-) diff --git a/src/populace_dynamics/data/asec_firm_size.py b/src/populace_dynamics/data/asec_firm_size.py index a4258c73..e4ce25d1 100644 --- a/src/populace_dynamics/data/asec_firm_size.py +++ b/src/populace_dynamics/data/asec_firm_size.py @@ -47,6 +47,7 @@ import re from pathlib import Path +import numpy as np import pandas as pd __all__ = [ @@ -242,11 +243,16 @@ def read_asec_firm_size( # float64 fallback would round distinct persons together, so it # must never pass through a numeric dtype. required = set(_REQUIRED_COLUMNS) - raw = pd.read_csv( - person_path, - usecols=lambda column: column in required, - dtype={"PERIDNUM": "string"}, - ) + try: + raw = pd.read_csv( + person_path, + usecols=lambda column: column in required, + dtype={"PERIDNUM": "string"}, + ) + except pd.errors.EmptyDataError: + raise ValueError( + f"{person_path.name} is empty; {_README_POINTER}." + ) from None missing = sorted(required - set(raw.columns)) if missing: raise ValueError( @@ -256,17 +262,43 @@ def read_asec_firm_size( ) raw = raw.loc[:, list(_REQUIRED_COLUMNS)].copy() + # The join key gets the same refuse-on-mismatch treatment as + # every coded column: a blank or duplicated PERIDNUM is exactly + # what a botched fixed-width conversion produces, and NaN ids + # vanish silently from downstream groupbys. + ids = raw["PERIDNUM"] + blank = ids.isna() | (ids.astype(str).str.strip() == "") + if blank.any(): + raise ValueError( + f"ASEC {year}: {int(blank.sum())} row(s) have a blank " + "PERIDNUM; refusing a file with unusable person ids." + ) + duplicated = ids[ids.duplicated()] + if len(duplicated): + raise ValueError( + f"ASEC {year}: PERIDNUM is not unique " + f"({len(duplicated)} duplicated id(s), e.g. " + f"{duplicated.iloc[0]!r}); refusing a file with a " + "corrupted person-id column." + ) + for column, low, high, cast in ( ("NOEMP", 0, 6, int), ("LJCW", 0, 7, int), ("WKSWORK", 0, 52, int), ("I_NOEMP", 0, 9, int), + ("WEIND", 0, 23, int), + ("INDUSTRY", 0, 9999, int), ("MARSUPWT", 0, None, float), ): values = pd.to_numeric(raw[column], errors="coerce") - out_of_domain = values.isna() | (values < low) + out_of_domain = values.isna() | (values < low) | ~np.isfinite(values) if high is not None: out_of_domain |= values > high + if cast is int: + # Dictionary codes are integers; 2.9 is not a code and + # must not silently truncate into one. + out_of_domain |= values != values.round() bad = raw[column][out_of_domain] if len(bad): raise _domain_error(year, column, bad) @@ -303,12 +335,10 @@ def read_asec_firm_size( "class_of_worker": universe["LJCW"].map( lambda code: CLASS_OF_WORKER_LABELS.get(int(code), "niu") ), - "industry_major": pd.to_numeric(universe["WEIND"]).astype(int), - "industry_detailed": pd.to_numeric(universe["INDUSTRY"]).astype( - int - ), + "industry_major": universe["WEIND"], + "industry_detailed": universe["INDUSTRY"], "wkswork": universe["WKSWORK"], - "weight": pd.to_numeric(universe["MARSUPWT"]).astype(float), + "weight": universe["MARSUPWT"], } ) @@ -332,8 +362,9 @@ def firm_size_tabulation( Returns: One row per group with ``weighted_persons`` (MARSUPWT sum), ``unweighted_n``, and ``allocated_share`` (weighted share of - the group whose NOEMP was allocated/edited), sorted by the - grouping columns. + the group whose NOEMP was allocated/edited; NaN for a group + whose weights sum to zero), sorted by the grouping columns. + NaN group keys are kept, never silently dropped. """ missing = sorted( (set(by) | {"weight", "noemp_allocated"}) - set(records.columns) @@ -346,11 +377,15 @@ def firm_size_tabulation( working = records.assign( _allocated_weight=records["weight"] * records["noemp_allocated"] ) - grouped = working.groupby(list(by), sort=True) + grouped = working.groupby(list(by), sort=True, dropna=False) out = grouped.agg( weighted_persons=("weight", "sum"), unweighted_n=("weight", "size"), _allocated_weight=("_allocated_weight", "sum"), ).reset_index() - out["allocated_share"] = out["_allocated_weight"] / out["weighted_persons"] + out["allocated_share"] = np.where( + out["weighted_persons"] > 0, + out["_allocated_weight"] / out["weighted_persons"], + np.nan, + ) return out.drop(columns="_allocated_weight") diff --git a/tests/data/test_asec_firm_size.py b/tests/data/test_asec_firm_size.py index 09f234ea..99377518 100644 --- a/tests/data/test_asec_firm_size.py +++ b/tests/data/test_asec_firm_size.py @@ -139,6 +139,38 @@ def test_negative_weight_raises(self, tmp_path): with pytest.raises(ValueError, match="MARSUPWT"): asec_firm_size.read_asec_firm_size(2021, path=path) + def test_blank_person_id_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"PERIDNUM": ""}]) + with pytest.raises(ValueError, match="blank PERIDNUM"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_duplicate_person_id_raises(self, tmp_path): + rows = [{"PERIDNUM": 77}, {"PERIDNUM": 77}] + path = _write_person_file(tmp_path, 2021, rows) + with pytest.raises(ValueError, match="not unique"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_fractional_code_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"NOEMP": 2.9}]) + with pytest.raises(ValueError, match="NOEMP"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_industry_garbage_gets_friendly_error(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"INDUSTRY": "XX"}]) + with pytest.raises(ValueError, match="INDUSTRY"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_infinite_weight_raises(self, tmp_path): + path = _write_person_file(tmp_path, 2021, [{"MARSUPWT": "inf"}]) + with pytest.raises(ValueError, match="MARSUPWT"): + asec_firm_size.read_asec_firm_size(2021, path=path) + + def test_empty_file_gets_friendly_error(self, tmp_path): + path = tmp_path / "pppub21.csv" + path.write_text("") + with pytest.raises(ValueError, match="empty"): + asec_firm_size.read_asec_firm_size(2021, path=path) + def test_wide_person_ids_survive_exactly(self, tmp_path): # 22-digit PERIDNUMs differing only in the last digit round # together under any numeric read (int64 tops out at 19 @@ -206,6 +238,20 @@ def test_regime_break_is_visible_across_years(self, tmp_path): out = asec_firm_size.firm_size_tabulation(pd.concat(frames)) assert set(out["firm_size_band"]) == {"10_49", "10_24"} + def test_zero_weight_group_share_is_nan(self, tmp_path): + path = _write_person_file(tmp_path, 2024, [{"MARSUPWT": 0.0}]) + records = asec_firm_size.read_asec_firm_size(2024, path=path) + out = asec_firm_size.firm_size_tabulation(records) + assert out.iloc[0]["weighted_persons"] == 0.0 + assert pd.isna(out.iloc[0]["allocated_share"]) + + def test_nan_group_keys_are_kept(self, tmp_path): + path = _write_person_file(tmp_path, 2024, [{}, {}]) + records = asec_firm_size.read_asec_firm_size(2024, path=path) + records.loc[0, "class_of_worker"] = None + out = asec_firm_size.firm_size_tabulation(records) + assert out["weighted_persons"].sum() == records["weight"].sum() + def test_wrong_frame_raises(self): with pytest.raises(ValueError, match="read_asec_firm_size"): asec_firm_size.firm_size_tabulation(pd.DataFrame({"x": [1]}))