diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bff7bc2c..ef417acc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -112,11 +112,10 @@ jobs: - name: Test OpenTimestamps anchoring run: uv run pytest -q --durations=30 tests/test_ots_anchor.py - - name: Build source input database + - name: Smoke-test the source-artifact database CLI run: | - uv run chronicle --db /tmp/chronicle-targets-ci.db init - uv run chronicle --db /tmp/chronicle-targets-ci.db load all - uv run chronicle --db /tmp/chronicle-targets-ci.db stats + uv run chronicle --db /tmp/chronicle-source-ci.db init + uv run chronicle --db /tmp/chronicle-source-ci.db source-stats - name: Build wheel run: | @@ -134,6 +133,7 @@ jobs: import policyengine_chronicle.normalization import policyengine_chronicle.sources import policyengine_chronicle.targets + from policyengine_chronicle.targets import hard_target_package_aliases from policyengine_chronicle.schema import ( CONSUMER_FACT_SCHEMA_SHA256, CONSUMER_FACT_SCHEMA_SHA256_BY_VERSION, @@ -141,6 +141,9 @@ jobs: ) assert policyengine_chronicle.__name__ == "policyengine_chronicle" + # Target source coverage ships; the legacy target tables do not. + assert "soi-table-1-1" in hard_target_package_aliases() + assert not hasattr(policyengine_chronicle.targets, "query_targets") # Every consumer-fact contract must ship inside the wheel: frozen v1, # the v2 labels contract, the v3 geography-name contract, and the v4 # contract rows are emitted under. diff --git a/README.md b/README.md index d62c5f28..2d053b9d 100644 --- a/README.md +++ b/README.md @@ -108,12 +108,12 @@ chronicle/ │ ├── sources/ # Source lineage helpers │ ├── facts/ # Source-backed facts │ ├── normalization/ # Low-assumption representation helpers -│ ├── targets/ # Target input schema, client, loaders +│ ├── targets/ # Target source coverage (US poverty/nonfiler) │ ├── jurisdictions/ # Temporary in-repo jurisdiction source prototypes -├── db/ # SQLModel persistence and source loaders -│ ├── schema.py # SQLModel: Target, Stratum, StratumConstraint +├── db/ # Source-artifact store and packaged source files +│ ├── schema.py # SQLModel: SourceArtifact, SourceTable, SourceRow │ ├── supabase_client.py # Supabase client helpers -│ └── etl_*.py # Source-specific ETL pipelines +│ └── data/ # Publisher files that source packages read ├── data/ # Cached data files └── docs/ # Architecture and source documentation ``` @@ -127,23 +127,16 @@ Solver execution and calibrated dataset construction belong in Microcosm. ### 1. Install +Chronicle is not published on PyPI. Clone this repository into a +Chronicle-named directory and install it with its development extras: + ```bash -pip install policyengine-chronicle -# Or for development, clone this repository into a Chronicle-named directory: git clone chronicle cd chronicle -pip install -e ".[dev]" -``` - -### 2. Initialize and Load Legacy Target Inputs - -```bash -chronicle init -chronicle load soi --years 2021 -chronicle stats +uv sync --all-extras ``` -### 3. Validate Fixture Facts +### 2. Validate Fixture Facts The standalone Chronicle fact harness validates JSONL aggregate facts and emits a JSON report with fact counts, QA counts, warnings, and validation errors: @@ -516,7 +509,7 @@ same `--axiom-cli` and `--axiom-root` flags, plus `--require-axiom-validation` when skipped concept checks should fail agent acceptance. -### 4. Run Chronicle Explorer +### 3. Run Chronicle Explorer Chronicle Explorer is a Next/Tailwind app that reads the fixture fact JSONL and source-cell JSONL, then shows aggregate facts, source-cell lineage, and @@ -538,25 +531,19 @@ By default, the workbench reads the current local suite outputs at LEDGER_EXPLORER_DATA_DIRS=/tmp/chronicle-build-a,/tmp/chronicle-build-b npm run dev -- --port 3090 ``` -### 5. Query Target Inputs in Python - -```python -from policyengine_chronicle.targets import DataSource, Target, TargetType, query_targets - -target_rows = query_targets(jurisdiction="us", year=2024) -``` - -## Target Input Schema - -Target inputs use a three-table schema: - -- **strata**: Population subgroups, such as California filers with AGI between - $50k and $75k. -- **stratum_constraints**: Rules defining each stratum. -- **targets**: Source-published aggregate values linked to strata. +## Retired legacy target tables -These are source-backed inputs. Microcosm owns the contracts that select them, -the active support-aware subset, and calibrated solver execution. +Chronicle used to load hand-written ETLs into a local SQLite `targets` / +`strata` / `stratum_constraints` schema (`chronicle load`, `chronicle stats`, +`chronicle query`) and to read a Supabase copy through +`policyengine_chronicle.targets.query_targets`. No live consumer read those +tables, and most of their values did not trace to publisher bytes, so they were +retired. +Source-backed facts reach consumers through source packages, the bundle and +the consumer-fact export. Microcosm owns the contracts that select them, the +active support-aware subset, and calibrated solver execution. See +[`docs/us-legacy-etl-cleanup.md`](docs/us-legacy-etl-cleanup.md) for what each +retired loader held and which package publishes it now. ## Identifier Epochs @@ -623,7 +610,7 @@ normalized_fact = convert_units(fact, 1000, "count") ## Current Coverage -### Aggregate Facts And Target Inputs +### Aggregate facts | Source | Coverage | Description | |--------|----------|-------------| diff --git a/changelog.d/legacy-targets-table-retired.removed.md b/changelog.d/legacy-targets-table-retired.removed.md new file mode 100644 index 00000000..06f01805 --- /dev/null +++ b/changelog.d/legacy-targets-table-retired.removed.md @@ -0,0 +1,22 @@ +Retire the legacy SQLite target tables (`targets`, `strata`, +`stratum_constraints`) and everything that wrote or read them: +- the `chronicle load`, `chronicle stats` and `chronicle query` commands +- the last two loaders, `db/etl_soi.py` and `db/etl_snap.py` +- the Supabase `query_targets` / `query_strata` / `insert_targets_batch` helpers +- `scripts/export_to_json.py` +- the `as_target` / `target_kwargs` / `TargetBlueprint` normalization helpers +- the legacy names in `chronicle.targets` and `policyengine_chronicle.targets` + +Nothing reads these tables today. The values the last two loaders wrote (IRS +SOI Tables 1.1 and 1.4, and SNAP) are published as source-package facts through +the bundle and consumer-fact export, which are unchanged. + +`chronicle init` now creates only the source-artifact tables. The AGI-bracket +tables the SOI harness used move into `chronicle/jurisdictions/us/soi.py`. +`policyengine_chronicle.targets` keeps the US poverty/nonfiler target source +coverage. + +Migration: Chronicle was never published on PyPI, and no live repository +imports these names. Archived Cosilico-era repositories once did. Code that called `query_targets` should read consumer facts from a +pinned bundle (`chronicle build-bundle`, `chronicle build-consumer-artifact`) +instead. diff --git a/chronicle/facts/models.py b/chronicle/facts/models.py index dba40441..49c16df6 100644 --- a/chronicle/facts/models.py +++ b/chronicle/facts/models.py @@ -20,7 +20,7 @@ class DerivationStep: @dataclass(frozen=True) class SourceFact: - """A source-backed fact before target-input materialization.""" + """A source-backed fact with its structural normalization lineage.""" name: str value: float diff --git a/chronicle/jurisdictions/us/__init__.py b/chronicle/jurisdictions/us/__init__.py index 45d40c57..898cca94 100644 --- a/chronicle/jurisdictions/us/__init__.py +++ b/chronicle/jurisdictions/us/__init__.py @@ -1,11 +1,7 @@ -"""United States Chronicle target loaders.""" +"""United States Chronicle source builders. -from chronicle.targets.loaders import ( - load_snap_targets, - load_soi_targets, -) +US measures are source packages. The legacy DB target loaders were retired +(see docs/us-legacy-etl-cleanup.md), so this module exports no loaders. +""" -__all__ = [ - "load_snap_targets", - "load_soi_targets", -] +__all__: list[str] = [] diff --git a/chronicle/jurisdictions/us/soi.py b/chronicle/jurisdictions/us/soi.py index f1196fe2..be1973c9 100644 --- a/chronicle/jurisdictions/us/soi.py +++ b/chronicle/jurisdictions/us/soi.py @@ -24,7 +24,6 @@ SourceRecordSpec, ) from chronicle.yaml_io import safe_load as safe_load_yaml -from db.etl_soi import AGI_BRACKETS, TABLE_1_1_AGI_LABEL_TO_BRACKET SOI_TABLE_1_1_SOURCE_NAME = "irs_soi" SOI_TABLE_1_1_SOURCE_TABLE = "Publication 1304 Table 1.1" @@ -48,6 +47,51 @@ "tax-year source record." ) +# AGI bracket definitions (lower, upper) in dollars. +AGI_BRACKETS = { + "under_1": (float("-inf"), 1), + "1_to_5k": (1, 5_000), + "5k_to_10k": (5_000, 10_000), + "10k_to_15k": (10_000, 15_000), + "15k_to_20k": (15_000, 20_000), + "20k_to_25k": (20_000, 25_000), + "25k_to_30k": (25_000, 30_000), + "30k_to_40k": (30_000, 40_000), + "40k_to_50k": (40_000, 50_000), + "50k_to_75k": (50_000, 75_000), + "75k_to_100k": (75_000, 100_000), + "100k_to_200k": (100_000, 200_000), + "200k_to_500k": (200_000, 500_000), + "500k_to_1m": (500_000, 1_000_000), + "1m_to_1_5m": (1_000_000, 1_500_000), + "1_5m_to_2m": (1_500_000, 2_000_000), + "2m_to_5m": (2_000_000, 5_000_000), + "5m_to_10m": (5_000_000, 10_000_000), + "10m_plus": (10_000_000, float("inf")), +} + +TABLE_1_1_AGI_LABEL_TO_BRACKET = { + "No adjusted gross income": "under_1", + "$1 under $5,000": "1_to_5k", + "$5,000 under $10,000": "5k_to_10k", + "$10,000 under $15,000": "10k_to_15k", + "$15,000 under $20,000": "15k_to_20k", + "$20,000 under $25,000": "20k_to_25k", + "$25,000 under $30,000": "25k_to_30k", + "$30,000 under $40,000": "30k_to_40k", + "$40,000 under $50,000": "40k_to_50k", + "$50,000 under $75,000": "50k_to_75k", + "$75,000 under $100,000": "75k_to_100k", + "$100,000 under $200,000": "100k_to_200k", + "$200,000 under $500,000": "200k_to_500k", + "$500,000 under $1,000,000": "500k_to_1m", + "$1,000,000 under $1,500,000": "1m_to_1_5m", + "$1,500,000 under $2,000,000": "1_5m_to_2m", + "$2,000,000 under $5,000,000": "2m_to_5m", + "$5,000,000 under $10,000,000": "5m_to_10m", + "$10,000,000 or more": "10m_plus", +} + def build_soi_table_1_1_facts(year: int = 2023) -> list[AggregateFact]: """Build source-lineaged Chronicle facts from SOI Table 1.1.""" diff --git a/chronicle/normalization/__init__.py b/chronicle/normalization/__init__.py index 88904c86..0f87a091 100644 --- a/chronicle/normalization/__init__.py +++ b/chronicle/normalization/__init__.py @@ -7,24 +7,18 @@ """ from chronicle.facts import DerivationStep, SourceFact -from .targets import ( - TargetBlueprint, +from .transforms import ( apply_share, - as_target, convert_units, format_derivation, scale_value, - target_kwargs, ) __all__ = [ "DerivationStep", "SourceFact", - "TargetBlueprint", "apply_share", - "as_target", "convert_units", "format_derivation", "scale_value", - "target_kwargs", ] diff --git a/chronicle/normalization/targets.py b/chronicle/normalization/targets.py deleted file mode 100644 index bc423f24..00000000 --- a/chronicle/normalization/targets.py +++ /dev/null @@ -1,197 +0,0 @@ -"""Helpers for source facts and target inputs. - -These helpers are for low-assumption structural normalization: unit/scale -standardization and same-source arithmetic with explicit lineage. Modeling -choices such as inflation, aging, target reconciliation, or simulator-specific -allocation belong in downstream calibration recipes. -""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - -from chronicle.facts import DerivationStep, SourceFact -from chronicle.targets import DataSource, GeographicLevel, Jurisdiction, TargetType - -Constraint = tuple[str, str, str] - - -@dataclass(frozen=True) -class TargetBlueprint: - """A target input ready for a loader to materialize.""" - - variable: str - value: float - period: int - target_type: TargetType - source: DataSource - jurisdiction: Jurisdiction - stratum_name: str - constraints: tuple[Constraint, ...] = () - geographic_level: GeographicLevel | None = None - source_table: str | None = None - source_url: str | None = None - notes: str | None = None - is_preliminary: bool = False - margin_of_error: float | None = None - derivation: tuple[DerivationStep, ...] = () - - -def scale_value( - fact: SourceFact, - factor: float, - *, - name: str | None = None, - unit: str | None = None, - operation: str = "scale_value", - note: str | None = None, -) -> SourceFact: - """Scale a published fact value while recording normalization lineage.""" - parameters: dict[str, Any] = { - "factor": factor, - "input_unit": fact.unit, - "output_unit": unit if unit is not None else fact.unit, - } - if note: - parameters["note"] = note - - return fact.with_step( - DerivationStep(operation=operation, parameters=parameters), - name=name, - value=fact.value * factor, - unit=unit, - ) - - -def convert_units( - fact: SourceFact, - factor: float, - to_unit: str, - *, - name: str | None = None, - note: str | None = None, -) -> SourceFact: - """Normalize a published unit or scale with a multiplicative factor.""" - return scale_value( - fact, - factor, - name=name, - unit=to_unit, - operation="convert_units", - note=note, - ) - - -def apply_share( - total_fact: SourceFact, - share: float, - *, - name: str, - share_name: str | None = None, - unit: str | None = None, -) -> SourceFact: - """Materialize a same-source total/share relationship. - - Use this only when the source publishes both the total and the share. Do - not use it for model-driven allocation or target reconciliation. - """ - return total_fact.with_step( - DerivationStep( - operation="apply_share", - parameters={ - "share": share, - "share_name": share_name, - "total_fact": total_fact.name, - }, - ), - name=name, - value=total_fact.value * share, - unit=unit, - ) - - -def as_target( - fact: SourceFact, - *, - variable: str | None = None, - target_type: TargetType = TargetType.COUNT, - source: DataSource | str | None = None, - jurisdiction: Jurisdiction | str | None = None, - stratum_name: str, - constraints: tuple[Constraint, ...] = (), - geographic_level: GeographicLevel | None = None, - notes: str | None = None, -) -> TargetBlueprint: - """Convert a source fact into a target input blueprint.""" - target_source = _coerce_data_source(source if source is not None else fact.source) - target_jurisdiction = _coerce_jurisdiction( - jurisdiction if jurisdiction is not None else fact.jurisdiction - ) - - derivation_note = format_derivation(fact.derivation) - if notes and derivation_note: - notes = f"{notes} Derived via: {derivation_note}." - elif derivation_note: - notes = f"Derived via: {derivation_note}." - - return TargetBlueprint( - variable=variable if variable is not None else fact.name, - value=fact.value, - period=fact.period, - target_type=target_type, - source=target_source, - jurisdiction=target_jurisdiction, - stratum_name=stratum_name, - constraints=constraints, - geographic_level=geographic_level, - source_table=fact.source_table, - source_url=fact.source_url, - notes=notes, - is_preliminary=fact.is_preliminary, - margin_of_error=fact.margin_of_error, - derivation=fact.derivation, - ) - - -def target_kwargs(blueprint: TargetBlueprint, stratum_id: int) -> dict[str, Any]: - """Return keyword arguments for constructing a target DB row.""" - return { - "stratum_id": stratum_id, - "variable": blueprint.variable, - "period": blueprint.period, - "value": blueprint.value, - "target_type": blueprint.target_type, - "geographic_level": blueprint.geographic_level, - "source": blueprint.source, - "source_table": blueprint.source_table, - "source_url": blueprint.source_url, - "notes": blueprint.notes, - "is_preliminary": blueprint.is_preliminary, - "margin_of_error": blueprint.margin_of_error, - } - - -def format_derivation(steps: tuple[DerivationStep, ...]) -> str: - """Format derivation steps for target notes or diagnostics.""" - parts = [] - for step in steps: - params = ", ".join(f"{key}={value}" for key, value in step.parameters.items()) - parts.append(f"{step.operation}({params})" if params else step.operation) - return "; ".join(parts) - - -def _coerce_data_source(source: DataSource | str | None) -> DataSource: - if source is None: - raise ValueError("Target source must be supplied by the fact or caller") - if isinstance(source, DataSource): - return source - return DataSource(source) - - -def _coerce_jurisdiction(jurisdiction: Jurisdiction | str | None) -> Jurisdiction: - if jurisdiction is None: - raise ValueError("Target jurisdiction must be supplied by the fact or caller") - if isinstance(jurisdiction, Jurisdiction): - return jurisdiction - return Jurisdiction(jurisdiction) diff --git a/chronicle/normalization/transforms.py b/chronicle/normalization/transforms.py new file mode 100644 index 00000000..db9b8332 --- /dev/null +++ b/chronicle/normalization/transforms.py @@ -0,0 +1,95 @@ +"""Lineage-recording transforms for source facts. + +These helpers are for low-assumption structural normalization: unit/scale +standardization and same-source arithmetic with explicit lineage. Modeling +choices such as inflation, aging, target reconciliation, or simulator-specific +allocation belong in downstream calibration recipes. +""" + +from __future__ import annotations + +from typing import Any + +from chronicle.facts import DerivationStep, SourceFact + + +def scale_value( + fact: SourceFact, + factor: float, + *, + name: str | None = None, + unit: str | None = None, + operation: str = "scale_value", + note: str | None = None, +) -> SourceFact: + """Scale a published fact value while recording normalization lineage.""" + parameters: dict[str, Any] = { + "factor": factor, + "input_unit": fact.unit, + "output_unit": unit if unit is not None else fact.unit, + } + if note: + parameters["note"] = note + + return fact.with_step( + DerivationStep(operation=operation, parameters=parameters), + name=name, + value=fact.value * factor, + unit=unit, + ) + + +def convert_units( + fact: SourceFact, + factor: float, + to_unit: str, + *, + name: str | None = None, + note: str | None = None, +) -> SourceFact: + """Normalize a published unit or scale with a multiplicative factor.""" + return scale_value( + fact, + factor, + name=name, + unit=to_unit, + operation="convert_units", + note=note, + ) + + +def apply_share( + total_fact: SourceFact, + share: float, + *, + name: str, + share_name: str | None = None, + unit: str | None = None, +) -> SourceFact: + """Materialize a same-source total/share relationship. + + Use this only when the source publishes both the total and the share. Do + not use it for model-driven allocation or target reconciliation. + """ + return total_fact.with_step( + DerivationStep( + operation="apply_share", + parameters={ + "share": share, + "share_name": share_name, + "total_fact": total_fact.name, + }, + ), + name=name, + value=total_fact.value * share, + unit=unit, + ) + + +def format_derivation(steps: tuple[DerivationStep, ...]) -> str: + """Format derivation steps for notes or diagnostics.""" + parts = [] + for step in steps: + params = ", ".join(f"{key}={value}" for key, value in step.parameters.items()) + parts.append(f"{step.operation}({params})" if params else step.operation) + return "; ".join(parts) diff --git a/chronicle/targets/__init__.py b/chronicle/targets/__init__.py index 7240b537..ce157c4c 100644 --- a/chronicle/targets/__init__.py +++ b/chronicle/targets/__init__.py @@ -1,19 +1,10 @@ -"""Chronicle target-input schema and query helpers.""" +"""Chronicle target source coverage. + +Chronicle records which source packages back a consumer's target families. The +legacy SQLite and Supabase target tables (``Target``, ``Stratum``, +``StratumConstraint``) were retired; see ``docs/us-legacy-etl-cleanup.md``. +""" -from db.schema import ( - DEFAULT_DB_PATH, - DataSource, - GeographicLevel, - Jurisdiction, - Stratum, - StratumConstraint, - Target, - TargetType, - get_engine, - get_session, - init_db, -) -from db.supabase_client import insert_targets_batch, query_strata, query_targets from .us_poverty import ( US_POVERTY_NONFILER_TARGET_COVERAGE, TargetSourceCoverage, @@ -25,24 +16,10 @@ ) __all__ = [ - "DEFAULT_DB_PATH", - "DataSource", - "GeographicLevel", - "Jurisdiction", - "Stratum", - "StratumConstraint", - "Target", "TargetSourceCoverage", - "TargetType", "US_POVERTY_NONFILER_TARGET_COVERAGE", "coverage_entries", - "get_engine", - "get_session", "hard_target_package_aliases", - "init_db", - "insert_targets_batch", - "query_strata", - "query_targets", "source_gap_family_ids", "validate_us_poverty_nonfiler_source_coverage", "validation_only_family_ids", diff --git a/chronicle/targets/client.py b/chronicle/targets/client.py deleted file mode 100644 index 4cbe8a92..00000000 --- a/chronicle/targets/client.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Supabase helpers for Chronicle target inputs.""" - -from db.supabase_client import insert_targets_batch, query_strata, query_targets - -__all__ = ["insert_targets_batch", "query_strata", "query_targets"] diff --git a/chronicle/targets/loaders.py b/chronicle/targets/loaders.py deleted file mode 100644 index c1f0dcb4..00000000 --- a/chronicle/targets/loaders.py +++ /dev/null @@ -1,9 +0,0 @@ -"""ETL loaders that populate Chronicle target inputs.""" - -from db.etl_snap import load_snap_targets -from db.etl_soi import load_soi_targets - -__all__ = [ - "load_snap_targets", - "load_soi_targets", -] diff --git a/db/cli.py b/db/cli.py index 15d0add6..2e921530 100644 --- a/db/cli.py +++ b/db/cli.py @@ -1,4 +1,4 @@ -"""CLI for managing Chronicle target input data.""" +"""CLI for Chronicle's local source-artifact database.""" import argparse from pathlib import Path @@ -6,14 +6,11 @@ from sqlalchemy import func from sqlmodel import Session, select -from .etl_soi import load_soi_targets from .schema import ( DEFAULT_DB_PATH, SourceArtifact, SourceRow, SourceTable, - Stratum, - Target, init_db, get_engine, ) @@ -26,51 +23,6 @@ def cmd_init(args): print(f"Initialized database at {db_path}") -def cmd_load(args): - """Load targets from a source.""" - db_path = Path(args.db) if args.db else DEFAULT_DB_PATH - engine = init_db(db_path) - - with Session(engine) as session: - if args.source == "soi" or args.source == "all": - years = [int(y) for y in args.years.split(",")] if args.years else None - load_soi_targets(session, years=years) - print(f"Loaded SOI targets for years: {years or 'all available'}") - - if args.source == "snap" or args.source == "all": - from .etl_snap import load_snap_targets - - years = [int(y) for y in args.years.split(",")] if args.years else None - load_snap_targets(session, years=years) - print(f"Loaded SNAP targets for years: {years or 'all available'}") - - -def cmd_stats(args): - """Show database statistics.""" - db_path = Path(args.db) if args.db else DEFAULT_DB_PATH - - if not db_path.exists(): - print(f"Database not found at {db_path}") - return - - engine = get_engine(db_path) - with Session(engine) as session: - strata_count = len(session.exec(select(Stratum)).all()) - targets_count = len(session.exec(select(Target)).all()) - - # Get unique sources - sources = session.exec(select(Target.source).distinct()).all() - - # Get unique years - years = session.exec(select(Target.period).distinct()).all() - - print(f"Database: {db_path}") - print(f"Strata: {strata_count}") - print(f"Targets: {targets_count}") - print(f"Sources: {', '.join(str(s) for s in sources)}") - print(f"Years: {', '.join(str(y) for y in sorted(years))}") - - def cmd_load_source_files(args): """Load full source files into the source artifact tables.""" db_path = Path(args.db) if args.db else DEFAULT_DB_PATH @@ -148,41 +100,6 @@ def cmd_source_stats(args): print(f" {source_id}: {count:,}") -def cmd_query(args): - """Query targets.""" - db_path = Path(args.db) if args.db else DEFAULT_DB_PATH - - if not db_path.exists(): - print(f"Database not found at {db_path}") - return - - engine = get_engine(db_path) - with Session(engine) as session: - query = select(Target, Stratum).join(Stratum) - - if args.variable: - query = query.where(Target.variable == args.variable) - if args.year: - query = query.where(Target.period == int(args.year)) - if args.source: - query = query.where(Target.source == args.source) - - results = session.exec(query).all() - - if not results: - print("No targets found matching criteria") - return - - print( - f"{'Stratum':<30} {'Variable':<25} {'Year':<6} {'Value':>15} {'Source':<10}" - ) - print("-" * 90) - for target, stratum in results[: args.limit]: - print( - f"{stratum.name:<30} {target.variable:<25} {target.period:<6} {target.value:>15,.0f} {target.source.value:<10}" - ) - - def _pe_source_root_env_default(jurisdiction: str) -> str | None: from .pe_source_inventory import ( PE_UK_DATA_ROOT_ENV, @@ -195,7 +112,9 @@ def _pe_source_root_env_default(jurisdiction: str) -> str | None: def main(): - parser = argparse.ArgumentParser(description="Manage Chronicle target input data") + parser = argparse.ArgumentParser( + description="Manage Chronicle's local source-artifact database" + ) parser.add_argument("--db", help=f"Database path (default: {DEFAULT_DB_PATH})") subparsers = parser.add_subparsers(dest="command", required=True) @@ -204,26 +123,6 @@ def main(): init_parser = subparsers.add_parser("init", help="Initialize database") init_parser.set_defaults(func=cmd_init) - # load - load_parser = subparsers.add_parser("load", help="Load targets from source") - load_parser.add_argument( - "source", - choices=[ - "soi", - "snap", - "all", - ], - help="Data source to load", - ) - load_parser.add_argument( - "--years", help="Comma-separated years to load (default: all)" - ) - load_parser.set_defaults(func=cmd_load) - - # stats - stats_parser = subparsers.add_parser("stats", help="Show database statistics") - stats_parser.set_defaults(func=cmd_stats) - # load-source-files source_parser = subparsers.add_parser( "load-source-files", @@ -263,14 +162,6 @@ def main(): ) source_stats_parser.set_defaults(func=cmd_source_stats) - # query - query_parser = subparsers.add_parser("query", help="Query targets") - query_parser.add_argument("--variable", help="Filter by variable name") - query_parser.add_argument("--year", help="Filter by year") - query_parser.add_argument("--source", help="Filter by source") - query_parser.add_argument("--limit", type=int, default=20, help="Max results") - query_parser.set_defaults(func=cmd_query) - args = parser.parse_args() args.func(args) diff --git a/db/etl_snap.py b/db/etl_snap.py deleted file mode 100644 index 783e24d5..00000000 --- a/db/etl_snap.py +++ /dev/null @@ -1,303 +0,0 @@ -"""ETL for USDA SNAP (Supplemental Nutrition Assistance Program) targets. - -Loads SNAP participation and benefit targets into the legacy targets database. - -Values are sourced from the trustworthy USDA FNS source package at -``packages/usda_snap/fy69_to_current``, which parses the FY24 workbook -(``SNAP Monthly State Participation and Benefit Summary``) cell-by-cell with -guard cells and checksum-locked provenance. This ETL previously hardcoded a -``SNAP_DATA`` dict of suspiciously round per-state values with no per-value -citation (see PolicyEngine/chronicle#77); those fabricated numbers have been -removed in favor of the parsed source facts, so ``chronicle load snap`` now emits -only source-backed values. - -Data source page: -https://www.fns.usda.gov/pd/supplemental-nutrition-assistance-program-snap -""" - -from __future__ import annotations - -from collections import defaultdict - -from sqlmodel import Session, select - -from chronicle.core import AggregateFact -from chronicle.normalization import SourceFact, as_target, convert_units, target_kwargs -from chronicle.source_package import load_source_package - -from .schema import ( - DataSource, - Jurisdiction, - Stratum, - StratumConstraint, - Target, - TargetType, - init_db, -) - -# Trustworthy FNS source package parsed cell-by-cell with guard cells. -SNAP_SOURCE_PACKAGE = "packages/usda_snap/fy69_to_current" - -SOURCE_URL = ( - "https://www.fns.usda.gov/pd/supplemental-nutrition-assistance-program-snap" -) - -# Source-package canonical concepts mapped to legacy target variables. Each -# tuple is (target variable, target type, output unit). The FNS workbook -# publishes household and person counts as absolute counts and benefits as -# absolute dollars, so no scale conversion is applied. -_MEASURE_TARGETS: dict[str, tuple[str, TargetType, str]] = { - "usda_snap.average_monthly_households": ( - "snap_household_count", - TargetType.COUNT, - "count", - ), - "usda_snap.average_monthly_persons": ( - "snap_participant_count", - TargetType.COUNT, - "count", - ), - "usda_snap.total_benefits": ( - "snap_benefits", - TargetType.AMOUNT, - "dollars", - ), -} - - -def _fips_from_geography_id(geography_id: str) -> str | None: - """Extract a 2-digit state FIPS code from a Census GEOID. - - State GEOIDs look like ``0400000US06`` (California). Non-state geographies - (e.g. the national ``0100000US``) return ``None``. - """ - marker = "US" - index = geography_id.rfind(marker) - if index == -1: - return None - suffix = geography_id[index + len(marker) :] - if len(suffix) == 2 and suffix.isdigit(): - return suffix - return None - - -def _provenance_note(fact: AggregateFact) -> str: - """Build a per-value provenance note from the fact's source lineage.""" - source = fact.source - parts = [ - f"USDA FNS {source.source_table}" if source.source_table else "USDA FNS", - ] - if source.source_file: - parts.append(f"source_file={source.source_file}") - if source.vintage: - parts.append(f"vintage={source.vintage}") - if source.source_sha256: - parts.append(f"sha256={source.source_sha256}") - if source.raw_r2_uri: - parts.append(f"raw={source.raw_r2_uri}") - if fact.source_record_id: - parts.append(f"record={fact.source_record_id}") - return "; ".join(parts) - - -def build_snap_target( - stratum: Stratum, - fact: AggregateFact, - *, - variable: str, - target_type: TargetType, - output_unit: str, - source_table: str, -) -> Target: - """Build a SNAP target input from a parsed source fact. - - The value already arrives in its published absolute unit, so this records a - no-op ``convert_units`` step purely to attach normalization lineage; the - numeric value is unchanged. - """ - source_fact = SourceFact( - name=variable, - value=fact.value, - period=fact.period.value, - unit=fact.measure.unit, - source=DataSource.USDA_SNAP, - jurisdiction=stratum.jurisdiction, - source_table=source_table, - source_url=SOURCE_URL, - ) - converted = convert_units(source_fact, 1, output_unit) - blueprint = as_target( - converted, - variable=variable, - target_type=target_type, - stratum_name=stratum.name, - notes=_provenance_note(fact), - ) - return Target(**target_kwargs(blueprint, stratum_id=stratum.id)) - - -def get_or_create_stratum( - session: Session, - name: str, - jurisdiction: Jurisdiction, - constraints: list[tuple[str, str, str]], - description: str | None = None, - parent_id: int | None = None, - stratum_group_id: str | None = None, -) -> Stratum: - """Get existing stratum or create new one.""" - definition_hash = Stratum.compute_hash(constraints, jurisdiction) - - existing = session.exec( - select(Stratum).where(Stratum.definition_hash == definition_hash) - ).first() - - if existing: - return existing - - stratum = Stratum( - name=name, - description=description, - jurisdiction=jurisdiction, - definition_hash=definition_hash, - parent_id=parent_id, - stratum_group_id=stratum_group_id, - ) - session.add(stratum) - session.flush() - - for variable, operator, value in constraints: - constraint = StratumConstraint( - stratum_id=stratum.id, - variable=variable, - operator=operator, - value=value, - ) - session.add(constraint) - - return stratum - - -def load_snap_facts(years: list[int] | None = None) -> list[AggregateFact]: - """Load source-backed SNAP facts from the trustworthy FNS source package. - - The package artifact is the FY2024 workbook, so every fact carries a - ``fiscal_year 2024`` reference period. ``years`` optionally filters to - matching fact reference periods; a year with no matching facts (e.g. a - prior fiscal year not yet added to the package) simply yields nothing - rather than mislabeling FY2024 data. - """ - package = load_source_package(SNAP_SOURCE_PACKAGE) - facts = package.build_facts(package.artifact.artifact_year) - if years is not None: - wanted = set(years) - facts = [fact for fact in facts if fact.period.value in wanted] - return facts - - -def load_snap_targets(session: Session, years: list[int] | None = None): - """Load SNAP targets into the database from parsed FNS source facts. - - Args: - session: Database session. - years: Optional list of fiscal-year reference periods to include. - Defaults to every period the source package publishes. - """ - facts = load_snap_facts(years) - - # Group facts by geography so each stratum gets its measures together. - by_geography: dict[str, list[AggregateFact]] = defaultdict(list) - for fact in facts: - by_geography[fact.geography.id].append(fact) - - national_stratum: Stratum | None = None - - # Materialize the national stratum first so state strata can parent to it. - for geography_facts in by_geography.values(): - if geography_facts[0].geography.level != "country": - continue - - national_stratum = get_or_create_stratum( - session, - name="US SNAP Recipients", - jurisdiction=Jurisdiction.US_FEDERAL, - constraints=[("snap", "==", "1")], - description="All SNAP recipient households/individuals in the US", - stratum_group_id="snap_national", - ) - - for fact in geography_facts: - mapping = _MEASURE_TARGETS.get(fact.measure.concept) - if mapping is None: - continue - variable, target_type, output_unit = mapping - session.add( - build_snap_target( - national_stratum, - fact, - variable=variable, - target_type=target_type, - output_unit=output_unit, - source_table="SNAP National Summary", - ) - ) - - # State-level strata and targets. - for geography_id, geography_facts in by_geography.items(): - first = geography_facts[0] - if first.geography.level != "state": - continue - - fips = _fips_from_geography_id(geography_id) - if fips is None: - continue - - state_name = first.geography.name or geography_id - state_stratum = get_or_create_stratum( - session, - name=f"{state_name} SNAP Recipients", - jurisdiction=Jurisdiction.US, - constraints=[ - ("snap", "==", "1"), - ("state_fips", "==", fips), - ], - description=f"SNAP recipients in {state_name}", - parent_id=national_stratum.id if national_stratum else None, - stratum_group_id="snap_states", - ) - - for fact in geography_facts: - mapping = _MEASURE_TARGETS.get(fact.measure.concept) - if mapping is None: - continue - variable, target_type, output_unit = mapping - session.add( - build_snap_target( - state_stratum, - fact, - variable=variable, - target_type=target_type, - output_unit=output_unit, - source_table="SNAP State Summary", - ) - ) - - session.commit() - - -def run_etl(db_path=None): - """Run the SNAP ETL pipeline.""" - from pathlib import Path - - from .schema import DEFAULT_DB_PATH - - path = Path(db_path) if db_path else DEFAULT_DB_PATH - engine = init_db(path) - - with Session(engine) as session: - load_snap_targets(session) - print(f"Loaded SNAP targets to {path}") - - -if __name__ == "__main__": - run_etl() diff --git a/db/etl_soi.py b/db/etl_soi.py deleted file mode 100644 index ee6df3bf..00000000 --- a/db/etl_soi.py +++ /dev/null @@ -1,626 +0,0 @@ -""" -ETL for IRS Statistics of Income (SOI) targets. - -Loads IRS SOI Publication 1304 Table 1.1 into the targets database. -""" - -from __future__ import annotations - -import hashlib -from functools import lru_cache -from importlib.resources import files -from io import BytesIO -from typing import Any, TypedDict -from urllib.request import Request, urlopen - -import pandas as pd -import yaml -from sqlmodel import Session, select - -from .schema import ( - DataSource, - Jurisdiction, - Stratum, - StratumConstraint, - Target, - TargetType, - init_db, -) - -SOURCE_URL = ( - "https://www.irs.gov/statistics/soi-tax-stats-individual-income-tax-statistics" -) -TABLE_1_1_SHEET_NAME = "TBL11" -TABLE_1_4_SHEET_NAME = "TBL14" -TABLE_1_1_MONEY_SCALE = 1_000 -TABLE_1_4_MONEY_SCALE = 1_000 -TABLE_1_1_PACKAGE_DIR = "data/irs_soi/table_1_1" -TABLE_1_4_PACKAGE_DIR = "data/irs_soi/table_1_4" -TABLE_1_1_MANIFEST = "manifest.yaml" -TABLE_1_4_MANIFEST = "manifest.yaml" - -# AGI bracket definitions (lower, upper) in dollars. -AGI_BRACKETS = { - "under_1": (float("-inf"), 1), - "1_to_5k": (1, 5_000), - "5k_to_10k": (5_000, 10_000), - "10k_to_15k": (10_000, 15_000), - "15k_to_20k": (15_000, 20_000), - "20k_to_25k": (20_000, 25_000), - "25k_to_30k": (25_000, 30_000), - "30k_to_40k": (30_000, 40_000), - "40k_to_50k": (40_000, 50_000), - "50k_to_75k": (50_000, 75_000), - "75k_to_100k": (75_000, 100_000), - "100k_to_200k": (100_000, 200_000), - "200k_to_500k": (200_000, 500_000), - "500k_to_1m": (500_000, 1_000_000), - "1m_to_1_5m": (1_000_000, 1_500_000), - "1_5m_to_2m": (1_500_000, 2_000_000), - "2m_to_5m": (2_000_000, 5_000_000), - "5m_to_10m": (5_000_000, 10_000_000), - "10m_plus": (10_000_000, float("inf")), -} - -TABLE_1_1_AGI_LABEL_TO_BRACKET = { - "No adjusted gross income": "under_1", - "$1 under $5,000": "1_to_5k", - "$5,000 under $10,000": "5k_to_10k", - "$10,000 under $15,000": "10k_to_15k", - "$15,000 under $20,000": "15k_to_20k", - "$20,000 under $25,000": "20k_to_25k", - "$25,000 under $30,000": "25k_to_30k", - "$30,000 under $40,000": "30k_to_40k", - "$40,000 under $50,000": "40k_to_50k", - "$50,000 under $75,000": "50k_to_75k", - "$75,000 under $100,000": "75k_to_100k", - "$100,000 under $200,000": "100k_to_200k", - "$200,000 under $500,000": "200k_to_500k", - "$500,000 under $1,000,000": "500k_to_1m", - "$1,000,000 under $1,500,000": "1m_to_1_5m", - "$1,500,000 under $2,000,000": "1_5m_to_2m", - "$2,000,000 under $5,000,000": "2m_to_5m", - "$5,000,000 under $10,000,000": "5m_to_10m", - "$10,000,000 or more": "10m_plus", -} - - -class SOITable11Data(TypedDict): - source_url: str - total_returns: int - total_agi: int - total_income_tax: int - returns_by_agi_bracket: dict[str, int] - agi_by_bracket: dict[str, int] - income_tax_by_bracket: dict[str, int] - - -class SOITable14Data(TypedDict): - source_url: str - total_employment_income_returns: int - total_employment_income: int - employment_income_returns_by_agi_bracket: dict[str, int] - employment_income_by_agi_bracket: dict[str, int] - - -@lru_cache(maxsize=1) -def _table_1_1_manifest() -> dict[str, Any]: - manifest_path = files("db").joinpath(TABLE_1_1_PACKAGE_DIR, TABLE_1_1_MANIFEST) - with manifest_path.open("r", encoding="utf-8") as f: - return yaml.safe_load(f) - - -@lru_cache(maxsize=1) -def _table_1_4_manifest() -> dict[str, Any]: - manifest_path = files("db").joinpath(TABLE_1_4_PACKAGE_DIR, TABLE_1_4_MANIFEST) - with manifest_path.open("r", encoding="utf-8") as f: - return yaml.safe_load(f) - - -def available_soi_years() -> list[int]: - """Return SOI Table 1.1 years available in the source manifest.""" - return sorted(int(year) for year in _table_1_1_manifest()["files"]) - - -def available_soi_table_1_4_years() -> list[int]: - """Return SOI Table 1.4 years available in the source manifest.""" - return sorted(int(year) for year in _table_1_4_manifest()["files"]) - - -def soi_table_1_1_source_url(year: int) -> str: - """Return the IRS source URL for a Table 1.1 year.""" - return _table_1_1_file_spec(year)["source_url"] - - -def soi_table_1_4_source_url(year: int) -> str: - """Return the IRS source URL for a Table 1.4 year.""" - return _table_1_4_file_spec(year)["source_url"] - - -def _table_1_1_file_spec(year: int) -> dict[str, str]: - files_by_year = _table_1_1_manifest()["files"] - try: - return files_by_year[year] - except KeyError: - return files_by_year[str(year)] - - -def _table_1_4_file_spec(year: int) -> dict[str, str]: - files_by_year = _table_1_4_manifest()["files"] - try: - return files_by_year[year] - except KeyError: - return files_by_year[str(year)] - - -# Compatibility manifest for older callers. Numeric SOI values are parsed from -# the source files at load time rather than embedded in Python. -SOI_DATA = { - year: {"source_url": soi_table_1_1_source_url(year)} - for year in available_soi_years() -} - - -@lru_cache(maxsize=None) -def _table_1_1_content(year: int) -> bytes: - return _table_content(year, _table_1_1_file_spec, TABLE_1_1_PACKAGE_DIR) - - -@lru_cache(maxsize=None) -def _table_1_4_content(year: int) -> bytes: - return _table_content(year, _table_1_4_file_spec, TABLE_1_4_PACKAGE_DIR) - - -def _table_content(year: int, file_spec_fn, package_dir: str) -> bytes: - spec = file_spec_fn(year) - package_path = files("db").joinpath(package_dir, spec["filename"]) - if package_path.is_file(): - content = package_path.read_bytes() - else: - request = Request( - spec["source_url"], - headers={"User-Agent": "policyengine-chronicle/0.1", "Accept": "*/*"}, - ) - with urlopen(request, timeout=120) as response: - content = response.read() - - expected_sha = spec.get("sha256") - if expected_sha: - actual_sha = hashlib.sha256(content).hexdigest() - if actual_sha != expected_sha: - raise ValueError( - f"SOI source file {year} checksum mismatch: " - f"expected {expected_sha}, got {actual_sha}" - ) - return content - - -@lru_cache(maxsize=None) -def _read_soi_table_1_1_frame(year: int) -> pd.DataFrame: - return pd.read_excel( - BytesIO(_table_1_1_content(year)), - sheet_name=TABLE_1_1_SHEET_NAME, - header=None, - dtype=object, - engine="xlrd", - ) - - -@lru_cache(maxsize=None) -def _read_soi_table_1_4_frame(year: int) -> pd.DataFrame: - return pd.read_excel( - BytesIO(_table_1_4_content(year)), - sheet_name=TABLE_1_4_SHEET_NAME, - header=None, - dtype=object, - engine="xlrd", - ) - - -def load_soi_table_1_1_data(year: int) -> SOITable11Data: - """Parse one year of IRS SOI Publication 1304 Table 1.1.""" - return _parse_soi_table_1_1_frame( - _read_soi_table_1_1_frame(year), - source_url=soi_table_1_1_source_url(year), - ) - - -def load_soi_table_1_4_data(year: int) -> SOITable14Data: - """Parse wage targets from IRS SOI Publication 1304 Table 1.4.""" - return _parse_soi_table_1_4_frame( - _read_soi_table_1_4_frame(year), - source_url=soi_table_1_4_source_url(year), - ) - - -def _clean_label(value: object) -> str: - if value is None or pd.isna(value): - return "" - return " ".join(str(value).replace("\n", " ").split()).strip() - - -def _numeric_value(value: object) -> float: - if value is None or pd.isna(value): - raise ValueError("Expected numeric SOI cell, got empty value") - if isinstance(value, str): - clean = value.replace("$", "").replace(",", "").strip() - if clean.startswith("["): - raise ValueError(f"Expected numeric SOI cell, got footnote marker {value}") - return float(clean) - return float(value) - - -def _count_value(value: object) -> int: - return int(round(_numeric_value(value))) - - -def _money_value(value: object) -> int: - return int(round(_numeric_value(value) * TABLE_1_1_MONEY_SCALE)) - - -def _table_1_4_money_value(value: object) -> int: - return int(round(_numeric_value(value) * TABLE_1_4_MONEY_SCALE)) - - -def _header_text(df: pd.DataFrame, column: int) -> str: - pieces = [_clean_label(df.iat[row, column]) for row in range(min(8, len(df)))] - return " ".join(piece for piece in pieces if piece).lower() - - -def _find_column(df: pd.DataFrame, *phrases: str) -> int: - for column in range(df.shape[1]): - header = _header_text(df, column) - if all(phrase.lower() in header for phrase in phrases): - return column - joined = ", ".join(phrases) - raise ValueError(f"Could not find SOI Table 1.1 column containing: {joined}") - - -def _table_1_1_size_rows(df: pd.DataFrame) -> pd.DataFrame: - labels = df.iloc[:, 0].map(_clean_label) - start_matches = labels[labels == "All returns"] - if start_matches.empty: - raise ValueError("Could not find SOI Table 1.1 'All returns' row") - start = int(start_matches.index[0]) - - end_label = "Accumulated from smallest size of adjusted gross income" - end_matches = labels[(labels == end_label) & (labels.index > start)] - end = int(end_matches.index[0]) if not end_matches.empty else len(df) - return df.loc[start : end - 1] - - -def _table_1_4_size_rows(df: pd.DataFrame) -> pd.DataFrame: - labels = df.iloc[:, 0].map(_clean_label) - start_matches = labels[labels == "All returns, total"] - if start_matches.empty: - raise ValueError("Could not find SOI Table 1.4 'All returns, total' row") - start = int(start_matches.index[0]) - - end_matches = labels[(labels == "Taxable returns, total") & (labels.index > start)] - end = int(end_matches.index[0]) if not end_matches.empty else len(df) - return df.loc[start : end - 1] - - -def _row_by_label(rows: pd.DataFrame, label: str) -> pd.Series: - labels = rows.iloc[:, 0].map(_clean_label) - matches = rows[labels == label] - if matches.empty: - raise ValueError(f"Could not find SOI Table 1.1 row: {label}") - return matches.iloc[0] - - -def _parse_soi_table_1_1_frame( - df: pd.DataFrame, - *, - source_url: str, -) -> SOITable11Data: - count_col = _find_column(df, "number of returns") - agi_col = _find_column(df, "adjusted gross income less deficit", "amount") - income_tax_col = _find_column(df, "total income tax", "amount") - rows = _table_1_1_size_rows(df) - all_returns = _row_by_label(rows, "All returns") - - returns_by_bracket = {} - agi_by_bracket = {} - income_tax_by_bracket = {} - for source_label, bracket_name in TABLE_1_1_AGI_LABEL_TO_BRACKET.items(): - row = _row_by_label(rows, source_label) - returns_by_bracket[bracket_name] = _count_value(row.iat[count_col]) - agi_by_bracket[bracket_name] = _money_value(row.iat[agi_col]) - income_tax_by_bracket[bracket_name] = _money_value(row.iat[income_tax_col]) - - return { - "source_url": source_url, - "total_returns": _count_value(all_returns.iat[count_col]), - "total_agi": _money_value(all_returns.iat[agi_col]), - "total_income_tax": _money_value(all_returns.iat[income_tax_col]), - "returns_by_agi_bracket": returns_by_bracket, - "agi_by_bracket": agi_by_bracket, - "income_tax_by_bracket": income_tax_by_bracket, - } - - -def _parse_soi_table_1_4_frame( - df: pd.DataFrame, - *, - source_url: str, -) -> SOITable14Data: - wage_count_col = _find_column(df, "wages", "number of returns") - wage_amount_col = wage_count_col + 1 - if "amount" not in _header_text(df, wage_amount_col): - raise ValueError("Could not find SOI Table 1.4 wage amount column") - rows = _table_1_4_size_rows(df) - all_returns = _row_by_label(rows, "All returns, total") - - wage_returns_by_bracket = {} - wage_amount_by_bracket = {} - for source_label, bracket_name in TABLE_1_1_AGI_LABEL_TO_BRACKET.items(): - row = _row_by_label(rows, source_label) - wage_returns_by_bracket[bracket_name] = _count_value(row.iat[wage_count_col]) - wage_amount_by_bracket[bracket_name] = _table_1_4_money_value( - row.iat[wage_amount_col] - ) - - return { - "source_url": source_url, - "total_employment_income_returns": _count_value( - all_returns.iat[wage_count_col] - ), - "total_employment_income": _table_1_4_money_value( - all_returns.iat[wage_amount_col] - ), - "employment_income_returns_by_agi_bracket": wage_returns_by_bracket, - "employment_income_by_agi_bracket": wage_amount_by_bracket, - } - - -def get_or_create_stratum( - session: Session, - name: str, - jurisdiction: Jurisdiction, - constraints: list[tuple[str, str, str]], - description: str | None = None, - parent_id: int | None = None, - stratum_group_id: str | None = None, -) -> Stratum: - """Get existing stratum or create new one.""" - definition_hash = Stratum.compute_hash(constraints, jurisdiction) - - existing = session.exec( - select(Stratum).where(Stratum.definition_hash == definition_hash) - ).first() - - if existing: - return existing - - stratum = Stratum( - name=name, - description=description, - jurisdiction=jurisdiction, - definition_hash=definition_hash, - parent_id=parent_id, - stratum_group_id=stratum_group_id, - ) - session.add(stratum) - session.flush() - - for variable, operator, value in constraints: - constraint = StratumConstraint( - stratum_id=stratum.id, - variable=variable, - operator=operator, - value=value, - ) - session.add(constraint) - - return stratum - - -def _add_target( - session: Session, - *, - stratum_id: int, - variable: str, - period: int, - value: float, - target_type: TargetType, - source_url: str, - source_table: str = "Table 1.1", -) -> None: - session.add( - Target( - stratum_id=stratum_id, - variable=variable, - period=period, - value=value, - target_type=target_type, - source=DataSource.IRS_SOI, - source_table=source_table, - source_url=source_url, - ) - ) - - -def load_soi_targets(session: Session, years: list[int] | None = None) -> None: - """ - Load SOI targets into database. - - Args: - session: Database session - years: Years to load (default: all available) - """ - if years is None: - years = available_soi_years() - - for year in years: - if year not in available_soi_years(): - continue - - data = load_soi_table_1_1_data(year) - source_url = data["source_url"] - table_1_4_data = None - table_1_4_source_url = None - if year in available_soi_table_1_4_years(): - table_1_4_data = load_soi_table_1_4_data(year) - table_1_4_source_url = table_1_4_data["source_url"] - - national_stratum = get_or_create_stratum( - session, - name="US All Filers", - jurisdiction=Jurisdiction.US_FEDERAL, - constraints=[("is_tax_filer", "==", "1")], - description="All individual income tax returns filed in the US", - stratum_group_id="national", - ) - - _add_target( - session, - stratum_id=national_stratum.id, - variable="tax_unit_count", - period=year, - value=data["total_returns"], - target_type=TargetType.COUNT, - source_url=source_url, - ) - _add_target( - session, - stratum_id=national_stratum.id, - variable="adjusted_gross_income", - period=year, - value=data["total_agi"], - target_type=TargetType.AMOUNT, - source_url=source_url, - ) - _add_target( - session, - stratum_id=national_stratum.id, - variable="income_tax_liability", - period=year, - value=data["total_income_tax"], - target_type=TargetType.AMOUNT, - source_url=source_url, - ) - - if table_1_4_data is not None and table_1_4_source_url is not None: - _add_target( - session, - stratum_id=national_stratum.id, - variable="employment_income", - period=year, - value=table_1_4_data["total_employment_income_returns"], - target_type=TargetType.COUNT, - source_url=table_1_4_source_url, - source_table="Table 1.4", - ) - _add_target( - session, - stratum_id=national_stratum.id, - variable="employment_income", - period=year, - value=table_1_4_data["total_employment_income"], - target_type=TargetType.AMOUNT, - source_url=table_1_4_source_url, - source_table="Table 1.4", - ) - - for bracket_name, (lower, upper) in AGI_BRACKETS.items(): - constraints = [] - if lower != float("-inf"): - constraints.append(("adjusted_gross_income", ">=", str(lower))) - if upper != float("inf"): - constraints.append(("adjusted_gross_income", "<", str(upper))) - - bracket_stratum = get_or_create_stratum( - session, - name=f"US Filers AGI {bracket_name}", - jurisdiction=Jurisdiction.US_FEDERAL, - constraints=constraints, - description=f"Tax filers with AGI in {bracket_name} bracket", - parent_id=national_stratum.id, - stratum_group_id="agi_brackets", - ) - - if bracket_name in data["returns_by_agi_bracket"]: - _add_target( - session, - stratum_id=bracket_stratum.id, - variable="tax_unit_count", - period=year, - value=data["returns_by_agi_bracket"][bracket_name], - target_type=TargetType.COUNT, - source_url=source_url, - ) - - if bracket_name in data["agi_by_bracket"]: - _add_target( - session, - stratum_id=bracket_stratum.id, - variable="adjusted_gross_income", - period=year, - value=data["agi_by_bracket"][bracket_name], - target_type=TargetType.AMOUNT, - source_url=source_url, - ) - - if bracket_name in data["income_tax_by_bracket"]: - _add_target( - session, - stratum_id=bracket_stratum.id, - variable="income_tax_liability", - period=year, - value=data["income_tax_by_bracket"][bracket_name], - target_type=TargetType.AMOUNT, - source_url=source_url, - ) - - if table_1_4_data is not None and table_1_4_source_url is not None: - if ( - bracket_name - in table_1_4_data["employment_income_returns_by_agi_bracket"] - ): - _add_target( - session, - stratum_id=bracket_stratum.id, - variable="employment_income", - period=year, - value=table_1_4_data[ - "employment_income_returns_by_agi_bracket" - ][bracket_name], - target_type=TargetType.COUNT, - source_url=table_1_4_source_url, - source_table="Table 1.4", - ) - - if bracket_name in table_1_4_data["employment_income_by_agi_bracket"]: - _add_target( - session, - stratum_id=bracket_stratum.id, - variable="employment_income", - period=year, - value=table_1_4_data["employment_income_by_agi_bracket"][ - bracket_name - ], - target_type=TargetType.AMOUNT, - source_url=table_1_4_source_url, - source_table="Table 1.4", - ) - - session.commit() - - -def run_etl(db_path=None) -> None: - """Run the SOI ETL pipeline.""" - from pathlib import Path - - from .schema import DEFAULT_DB_PATH - - path = Path(db_path) if db_path else DEFAULT_DB_PATH - engine = init_db(path) - - with Session(engine) as session: - load_soi_targets(session) - print(f"Loaded SOI targets to {path}") - - -if __name__ == "__main__": - run_etl() diff --git a/db/schema.py b/db/schema.py index 9bb63bc0..41cd9f9e 100644 --- a/db/schema.py +++ b/db/schema.py @@ -1,13 +1,13 @@ """ -Database schema for targets and strata. +Database schema for Chronicle's local source-artifact store. -Three-table design (following policyengine-us-data): -- strata: Population subgroups defined by constraints -- stratum_constraints: Rules defining each stratum -- targets: Administrative totals linked to strata +- source_artifacts: source files with provenance and checksum metadata +- source_tables / source_columns / source_rows: their parsed contents + +The legacy strata / stratum_constraints / targets tables were retired; see +docs/us-legacy-etl-cleanup.md. """ -import hashlib from datetime import datetime, timezone from enum import Enum from pathlib import Path @@ -15,7 +15,7 @@ from sqlmodel import Field, Relationship, Session, SQLModel, create_engine -# Default storage location (local dev; production uses Supabase) +# Default local storage location for the source-artifact database. DEFAULT_DB_PATH = Path(__file__).parent.parent / "macro" / "targets.db" @@ -54,134 +54,6 @@ class DataSource(str, Enum): OBR = "obr" # Office for Budget Responsibility projections -class TargetType(str, Enum): - """Whether target is a count or monetary amount.""" - - COUNT = "count" - AMOUNT = "amount" - RATE = "rate" - - -class GeographicLevel(str, Enum): - """Geographic aggregation levels for targets.""" - - NATIONAL = "national" - STATE = "state" - CONGRESSIONAL_DISTRICT = "congressional_district" - COUNTY = "county" - ZIP_CODE = "zip_code" - METRO_AREA = "metro_area" - - -class Stratum(SQLModel, table=True): - """ - A population subgroup defined by constraints. - - Examples: - - All US tax filers - - California residents with AGI $50k-$75k - - Single filers under age 65 - """ - - __tablename__ = "strata" - - id: Optional[int] = Field(default=None, primary_key=True) - name: str = Field(index=True) - description: Optional[str] = None - jurisdiction: Jurisdiction - definition_hash: str = Field(unique=True, index=True) - - # Hierarchy: strata can nest (e.g., state within national) - parent_id: Optional[int] = Field(default=None, foreign_key="strata.id") - stratum_group_id: Optional[str] = Field( - default=None, - index=True, - description="For grouping related strata in calibration", - ) - - created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) - - # Relationships - constraints: list["StratumConstraint"] = Relationship(back_populates="stratum") - targets: list["Target"] = Relationship(back_populates="stratum") - - @classmethod - def compute_hash( - cls, - constraints: list[tuple[str, str, str]], - jurisdiction: "Jurisdiction | None" = None, - ) -> str: - """Compute unique hash from constraint definitions and jurisdiction.""" - sorted_constraints = sorted(constraints) - # Include jurisdiction to avoid collisions between US/UK strata - hash_input = f"{jurisdiction}:{sorted_constraints}" - return hashlib.sha256(hash_input.encode()).hexdigest()[:16] - - -class StratumConstraint(SQLModel, table=True): - """ - A single constraint defining a stratum. - - Examples: - - variable="state_fips", operator="==", value="06" (California) - - variable="agi", operator=">=", value="50000" - - variable="filing_status", operator="==", value="single" - """ - - __tablename__ = "stratum_constraints" - - id: Optional[int] = Field(default=None, primary_key=True) - stratum_id: int = Field(foreign_key="strata.id", index=True) - - variable: str = Field(index=True) - operator: str # "==", "!=", ">", ">=", "<", "<=", "in" - value: str # Stored as string, parsed based on variable dtype - - # Relationship - stratum: Optional[Stratum] = Relationship(back_populates="constraints") - - -class Target(SQLModel, table=True): - """ - An administrative target value for a stratum. - - Examples: - - Total returns in California: 18.5M - - Total AGI for $50k-$75k bracket: $1.2T - - SNAP households in Texas: 1.4M - """ - - __tablename__ = "targets" - - id: Optional[int] = Field(default=None, primary_key=True) - stratum_id: int = Field(foreign_key="strata.id", index=True) - - variable: str = Field(index=True, description="Chronicle target input variable ID") - period: int = Field(index=True, description="Year") - value: float - target_type: TargetType = Field(default=TargetType.COUNT) - - # Geographic level for this target - geographic_level: Optional[GeographicLevel] = Field( - default=None, index=True, description="Geographic aggregation level" - ) - - source: DataSource - source_table: Optional[str] = None # e.g., "Table 1.1" - source_url: Optional[str] = None - notes: Optional[str] = None - - # Confidence/quality indicators - is_preliminary: bool = Field(default=False) - margin_of_error: Optional[float] = None - - created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) - updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) - - # Relationship - stratum: Optional[Stratum] = Relationship(back_populates="targets") - - class SourceArtifact(SQLModel, table=True): """A source file ingested into Chronicle with provenance and checksum metadata.""" @@ -248,7 +120,7 @@ class SourceRow(SQLModel, table=True): def get_engine(db_path: Path = DEFAULT_DB_PATH): - """Get SQLAlchemy engine for the targets database.""" + """Get SQLAlchemy engine for the source-artifact database.""" db_path.parent.mkdir(parents=True, exist_ok=True) return create_engine(f"sqlite:///{db_path}", echo=False) diff --git a/db/supabase_client.py b/db/supabase_client.py index e14bb5bb..83fdf328 100644 --- a/db/supabase_client.py +++ b/db/supabase_client.py @@ -1,9 +1,10 @@ """ Supabase client for Chronicle. -Provides connection to PolicyEngine Supabase database for: -- Source metadata and dataset registries -- Target inputs +Provides connection to PolicyEngine Supabase database for source metadata and +dataset registries. The legacy Supabase target-input helpers (query_targets, +query_strata, insert_targets_batch) were retired; see +docs/us-legacy-etl-cleanup.md. """ from __future__ import annotations @@ -27,7 +28,6 @@ def _env(*names: str) -> str | None: LEDGER_SCHEMA = _env("POLICYENGINE_LEDGER_SCHEMA") or "ledger" -TARGETS_SCHEMA = _env("POLICYENGINE_TARGETS_SCHEMA") or "targets" @dataclass @@ -123,103 +123,3 @@ def query_sources( result = query.execute() return result.data - - -# ============================================================================= -# Targets and strata -# ============================================================================= - - -def query_strata( - jurisdiction: Optional[str] = None, -) -> List[Dict[str, Any]]: - """ - Query target input strata with their constraints. - - Args: - jurisdiction: Filter by jurisdiction - - Returns: - List of strata records with nested constraints - """ - client = get_supabase_client() - query = _table(client, TARGETS_SCHEMA, "strata").select("*, stratum_constraints(*)") - - if jurisdiction: - query = query.eq("jurisdiction", jurisdiction) - - result = query.execute() - return result.data - - -def query_targets( - jurisdiction: Optional[str] = None, - year: Optional[int] = None, - source_id: Optional[str] = None, - variable: Optional[str] = None, -) -> List[Dict[str, Any]]: - """ - Query Chronicle target inputs. - - Args: - jurisdiction: Filter by jurisdiction (via stratum join) - year: Filter by period/year - source_id: Filter by source UUID - variable: Filter by variable name - - Returns: - List of target records with stratum info - """ - client = get_supabase_client() - # Nested join: strata with their stratum_constraints - query = _table(client, TARGETS_SCHEMA, "targets").select( - "*, strata(*, stratum_constraints(*)), sources(*)" - ) - - if year: - query = query.eq("period", year) - if source_id: - query = query.eq("source_id", source_id) - if variable: - query = query.eq("variable", variable) - - result = query.execute() - - # Filter by jurisdiction if specified (post-query since it's on joined table) - data = result.data - if jurisdiction: - data = [ - t for t in data if t.get("strata", {}).get("jurisdiction") == jurisdiction - ] - - return data - - -# ============================================================================= -# Insert operations -# ============================================================================= - - -def insert_targets_batch( - targets: List[Dict[str, Any]], - chunk_size: int = 100, -) -> int: - """ - Insert target records in batches. - - Args: - targets: List of target dicts - chunk_size: Records per batch - - Returns: - Number of records inserted - """ - client = get_supabase_client() - total = 0 - - for i in range(0, len(targets), chunk_size): - chunk = targets[i : i + chunk_size] - _table(client, TARGETS_SCHEMA, "targets").insert(chunk).execute() - total += len(chunk) - - return total diff --git a/docs/adr-chronicle-fact-identity-v2.md b/docs/adr-chronicle-fact-identity-v2.md index 8ce3dd56..a02f4b52 100644 --- a/docs/adr-chronicle-fact-identity-v2.md +++ b/docs/adr-chronicle-fact-identity-v2.md @@ -278,6 +278,12 @@ universe constraint. ## Legacy Migration +> **Superseded (2026-10).** The synthetic-legacy migration below was never +> implemented. The legacy target rows were retired instead of migrated: no live +> consumer read them, and most of their values did not trace to publisher bytes. See +> `docs/us-legacy-etl-cleanup.md`. No Chronicle code emits `synthetic_legacy` +> records or `L0_SYNTHETIC` facts. + Legacy direct target rows are compatibility inputs, not production Chronicle facts. Migration path: diff --git a/docs/agent-source-package-harness.md b/docs/agent-source-package-harness.md index 41f540cb..894b1f71 100644 --- a/docs/agent-source-package-harness.md +++ b/docs/agent-source-package-harness.md @@ -4,12 +4,11 @@ Chronicle source-population agents should fill constrained source packages and l the build suite decide whether the package is admissible. Agents should not hand-edit Chronicle DB artifacts, generated JSONL outputs, or core schema modules. -The older Python ETL loaders that write directly into the legacy `targets` -tables are compatibility and migration inputs, not the preferred agent -population path. They are useful for proving source coverage against Microcosm -today, but a source family should become agent-ready only after it is expressed -as a source package with full-document parsing, source-row or source-cell -lineage, provenance, constraints, and a passing `build-suite` report. +The older Python ETL loaders that wrote directly into the legacy `targets` +tables were retired (see `docs/us-legacy-etl-cleanup.md`). A source family +becomes agent-ready only after it is expressed as a source package with +full-document parsing, source-row or source-cell lineage, provenance, +constraints, and a passing `build-suite` report. The first gate for a new package is source-artifact acquisition. Agents should register raw source files with `uv run chronicle fetch-artifact` before authoring diff --git a/docs/architecture.md b/docs/architecture.md index 72438fbf..e93b6413 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -43,7 +43,6 @@ sources/ |--------|---------|----------------| | `chronicle` | Source metadata and lineage | sources, files, content, fetch_log | | `indices` | Source time series | series, values (CPI, wage growth) | -| `targets` | Target inputs | strata, constraints, targets | | `microcosm` | Final calibrated data | households, persons, tax_units | ## Python Namespaces @@ -53,12 +52,14 @@ New code should use the `policyengine_chronicle` namespace: ```python from policyengine_chronicle.sources import SourceFile, SourceReference, query_sources from policyengine_chronicle.facts import SourceFact -from policyengine_chronicle.targets import Target, query_targets +from policyengine_chronicle.targets import hard_target_package_aliases from policyengine_chronicle.normalization import convert_units ``` -The `db` package contains the current SQLModel persistence and loader -implementation behind the public `policyengine_chronicle` namespace. +The `db` package holds the local source-artifact store, the Supabase client +and the packaged publisher files behind the public `policyengine_chronicle` +namespace. The legacy `targets` / `strata` / `stratum_constraints` tables and +their loaders were retired (see `docs/us-legacy-etl-cleanup.md`). Jurisdiction source packages should use short import namespaces and published distribution names with a PolicyEngine prefix: @@ -185,19 +186,12 @@ fact and its provenance are identified. ## Downstream Target Composition -### 1. Target Inputs (from `targets.*` schema) +### 1. Source facts (from the consumer-fact export) -Target inputs define source-backed aggregates that Microcosm may use: - -```sql --- targets.strata: Population subgroups -INSERT INTO targets.strata (name, jurisdiction, constraints) -VALUES ('CA adults 18-64', 'us', '[{"variable": "age", "operator": ">=", "value": "18"}, ...]'); - --- targets.targets: Source-backed aggregate values -INSERT INTO targets.targets (stratum_id, variable, value, period) -VALUES (1, 'eitc_recipients', 2500000, 2023); -``` +Microcosm reads source-backed facts from a pinned Chronicle build: it runs +`chronicle build-bundle` and `chronicle build-consumer-artifact` and consumes +the digest-pinned consumer facts. Chronicle has no separate target table; +target selection starts from those facts. ### 2. Variable Mapping @@ -212,7 +206,7 @@ sets: ```python target_set = microcosm.targets.compose( - inputs=chronicle_targets, + inputs=chronicle_facts, target_year=2024, reconciliation="scale_states_to_national", aging="apply_published_growth_factor", @@ -319,17 +313,6 @@ indexing_rule eitc_inflation: rounding: 10 ``` -### targets.* (Target Inputs) - -```sql -targets.strata -- population subgroups with constraints -targets.constraints -- variable, operator, value per stratum -targets.targets -- stratum_id, variable, value, period, source -``` - -Targets in Chronicle are source-backed inputs. Active, aged, reconciled calibration -target sets belong to Microcosm Targets. - ### microcosm.* (Final Output) ```sql diff --git a/docs/pe-calibration-targets.md b/docs/pe-calibration-targets.md index 66fe8cd0..44c6dcd2 100644 --- a/docs/pe-calibration-targets.md +++ b/docs/pe-calibration-targets.md @@ -1,7 +1,8 @@ # US Poverty and Nonfiler Source-Target Coverage This document records the Chronicle-side source coverage contract for US poverty and -nonfiler calibration work. Chronicle stores source-backed facts and target inputs. +nonfiler calibration work. Chronicle stores source-backed facts and records which +source packages back each target family. Microcosm owns the active calibration profile, source reconciliation, aging, and model variable mapping. diff --git a/docs/us-legacy-etl-cleanup.md b/docs/us-legacy-etl-cleanup.md index a7dea079..4bae1551 100644 --- a/docs/us-legacy-etl-cleanup.md +++ b/docs/us-legacy-etl-cleanup.md @@ -11,27 +11,24 @@ series, in the same form as the UK "Legacy ETL cleanup" table in | Loader | Outcome | Why | |---|---|---| -| `db/etl_soi.py` | kept | Parses the IRS Table 1.1 and 1.4 workbooks committed under `db/data/irs_soi/`, checks each against its manifest sha256, and cites the exact `.xls` file on every row. | -| `db/etl_snap.py` | kept | Reads the `usda-snap-fy69-to-current` source package (chronicle#81); every row carries the file, vintage and sha256. | +| `db/etl_soi.py` | retired with the table | Parsed the IRS Table 1.1 and 1.4 workbooks committed under `db/data/irs_soi/` and checked each against its manifest sha256. Its values were publisher-backed, but they only fed the legacy `targets` table, which nothing read. The `soi-table-1-1` and `soi-table-1-4` packages publish the same cells as facts. | +| `db/etl_snap.py` | retired with the table | Copied facts from the `usda-snap-fy69-to-current` source package (chronicle#81) into the legacy `targets` table. The package itself still publishes them. | | `db/etl_soi_state.py`, `db/etl_soi_credits.py`, `db/etl_soi_deductions.py`, `db/etl_soi_income_sources.py` | retired | Invented values presented as IRS figures (below). | | `db/etl_aca_enrollment.py`, `db/etl_bls.py`, `db/etl_cbo.py`, `db/etl_census.py`, `db/etl_cps.py`, `db/etl_medicaid.py`, `db/etl_ssa.py`, `db/etl_ssi.py` | retired | Hand-typed values with landing-page URLs and no artifact; each module mixes transcribed, rounded, computed and invented numbers (below). | -A boundary test (`tests/test_chronicle_boundaries.py`) now runs `chronicle load -all` and requires every row it writes to reproduce its own publisher cell: the -one source-package fact with the same measure, period and complete stratum -definition, built from a publisher file whose bytes hash to its manifest sha256, -with matching source and lineage fields, and a stratum whose jurisdiction, -labels, definition hash and parent are the ones the loader assigns. Every column -of the target, stratum and constraint tables is checked except row ids and -timestamps. Twenty-seven mutation cases, including a row given another real -cell's value and a stratum whose constraint is negated, show it flags each -change. +Until the legacy targets table was retired, a boundary test ran `chronicle load +all` and required every row it wrote to reproduce its own publisher cell (with +27 mutation cases). With the table gone, boundary tests +(`tests/test_chronicle_boundaries.py`) check instead that none of the 14 modules +ships or imports, that no `db/etl_*.py` loader remains, that the CLI rejects +`chronicle load`, `stats` and `query`, and that `chronicle init` creates only the +source-artifact tables. ## Consumers No current reader outside Chronicle was found in the default branches searched. Inside Chronicle the loaders were reached only through `db/cli.py` (`chronicle load `, and `load all`, -which CI's surface job runs), the re-exports in `chronicle/targets/loaders.py` +which CI's surface job ran until the targets table was retired), the re-exports in `chronicle/targets/loaders.py` and `chronicle/jurisdictions/us/__init__.py`, and their own tests. The legacy `targets` table is not part of the source-package bundle, the consumer-fact export or the hosted mirror (`chronicle/mirror.py` exports only the @@ -170,5 +167,60 @@ numbers with rounded, computed and invented ones. The 12 modules, their tests, their `chronicle load` sources and their re-exports are deleted. `db/etl_census.py`'s `load_congressional_district_targets` took caller-supplied values and was called only from -`tests/test_congressional_districts.py`, which keeps its schema check. -`DataSource` members stay: existing databases carry the strings. +`tests/test_congressional_districts.py`. That test file's last check pinned the +legacy `GeographicLevel` enum and went with it when the table was retired. +`DataSource` members stay: `SourceFact` and `SourceReference` still accept them. + +## Retired: the legacy targets table + +After the 12 loaders went, `etl_soi.py` and `etl_snap.py` were the only writers +of the SQLite `targets` / `strata` / `stratum_constraints` tables, and no live +consumer read those tables. The follow-on to chronicle#348 retired the whole path: + +- `db/schema.py`'s `Target`, `Stratum` and `StratumConstraint` tables and the + `TargetType` and `GeographicLevel` enums. `chronicle init` now creates only + the source-artifact tables, which `load-source-files` and `source-stats` use. +- `chronicle load`, `chronicle stats` and `chronicle query` (`db/cli.py`). +- `db/etl_soi.py` and `db/etl_snap.py`, with their tests. The two AGI-bracket + tables that `chronicle/jurisdictions/us/soi.py` imported from `etl_soi.py` + now live in `soi.py`. +- `db/supabase_client.py`'s `query_targets`, `query_strata`, + `insert_targets_batch` and `TARGETS_SCHEMA`. +- `chronicle/targets/loaders.py`, `chronicle/targets/client.py` and the legacy + names re-exported by `chronicle.targets` and `policyengine_chronicle.targets`. + Both modules keep the US poverty/nonfiler target source coverage + (`chronicle/targets/us_poverty.py`). +- `chronicle.normalization`'s `as_target`, `target_kwargs` and + `TargetBlueprint`, which built rows for the table. The lineage helpers + (`convert_units`, `scale_value`, `apply_share`, `format_derivation`) stay, in + `chronicle/normalization/transforms.py`. +- `scripts/export_to_json.py`, which wrote `targets_summary.json` from the table. + +What the audit checked before deleting: + +- **Chronicle itself.** No module under `chronicle/` reads the three tables. + `chronicle/database.py` and `chronicle/mirror.py` work on their own ledger + tables. The bundle, the consumer-fact export and the hosted mirror never + carried the table. +- **Other repositories.** No current reader or writer (see Consumers above for + the search and the three historical readers, all in archived repositories). + Microcosm and chronicle.institute call `chronicle build-bundle` / + `ledger build-bundle`, which this change does not touch. +- **Supabase.** Chronicle's configured project has no `targets` schema, so + `query_targets` and `query_strata` failed with PostgREST `PGRST106 Invalid + schema: targets`. The older copy of the three tables lives in that project's + `microplex` schema. It was loaded in December 2025, has not been modified + since 2026-01-02, and its last observed index read was on 2026-05-30. + Apart from the audit's own probes, no request touched it in the 90 days of + retained edge logs. This change leaves that hosted data alone. +- **Websites.** `targets_summary.json` was written for a CosilicoAI/cosilico.ai + calibration page that was deleted on 2026-02-23. The script's + `../policyengine.org` default path came from a later rename. No + `PolicyEngine/policyengine.org` repository exists, policyengine.org returns + 404 for the file, and policyengine-app-v2 never read it. +- **Packaging.** Chronicle has never been published on PyPI under any of its + names, and no repository installs it as a dependency. + +The `codex/thesis-ledger-facts` branch carries its own frozen copy of the +legacy loaders, and its "Arch checks" job runs `arch load all`. Retiring the +path on `main` does not change that branch; chronicle#352 retires that copy. diff --git a/policyengine_chronicle/cli.py b/policyengine_chronicle/cli.py index 43a9b7fd..22c019c6 100644 --- a/policyengine_chronicle/cli.py +++ b/policyengine_chronicle/cli.py @@ -8,18 +8,19 @@ def main() -> None: - """Dispatch Chronicle commands without eager-loading legacy DB clients.""" + """Dispatch Chronicle commands without eager-loading the DB clients.""" if sys.argv[1:2] in ([], ["-h"], ["--help"]): print( "Usage: chronicle [options]\n\n" "Common commands:\n" - " init\n" - " load\n" - " stats\n" + " build-bundle\n" " validate-facts\n" " validate-source-cells\n" " export-consumer-facts\n" - " build-consumer-artifact\n\n" + " build-consumer-artifact\n" + " init\n" + " load-source-files\n" + " source-stats\n\n" "Run `chronicle --help` for command-specific help." ) return diff --git a/policyengine_chronicle/normalization/__init__.py b/policyengine_chronicle/normalization/__init__.py index 985d88b6..bbd34895 100644 --- a/policyengine_chronicle/normalization/__init__.py +++ b/policyengine_chronicle/normalization/__init__.py @@ -11,7 +11,7 @@ def __getattr__(name: str): - """Load legacy normalization helpers only when explicitly requested.""" + """Load normalization helpers only when explicitly requested.""" if name not in __all__: raise AttributeError(name) from chronicle import normalization diff --git a/policyengine_chronicle/targets/__init__.py b/policyengine_chronicle/targets/__init__.py index 4055e36a..621071c4 100644 --- a/policyengine_chronicle/targets/__init__.py +++ b/policyengine_chronicle/targets/__init__.py @@ -1,29 +1,15 @@ -"""Chronicle target-input helpers. +"""Chronicle target source coverage helpers. -Chronicle owns source-backed facts, target-eligible source inputs, and target -source coverage. Consumers such as Microcosm own selection contracts, decide -which rows their support universe can activate, and calibrate outside Chronicle. +Chronicle owns source-backed facts and target source coverage. Consumers such +as Microcosm own selection contracts, decide which rows their support universe +can activate, and calibrate outside Chronicle. """ __all__ = [ - "DEFAULT_DB_PATH", - "DataSource", - "GeographicLevel", - "Jurisdiction", - "Stratum", - "StratumConstraint", - "Target", "TargetSourceCoverage", - "TargetType", "US_POVERTY_NONFILER_TARGET_COVERAGE", "coverage_entries", - "get_engine", - "get_session", "hard_target_package_aliases", - "init_db", - "insert_targets_batch", - "query_strata", - "query_targets", "source_gap_family_ids", "validate_us_poverty_nonfiler_source_coverage", "validation_only_family_ids", @@ -31,7 +17,7 @@ def __getattr__(name: str): - """Load legacy target helpers only when explicitly requested.""" + """Load target source coverage helpers only when explicitly requested.""" if name not in __all__: raise AttributeError(name) from chronicle import targets diff --git a/scripts/export_to_json.py b/scripts/export_to_json.py deleted file mode 100644 index f8cd191e..00000000 --- a/scripts/export_to_json.py +++ /dev/null @@ -1,137 +0,0 @@ -"""Export targets database to JSON for frontend consumption.""" - -import json -from pathlib import Path -from collections import defaultdict - -from sqlmodel import Session, select - -from db.schema import ( - Stratum, - Target, - get_engine, - DEFAULT_DB_PATH, -) - - -def export_targets_summary(output_path: Path | None = None): - """Export summary statistics for the target dashboard.""" - - if output_path is None: - output_path = ( - Path(__file__).parent.parent.parent - / "policyengine.org" - / "public" - / "data" - / "targets_summary.json" - ) - - engine = get_engine(DEFAULT_DB_PATH) - - with Session(engine) as session: - targets = session.exec(select(Target)).all() - strata = session.exec(select(Stratum)).all() - - # Build summary - by_source = defaultdict( - lambda: {"count": 0, "years": set(), "variables": set()} - ) - by_year = defaultdict(lambda: {"count": 0, "sources": set()}) - by_jurisdiction = defaultdict(lambda: {"count": 0, "sources": set()}) - - # Get stratum jurisdiction lookup - stratum_jurisdiction = {s.id: s.jurisdiction.value for s in strata} - - for t in targets: - source = t.source.value - year = t.period - jurisdiction = stratum_jurisdiction.get(t.stratum_id, "unknown") - - by_source[source]["count"] += 1 - by_source[source]["years"].add(year) - by_source[source]["variables"].add(t.variable) - - by_year[year]["count"] += 1 - by_year[year]["sources"].add(source) - - by_jurisdiction[jurisdiction]["count"] += 1 - by_jurisdiction[jurisdiction]["sources"].add(source) - - # Convert sets to lists for JSON - sources_summary = [] - for source, data in by_source.items(): - years = sorted(data["years"]) - sources_summary.append( - { - "source": source, - "display_name": source.upper().replace("-", " ").replace("_", " "), - "count": data["count"], - "variables": len(data["variables"]), - "year_min": min(years) if years else None, - "year_max": max(years) if years else None, - "years": years, - "is_projection": min(years) >= 2024 if years else False, - } - ) - - # Sort: historical sources first, then projections - sources_summary.sort(key=lambda x: (x["is_projection"], x["source"])) - - years_summary = [ - { - "year": year, - "count": data["count"], - "sources": sorted(data["sources"]), - } - for year, data in sorted(by_year.items()) - ] - - jurisdictions_summary = [ - { - "jurisdiction": jurisdiction, - "count": data["count"], - "sources": sorted(data["sources"]), - } - for jurisdiction, data in sorted(by_jurisdiction.items()) - ] - - # Sample targets for display - sample_targets = [] - for t in targets[:50]: - stratum = session.get(Stratum, t.stratum_id) - sample_targets.append( - { - "stratum": stratum.name if stratum else "Unknown", - "variable": t.variable, - "year": t.period, - "value": t.value, - "type": t.target_type.value, - "source": t.source.value, - } - ) - - summary = { - "generated_at": str(Path(DEFAULT_DB_PATH).stat().st_mtime), - "total_targets": len(targets), - "total_strata": len(strata), - "sources": sources_summary, - "years": years_summary, - "jurisdictions": jurisdictions_summary, - "sample_targets": sample_targets, - } - - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "w") as f: - json.dump(summary, f, indent=2) - - print(f"Exported summary to {output_path}") - print(f" {len(targets)} targets across {len(strata)} strata") - print( - f" {len(sources_summary)} sources: {', '.join(s['source'] for s in sources_summary)}" - ) - - return summary - - -if __name__ == "__main__": - export_targets_summary() diff --git a/tests/test_chronicle_boundaries.py b/tests/test_chronicle_boundaries.py index 2b509cb0..7fd0ec9a 100644 --- a/tests/test_chronicle_boundaries.py +++ b/tests/test_chronicle_boundaries.py @@ -3,14 +3,12 @@ from __future__ import annotations import ast -import hashlib import importlib.util import re import sys from pathlib import Path import pytest -import yaml FORBIDDEN_PACKAGE_ROOTS = {"calibration", "micro"} @@ -120,9 +118,10 @@ def test_load_cli_no_longer_accepts_retired_uk_legacy_etl_sources(): assert '"hmrc",' not in cli_source -# Legacy US loaders retired because their hand-typed values do not trace to -# publisher bytes, keyed by module name, with the `chronicle load` source each -# one answered to. +# Legacy US loaders, keyed by module name, with the `chronicle load` source each +# one answered to. The first twelve were retired because their hand-typed values +# do not trace to publisher bytes; `etl_soi` and `etl_snap` went with the legacy +# targets table they wrote, which nothing read. RETIRED_US_LEGACY_ETLS = { "etl_soi_state": "soi-state", "etl_soi_credits": "soi-credits", @@ -136,8 +135,16 @@ def test_load_cli_no_longer_accepts_retired_uk_legacy_etl_sources(): "etl_medicaid": "medicaid", "etl_ssa": "ssa", "etl_ssi": "ssi", + "etl_soi": "soi", + "etl_snap": "snap", +} +LEGACY_TARGET_TABLES = {"strata", "stratum_constraints", "targets"} +SOURCE_ARTIFACT_TABLES = { + "source_artifacts", + "source_columns", + "source_rows", + "source_tables", } -PACKAGE_BACKED_US_LOADERS = ["load_snap_targets", "load_soi_targets"] def test_retired_us_legacy_etl_modules_are_not_shipped(): @@ -149,11 +156,25 @@ def test_retired_us_legacy_etl_modules_are_not_shipped(): assert importlib.util.find_spec(f"db.{module}") is None -@pytest.mark.parametrize("source", sorted(RETIRED_US_LEGACY_ETLS.values())) -def test_load_cli_rejects_retired_us_legacy_sources(monkeypatch, capsys, source): +def test_no_legacy_db_etl_loaders_remain(): + db_root = Path(__file__).resolve().parents[1] / "db" + + assert sorted(path.name for path in db_root.glob("etl_*.py")) == [] + + +@pytest.mark.parametrize( + "argv", + [ + ["load", "all"], + *(["load", source] for source in sorted(RETIRED_US_LEGACY_ETLS.values())), + ["stats"], + ["query", "--variable", "snap_household_count"], + ], +) +def test_cli_rejects_retired_legacy_target_commands(monkeypatch, capsys, argv): from db import cli - monkeypatch.setattr(sys, "argv", ["chronicle", "load", source]) + monkeypatch.setattr(sys, "argv", ["chronicle", *argv]) with pytest.raises(SystemExit) as excinfo: cli.main() @@ -161,504 +182,81 @@ def test_load_cli_rejects_retired_us_legacy_sources(monkeypatch, capsys, source) assert "invalid choice" in capsys.readouterr().err -def test_us_loader_reexports_only_package_backed_loaders(): - import chronicle.jurisdictions.us as us_loaders - import chronicle.targets.loaders as target_loaders - - for module in (target_loaders, us_loaders): - assert sorted(module.__all__) == PACKAGE_BACKED_US_LOADERS - assert all(callable(getattr(module, name)) for name in module.__all__) - - -# The publisher measure each surviving legacy target row must reproduce, keyed by -# the row's variable and target type. IRS rows name the Publication 1304 table -# whose source package builds the fact. -IRS_ROW_MEASURES = { - ("tax_unit_count", "COUNT"): ( - "soi-table-1-1", - "irs_soi.individual_income_tax_returns", - ), - ("adjusted_gross_income", "AMOUNT"): ( - "soi-table-1-1", - "us:statutes/26/62#adjusted_gross_income", - ), - ("income_tax_liability", "AMOUNT"): ("soi-table-1-1", "irs_soi.total_income_tax"), - ("employment_income", "COUNT"): ( - "soi-table-1-4", - "irs_soi.returns_with_total_wages", - ), - ("employment_income", "AMOUNT"): ("soi-table-1-4", "us:statutes/26/62#input.wages"), -} -SNAP_ROW_MEASURES = { - ("snap_household_count", "COUNT"): ( - "usda_snap.average_monthly_households", - "count", - ), - ("snap_participant_count", "COUNT"): ("usda_snap.average_monthly_persons", "count"), - ("snap_benefits", "AMOUNT"): ("usda_snap.total_benefits", "dollars"), -} -IRS_AGI_CONCEPT = "us:statutes/26/62#adjusted_gross_income" -IRS_SOURCE_TABLES = {"soi-table-1-1": "Table 1.1", "soi-table-1-4": "Table 1.4"} - - -def _constraint_value(value: object) -> object: - try: - return float(value) - except (TypeError, ValueError): - return value - +def test_init_creates_only_source_artifact_tables(monkeypatch, tmp_path): + import sqlite3 -def _irs_stratum(fact) -> frozenset: - """The stratum definition a legacy row needs to carry this IRS fact.""" - bounds = { - ("adjusted_gross_income", c.operator, _constraint_value(c.value)) - for c in fact.constraints - if c.variable == IRS_AGI_CONCEPT - } - if len(bounds) != len(fact.constraints): - return frozenset({("unsupported fact constraint", "", "")}) - return frozenset(bounds or {("is_tax_filer", "==", 1.0)}) - - -def _snap_stratum(fact) -> frozenset: - """The stratum definition a legacy row needs to carry this SNAP fact.""" - stratum = {("snap", "==", 1.0)} - if fact.geography.id.startswith("0400000US"): - fips = fact.geography.id.removeprefix("0400000US") - stratum.add(("state_fips", "==", _constraint_value(fips))) - elif fact.geography.id != "0100000US" or fact.constraints: - return frozenset({("unsupported fact geography", "", "")}) - return frozenset(stratum) - - -# The top stratum of each surviving loader. Every other stratum it writes must -# hang directly under it. -NATIONAL_STRATA = { - "irs-soi": frozenset({("is_tax_filer", "==", 1.0)}), - "usda-snap": frozenset({("snap", "==", 1.0)}), -} + from db import cli + db_path = tmp_path / "chronicle.db" + monkeypatch.setattr(sys, "argv", ["chronicle", "--db", str(db_path), "init"]) + cli.main() -def _snap_provenance_note(fact, output_unit: str) -> str: - """The lineage a SNAP row must record for this fact, field by field. - - The trailing derivation records the loader's identity unit conversion - (factor 1), so the stored value is the published one. - """ - source = fact.source - parts = [f"USDA FNS {source.source_table}" if source.source_table else "USDA FNS"] - for key, value in ( - ("source_file", source.source_file), - ("vintage", source.vintage), - ("sha256", source.source_sha256), - ("raw", source.raw_r2_uri), - ("record", fact.source_record_id), - ): - if value: - parts.append(f"{key}={value}") - return ( - "; ".join(parts) + " Derived via: convert_units(factor=1, " - f"input_unit={fact.measure.unit}, output_unit={output_unit})." - ) - - -def _stratum_labels(fact, source: str) -> tuple[str, str, str]: - """The name, description and group a loader gives the stratum of this fact.""" - if source == "irs-soi": - band = fact.source_record_id.split(".")[3] - if band == "all": - return ( - "US All Filers", - "All individual income tax returns filed in the US", - "national", + with sqlite3.connect(db_path) as connection: + tables = { + name + for (name,) in connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" ) - return ( - f"US Filers AGI {band}", - f"Tax filers with AGI in {band} bracket", - "agi_brackets", - ) - if fact.geography.id == "0100000US": - return ( - "US SNAP Recipients", - "All SNAP recipient households/individuals in the US", - "snap_national", - ) - name = fact.geography.name - return f"{name} SNAP Recipients", f"SNAP recipients in {name}", "snap_states" - - -def _legacy_rows_without_publisher_fact(db_path: Path) -> list[tuple]: - """Return target rows that do not reproduce their own publisher cell. - - Each row must match exactly one source-package fact on measure, period and - the complete stratum definition (every constraint, with its operator), and - equal that fact's value. IRS rows must cite the Table 1.1 or 1.4 workbook - the fact was built from, by URL and table name, and the workbook's committed - bytes must hash to its manifest sha256. SNAP rows must cite the package's - source page and summary table, name the fact's source record and carry its - sha256, and record every other lineage field (file, vintage, raw URI) as the - fact does; IRS rows carry no notes. Each stratum must have the jurisdiction - the loader assigns that geography, the name, description and group the loader - gives it, and a definition hash that matches its constraints, and must sit - directly under the loader's top stratum, whose own parent is null. No row may add a margin of error, a preliminary flag or a geographic - level the fact does not support, or repeat a cell another row carries. - """ - from sqlmodel import Session, select - - from chronicle.source_package import load_source_package - from db.schema import ( - DataSource, - Jurisdiction, - Stratum, - StratumConstraint, - Target, - get_engine, - ) - - repo_root = Path(__file__).resolve().parents[1] - irs_files = {} - for table, package_id in ( - ("table_1_1", "soi-table-1-1"), - ("table_1_4", "soi-table-1-4"), - ): - table_dir = repo_root / "db" / "data" / "irs_soi" / table - manifest = yaml.safe_load((table_dir / "manifest.yaml").read_text()) - for year, spec in manifest["files"].items(): - content = (table_dir / spec["filename"]).read_bytes() - if hashlib.sha256(content).hexdigest() == spec["sha256"]: - irs_files[spec["source_url"]] = (package_id, int(year), spec) - snap_manifest = yaml.safe_load( - ( - repo_root - / "db" - / "data" - / "usda_snap" - / "fy69_to_current" - / "manifest.yaml" - ).read_text() - ) - - irs_facts: dict[tuple[str, int], list] = {} - snap_package = load_source_package("usda-snap-fy69-to-current") - snap_facts = snap_package.build_facts(snap_package.artifact.artifact_year) - - with Session(get_engine(db_path)) as session: - rows = session.exec(select(Target)).all() - stratum_rows = { - stratum.id: stratum for stratum in session.exec(select(Stratum)).all() } - strata: dict[int, set] = {} - raw_constraints: dict[int, list] = {} - for constraint in session.exec(select(StratumConstraint)).all(): - raw_constraints.setdefault(constraint.stratum_id, []).append( - (constraint.variable, constraint.operator, constraint.value) - ) - strata.setdefault(constraint.stratum_id, set()).add( - ( - constraint.variable, - constraint.operator, - _constraint_value(constraint.value), - ) - ) - untraced = [] - reproduced: set[tuple] = set() - for row in rows: - stratum = frozenset(strata.get(row.stratum_id, set())) - measure_key = (row.variable, row.target_type.name) - matched = [] - if ( - row.source == DataSource.IRS_SOI - and row.source_url in irs_files - and measure_key in IRS_ROW_MEASURES - ): - package_id, year, spec = irs_files[row.source_url] - row_package, concept = IRS_ROW_MEASURES[measure_key] - if (package_id, year) not in irs_facts: - irs_facts[package_id, year] = load_source_package( - package_id - ).build_facts(year) - if ( - row_package == package_id - and row.source_table == IRS_SOURCE_TABLES[package_id] - ): - matched = [ - fact - for fact in irs_facts[package_id, year] - if fact.measure.concept == concept - and fact.period.value == row.period - and _irs_stratum(fact) == stratum - and fact.source.source_file == spec["filename"] - and fact.source.source_sha256 == spec["sha256"] - ] - elif ( - row.source == DataSource.USDA_SNAP - and measure_key in SNAP_ROW_MEASURES - and row.source_url == snap_manifest["source_page"] - ): - record = re.search(r"record=([^;\s]+)", row.notes or "") - sha = re.search(r"sha256=([0-9a-f]{64})", row.notes or "") - matched = [ - fact - for fact in snap_facts - if fact.measure.concept == SNAP_ROW_MEASURES[measure_key][0] - and fact.period.value == row.period - and _snap_stratum(fact) == stratum - and row.source_table - == ( - "SNAP National Summary" - if fact.geography.id == "0100000US" - else "SNAP State Summary" - ) - and record is not None - and sha is not None - and fact.source_record_id == record.group(1) - and fact.source.source_sha256 == sha.group(1) - ] - fact = matched[0] if len(matched) == 1 else None - if fact is None: - untraced.append((row.source, row.variable, row.period, row.value)) - continue - national = fact.geography.id == "0100000US" - top = NATIONAL_STRATA[row.source] - row_stratum = stratum_rows[row.stratum_id] - parent = stratum_rows.get(row_stratum.parent_id) - if row.source == DataSource.IRS_SOI: - jurisdiction = Jurisdiction.US_FEDERAL - notes_ok = not row.notes - else: - jurisdiction = Jurisdiction.US_FEDERAL if national else Jurisdiction.US - notes_ok = (row.notes or "") == _snap_provenance_note( - fact, SNAP_ROW_MEASURES[measure_key][1] - ) - if stratum == top: - parent_ok = row_stratum.parent_id is None - else: - parent_ok = ( - parent is not None - and frozenset(strata.get(parent.id, set())) == top - and parent.parent_id is None - and parent.jurisdiction == Jurisdiction.US_FEDERAL - ) - if ( - fact.value != row.value - or (fact.source_record_id, fact.measure.concept) in reproduced - or row.margin_of_error is not None - or row.is_preliminary - or row.geographic_level not in (None, "national" if national else "state") - or row_stratum.jurisdiction != jurisdiction - or (row_stratum.name, row_stratum.description, row_stratum.stratum_group_id) - != _stratum_labels(fact, row.source) - or row_stratum.definition_hash - != Stratum.compute_hash( - raw_constraints.get(row.stratum_id, []), row_stratum.jurisdiction - ) - or not parent_ok - or not notes_ok - ): - untraced.append((row.source, row.variable, row.period, row.value)) - else: - reproduced.add((fact.source_record_id, fact.measure.concept)) - return untraced - - -@pytest.fixture(scope="module") -def loaded_legacy_db(tmp_path_factory): - """A targets database built by `chronicle load all`.""" - from db.cli import cmd_load - - db_path = tmp_path_factory.mktemp("legacy") / "targets.db" - cmd_load(type("Args", (), {"db": str(db_path), "source": "all", "years": None})) - return db_path - - -def test_every_legacy_target_row_equals_a_checksummed_publisher_fact(loaded_legacy_db): - """Every row `chronicle load all` writes reproduces its own publisher cell.""" - from sqlmodel import Session, select - - from db.schema import DataSource, Target, get_engine - - with Session(get_engine(loaded_legacy_db)) as session: - sources = {row.source for row in session.exec(select(Target)).all()} + assert tables == SOURCE_ARTIFACT_TABLES + assert not tables & LEGACY_TARGET_TABLES - assert _legacy_rows_without_publisher_fact(loaded_legacy_db) == [] - assert sources == {DataSource.IRS_SOI, DataSource.USDA_SNAP} +def test_legacy_target_api_is_retired(): + import chronicle.jurisdictions.us as us_jurisdiction + import chronicle.normalization as chronicle_normalization + import chronicle.targets as chronicle_targets + import db.schema as schema + import db.supabase_client as supabase_client + import policyengine_chronicle.targets as public_targets -_NATIONAL_2021 = "stratum_id = (select id from strata where name = 'US All Filers') and period = 2021" -_SNAP_STATE_ROW = ( - "(select t.{column} from targets t join stratum_constraints c " - "on c.stratum_id = t.stratum_id where c.variable = 'state_fips' " - "and c.value = '{fips}' and t.variable = 'snap_household_count')" -) -PROVENANCE_MUTATIONS = { - "every value set to 1.0": ("update targets set value = 1.0", 562), - "every period set to 1900": ("update targets set period = 1900", 562), - "SNAP checksums zeroed": ( - "update targets set notes = replace(notes, substr(notes, " - "instr(notes, 'sha256=') + 7, 64), '" + "0" * 64 + "') " - "where source = 'USDA_SNAP'", - 162, - ), - "one IRS value plus 1": ( - "update targets set value = value + 1 where id = " - "(select min(id) from targets where source = 'IRS_SOI')", - 1, - ), - "one SNAP value plus 1": ( - "update targets set value = value + 1 where id = " - "(select min(id) from targets where source = 'USDA_SNAP')", - 1, - ), - "national income tax replaced by national AGI from the same workbook": ( - f"update targets set value = (select value from targets where {_NATIONAL_2021} " - f"and variable = 'adjusted_gross_income') where {_NATIONAL_2021} " - "and variable = 'income_tax_liability'", - 1, - ), - "Maine SNAP households replaced by Connecticut's value and record": ( - "update targets set value = " - + _SNAP_STATE_ROW.format(column="value", fips="09") - + ", notes = " - + _SNAP_STATE_ROW.format(column="notes", fips="09") - + " where id = " - + _SNAP_STATE_ROW.format(column="id", fips="23"), - 1, - ), - "Maine's state_fips constraint negated": ( - "update stratum_constraints set operator = '!=' " - "where variable = 'state_fips' and value = '23'", - 3, - ), - "national SNAP households moved into an AGI band": ( - "update targets set stratum_id = " - "(select id from strata where name = 'US Filers AGI 50k_to_75k') " - "where variable = 'snap_household_count' and stratum_id = " - "(select id from strata where name = 'US SNAP Recipients')", - 1, - ), - "SNAP rows citing an IRS workbook": ( - "update targets set source_url = 'https://www.irs.gov/pub/irs-soi/23in11si.xls' " - "where source = 'USDA_SNAP'", - 162, - ), - "every source table renamed": ( - "update targets set source_table = 'Table 9.9'", - 562, - ), - "a valid row duplicated": ( - "insert into targets (stratum_id, variable, period, value, target_type, " - "source, source_table, source_url, notes, is_preliminary, created_at, " - "updated_at) select stratum_id, variable, period, value, target_type, source, " - "source_table, source_url, notes, is_preliminary, created_at, updated_at " - "from targets where source = 'USDA_SNAP' limit 1", - 1, - ), - "a margin of error added": ( - "update targets set margin_of_error = 1000 where id = (select min(id) from targets)", - 1, - ), - "a national row labelled state": ( - f"update targets set geographic_level = 'STATE' where {_NATIONAL_2021} " - "and variable = 'tax_unit_count'", - 1, - ), - # The 20 national rows, plus the 380 AGI-band rows whose parent it is. - "national IRS stratum moved to the UK": ( - "update strata set jurisdiction = 'UK' where name = 'US All Filers'", - 400, - ), - "SNAP rows naming an IRS workbook as their file": ( - "update targets set notes = replace(notes, " - "'source_file=snap-zip-fy69tocurrent-6.zip!FY24.xlsx', " - "'source_file=23in11si.xls') where source = 'USDA_SNAP'", - 162, - ), - "an AGI band re-parented under the SNAP stratum": ( - "update strata set parent_id = (select id from strata " - "where name = 'US SNAP Recipients') where name = 'US Filers AGI 50k_to_75k'", - 20, - ), - "an IRS row given a note": ( - "update targets set notes = 'sha256=' || hex(randomblob(32)) " - "where id = (select min(id) from targets where source = 'IRS_SOI')", - 1, - ), - "national IRS stratum renamed": ( - "update strata set name = 'Canada All Filers' where name = 'US All Filers'", - 20, - ), - "national IRS stratum described as the UK": ( - "update strata set description = " - "'All individual income tax returns filed in the UK' " - "where name = 'US All Filers'", - 20, - ), - "national IRS stratum filed under the AGI-band group": ( - "update strata set stratum_group_id = 'agi_brackets' " - "where name = 'US All Filers'", - 20, - ), - "SNAP state strata filed under the national group": ( - "update strata set stratum_group_id = 'snap_national' " - "where stratum_group_id = 'snap_states'", - 159, - ), - # A dangling parent on a top stratum flags its rows and every row under it. - "national IRS stratum given a dangling parent": ( - "update strata set parent_id = 999999 where name = 'US All Filers'", - 400, - ), - "national SNAP stratum given a dangling parent": ( - "update strata set parent_id = 999999 where name = 'US SNAP Recipients'", - 162, - ), - "national IRS definition hash zeroed": ( - "update strata set definition_hash = '0000000000000000' " - "where name = 'US All Filers'", - 20, - ), - # Several statements, because the hash column is unique. - "national and one AGI-band definition hashes swapped": ( - ( - "create temp table saved as select name, definition_hash from strata " - "where name in ('US All Filers', 'US Filers AGI 50k_to_75k')", - "update strata set definition_hash = 'swapping' where name = 'US All Filers'", - "update strata set definition_hash = (select definition_hash from saved " - "where name = 'US All Filers') where name = 'US Filers AGI 50k_to_75k'", - "update strata set definition_hash = (select definition_hash from saved " - "where name = 'US Filers AGI 50k_to_75k') where name = 'US All Filers'", - ), - 40, - ), - "row from a retired source": ( - "insert into targets (stratum_id, variable, period, value, target_type, " - "source, is_preliminary, created_at, updated_at) select stratum_id, " - "'ssi_recipients', 2023, 7425331, target_type, 'SSA', 0, created_at, " - "updated_at from targets limit 1", - 1, - ), -} - - -@pytest.mark.parametrize("mutation", sorted(PROVENANCE_MUTATIONS)) -def test_provenance_check_flags_rows_that_do_not_reproduce_their_cell( - loaded_legacy_db, tmp_path, mutation -): - import shutil - import sqlite3 + repo_root = Path(__file__).resolve().parents[1] + legacy_names = { + "DEFAULT_DB_PATH", + "DataSource", + "GeographicLevel", + "Jurisdiction", + "Stratum", + "StratumConstraint", + "Target", + "TargetType", + "get_engine", + "get_session", + "init_db", + "insert_targets_batch", + "query_strata", + "query_targets", + } - sql, expected = PROVENANCE_MUTATIONS[mutation] - statements = (sql,) if isinstance(sql, str) else sql - mutated = tmp_path / "mutated.db" - shutil.copy(loaded_legacy_db, mutated) - with sqlite3.connect(mutated) as connection: - for statement in statements: - changed = connection.execute(statement).rowcount - if statement.startswith(("update", "insert")): - assert changed >= 1 - - assert len(_legacy_rows_without_publisher_fact(mutated)) == expected + assert us_jurisdiction.__all__ == [] + assert importlib.util.find_spec("chronicle.targets.loaders") is None + assert importlib.util.find_spec("chronicle.targets.client") is None + assert not legacy_names & set(chronicle_targets.__all__) + assert not legacy_names & set(public_targets.__all__) + for name in legacy_names: + assert not hasattr(chronicle_targets, name) + with pytest.raises(AttributeError): + getattr(public_targets, name) + for name in ( + "GeographicLevel", + "Stratum", + "StratumConstraint", + "Target", + "TargetType", + ): + assert not hasattr(schema, name) + for name in ( + "TARGETS_SCHEMA", + "insert_targets_batch", + "query_strata", + "query_targets", + ): + assert not hasattr(supabase_client, name) + for name in ("TargetBlueprint", "as_target", "target_kwargs"): + assert not hasattr(chronicle_normalization, name) + assert importlib.util.find_spec("chronicle.normalization.targets") is None + assert not (repo_root / "scripts" / "export_to_json.py").exists() def test_us_legacy_etl_cleanup_doc_accounts_for_retired_series(): diff --git a/tests/test_chronicle_facts.py b/tests/test_chronicle_facts.py index f977bb38..329cf793 100644 --- a/tests/test_chronicle_facts.py +++ b/tests/test_chronicle_facts.py @@ -2,7 +2,7 @@ from chronicle.facts import DerivationStep, SourceFact from chronicle.sources import SourceFile, SourceReference -from chronicle.targets import DataSource, Jurisdiction +from db.schema import DataSource, Jurisdiction from chronicle.normalization import convert_units diff --git a/tests/test_chronicle_namespace.py b/tests/test_chronicle_namespace.py index 9b0ffd41..4b17c002 100644 --- a/tests/test_chronicle_namespace.py +++ b/tests/test_chronicle_namespace.py @@ -2,26 +2,16 @@ from chronicle.client import get_supabase_client from chronicle.normalization import convert_units -from chronicle.targets import ( - Target, - TargetType, - query_targets, +from chronicle.targets import TargetSourceCoverage, coverage_entries +from chronicle.targets.us_poverty import ( + TargetSourceCoverage as UsPovertyTargetSourceCoverage, ) -from db.schema import Target as DbTarget -from db.supabase_client import ( - LEDGER_SCHEMA, - TARGETS_SCHEMA, - query_targets as db_query_targets, -) - - -def test_chronicle_targets_reexport_schema_objects(): - assert Target is DbTarget - assert TargetType.COUNT.value == "count" +from db.supabase_client import LEDGER_SCHEMA -def test_chronicle_targets_reexport_client_helpers(): - assert query_targets is db_query_targets +def test_chronicle_targets_reexports_source_coverage_helpers(): + assert TargetSourceCoverage is UsPovertyTargetSourceCoverage + assert callable(coverage_entries) def test_chronicle_client_reexports_supabase_client(): @@ -30,7 +20,6 @@ def test_chronicle_client_reexports_supabase_client(): def test_chronicle_supabase_schema_boundaries_are_defaulted(): assert LEDGER_SCHEMA == "ledger" - assert TARGETS_SCHEMA == "targets" def test_chronicle_normalization_exports_helpers(): diff --git a/tests/test_chronicle_normalization.py b/tests/test_chronicle_normalization.py index c34a4c66..1296617c 100644 --- a/tests/test_chronicle_normalization.py +++ b/tests/test_chronicle_normalization.py @@ -2,13 +2,12 @@ import pytest -from chronicle.targets import DataSource, Jurisdiction, TargetType +from db.schema import DataSource, Jurisdiction from chronicle.normalization import ( SourceFact, apply_share, - as_target, convert_units, - target_kwargs, + format_derivation, ) @@ -30,7 +29,7 @@ def test_convert_units_records_derivation(): assert converted.derivation[0].parameters["factor"] == 1000 -def test_apply_share_and_as_target_preserve_provenance(): +def test_apply_share_preserves_provenance(): total = SourceFact( name="aca_marketplace_enrollment", value=21_446_150, @@ -42,23 +41,20 @@ def test_apply_share_and_as_target_preserve_provenance(): ) silver = apply_share(total, 0.54, name="silver_marketplace_enrollment") - target = as_target( - silver, - variable="aca_metal_level_enrollment", - target_type=TargetType.COUNT, - stratum_name="US ACA Marketplace Silver Enrollment", - constraints=(("metal_level", "==", "silver"),), - ) - assert target.value == pytest.approx(11_580_921) - assert target.source == DataSource.CMS_ACA - assert target.jurisdiction == Jurisdiction.US - assert target.source_table == "Marketplace OEP Report" - assert target.constraints == (("metal_level", "==", "silver"),) - assert "apply_share" in target.notes + assert silver.name == "silver_marketplace_enrollment" + assert silver.value == pytest.approx(11_580_921) + assert silver.period == 2024 + assert silver.source == "cms-aca" + assert silver.jurisdiction == "us" + assert silver.source_table == "Marketplace OEP Report" + assert silver.derivation[-1].operation == "apply_share" + assert silver.derivation[-1].parameters["total_fact"] == ( + "aca_marketplace_enrollment" + ) -def test_target_kwargs_maps_blueprint_to_target_fields(): +def test_format_derivation_lists_steps_in_order(): fact = SourceFact( name="snap_households", value=22_323, @@ -67,16 +63,12 @@ def test_target_kwargs_maps_blueprint_to_target_fields(): source=DataSource.USDA_SNAP, jurisdiction=Jurisdiction.US, ) - target = as_target( - convert_units(fact, 1000, "count"), - target_type=TargetType.COUNT, - stratum_name="US SNAP Households", - ) - kwargs = target_kwargs(target, stratum_id=7) + converted = convert_units(fact, 1000, "count") + shared = apply_share(converted, 0.5, name="half_snap_households") + + formatted = format_derivation(shared.derivation) - assert kwargs["stratum_id"] == 7 - assert kwargs["variable"] == "snap_households" - assert kwargs["value"] == 22_323_000 - assert kwargs["source"] == DataSource.USDA_SNAP - assert "convert_units" in kwargs["notes"] + assert formatted.startswith("convert_units(factor=1000") + assert "; apply_share(share=0.5" in formatted + assert format_derivation(()) == "" diff --git a/tests/test_chronicle_source_package.py b/tests/test_chronicle_source_package.py index 4c353a49..c28d2ff4 100644 --- a/tests/test_chronicle_source_package.py +++ b/tests/test_chronicle_source_package.py @@ -6,6 +6,7 @@ import hashlib from io import BytesIO from pathlib import Path +import re from zipfile import ZipFile import openpyxl @@ -4066,6 +4067,14 @@ def test_usda_snap_source_package_builds_fy24_national_and_state_facts(): assert len(cells) == 6_288 assert len(facts) == 216 assert all(fact.source.raw_r2_uri for fact in facts) + # The package artifact is the FY2024 workbook: every fact refers to fiscal + # 2024 and cites the workbook's checksum. + assert {(fact.period.type, fact.period.value) for fact in facts} == { + ("fiscal_year", 2024) + } + assert all( + re.fullmatch(r"[0-9a-f]{64}", fact.source.source_sha256 or "") for fact in facts + ) households = ( "usda_snap.fy2024.national_average_monthly_households." diff --git a/tests/test_congressional_districts.py b/tests/test_congressional_districts.py deleted file mode 100644 index d40c636a..00000000 --- a/tests/test_congressional_districts.py +++ /dev/null @@ -1,10 +0,0 @@ -"""Tests for the congressional district geographic level.""" - -from db.schema import GeographicLevel - - -def test_geographic_level_enum_exists(): - """GeographicLevel enum should include congressional_district.""" - assert GeographicLevel.CONGRESSIONAL_DISTRICT == "congressional_district" - assert GeographicLevel.STATE == "state" - assert GeographicLevel.NATIONAL == "national" diff --git a/tests/test_etl_snap.py b/tests/test_etl_snap.py deleted file mode 100644 index cc05be3b..00000000 --- a/tests/test_etl_snap.py +++ /dev/null @@ -1,247 +0,0 @@ -"""Tests for SNAP ETL. - -Values are asserted against the trustworthy USDA FNS source package -(``packages/usda_snap/fy69_to_current``) rather than any hardcoded table, so -the tests cannot silently bless fabricated numbers (see PolicyEngine/chronicle#77). -""" - -import tempfile -from pathlib import Path - -import pytest -from sqlmodel import Session, select - -from db.schema import ( - DataSource, - Stratum, - Target, - TargetType, - init_db, -) -from db.etl_snap import load_snap_facts, load_snap_targets - -# The FNS source package artifact is the FY2024 workbook. -SNAP_YEAR = 2024 - - -@pytest.fixture -def temp_db(): - """Create a temporary database for testing.""" - with tempfile.TemporaryDirectory() as tmpdir: - db_path = Path(tmpdir) / "test_snap.db" - engine = init_db(db_path) - yield engine - - -@pytest.fixture(scope="module") -def snap_facts(): - """Parsed SNAP facts from the trustworthy source package.""" - return load_snap_facts() - - -def _fact_value(facts, *, geography_id, concept): - """Return the parsed value for a geography/concept pair.""" - for fact in facts: - if fact.geography.id == geography_id and fact.measure.concept == concept: - return fact.value - raise AssertionError(f"no source fact for {geography_id} / {concept}") - - -class TestSnapSourceFacts: - """The ETL must source its values from the FNS source package.""" - - def test_facts_are_fiscal_year_2024(self, snap_facts): - """Every parsed fact carries the FY2024 reference period.""" - assert snap_facts - assert {fact.period.value for fact in snap_facts} == {SNAP_YEAR} - - def test_facts_carry_source_provenance(self, snap_facts): - """Each parsed fact traces to the checksum-locked FNS artifact.""" - for fact in snap_facts: - assert fact.source.source_name == "usda_snap" - assert fact.source.source_sha256 - assert fact.source.raw_r2_uri - - def test_year_filter_excludes_absent_periods(self): - """Filtering to a year with no facts yields nothing, not mislabeled data.""" - assert load_snap_facts([SNAP_YEAR]) - assert load_snap_facts([2021]) == [] - - -class TestSnapETL: - """Tests for SNAP ETL loader.""" - - def test_load_snap_creates_national_stratum(self, temp_db): - """Loading SNAP data should create a national stratum.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - national = session.exec( - select(Stratum).where(Stratum.name == "US SNAP Recipients") - ).first() - - assert national is not None - assert national.stratum_group_id == "snap_national" - - def test_load_snap_creates_national_targets(self, temp_db, snap_facts): - """National household target should match the parsed FNS value.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - national = session.exec( - select(Stratum).where(Stratum.name == "US SNAP Recipients") - ).first() - - hh_target = session.exec( - select(Target) - .where(Target.stratum_id == national.id) - .where(Target.variable == "snap_household_count") - .where(Target.period == SNAP_YEAR) - ).first() - - assert hh_target is not None - expected_hh = _fact_value( - snap_facts, - geography_id="0100000US", - concept="usda_snap.average_monthly_households", - ) - assert hh_target.value == expected_hh - assert hh_target.target_type == TargetType.COUNT - assert hh_target.source == DataSource.USDA_SNAP - # Provenance must be present and cite the FNS artifact. - assert "usda_snap" in hh_target.notes.lower() or "USDA FNS" in hh_target.notes - - def test_load_snap_creates_participant_targets(self, temp_db, snap_facts): - """Participant target should match the parsed FNS person count.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - national = session.exec( - select(Stratum).where(Stratum.name == "US SNAP Recipients") - ).first() - - participant_target = session.exec( - select(Target) - .where(Target.stratum_id == national.id) - .where(Target.variable == "snap_participant_count") - .where(Target.period == SNAP_YEAR) - ).first() - - assert participant_target is not None - expected = _fact_value( - snap_facts, - geography_id="0100000US", - concept="usda_snap.average_monthly_persons", - ) - assert participant_target.value == expected - - def test_load_snap_creates_benefit_targets(self, temp_db, snap_facts): - """Benefit target should match the parsed FNS total benefits.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - national = session.exec( - select(Stratum).where(Stratum.name == "US SNAP Recipients") - ).first() - - benefit_target = session.exec( - select(Target) - .where(Target.stratum_id == national.id) - .where(Target.variable == "snap_benefits") - .where(Target.period == SNAP_YEAR) - ).first() - - assert benefit_target is not None - expected = _fact_value( - snap_facts, - geography_id="0100000US", - concept="usda_snap.total_benefits", - ) - assert benefit_target.value == expected - assert benefit_target.target_type == TargetType.AMOUNT - - def test_load_snap_creates_state_strata(self, temp_db, snap_facts): - """Loading SNAP should create one stratum per parsed state geography.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - state_strata = session.exec( - select(Stratum).where(Stratum.stratum_group_id == "snap_states") - ).all() - - expected_states = { - fact.geography.id - for fact in snap_facts - if fact.geography.level == "state" - } - assert len(state_strata) == len(expected_states) - - def test_load_snap_state_targets_correct(self, temp_db, snap_facts): - """State-level SNAP targets should match parsed FNS values.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - ca_stratum = session.exec( - select(Stratum).where(Stratum.name == "California SNAP Recipients") - ).first() - - assert ca_stratum is not None - - ca_hh = session.exec( - select(Target) - .where(Target.stratum_id == ca_stratum.id) - .where(Target.variable == "snap_household_count") - ).first() - - expected_ca_hh = _fact_value( - snap_facts, - geography_id="0400000US06", - concept="usda_snap.average_monthly_households", - ) - assert ca_hh.value == expected_ca_hh - - def test_load_year_present(self, temp_db): - """Loading the available year should create national targets for it.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - national = session.exec( - select(Stratum).where(Stratum.name == "US SNAP Recipients") - ).first() - - targets = session.exec( - select(Target) - .where(Target.stratum_id == national.id) - .where(Target.variable == "snap_household_count") - ).all() - - years = {t.period for t in targets} - assert years == {SNAP_YEAR} - - def test_load_snap_idempotent(self, temp_db): - """Loading SNAP twice should not duplicate the national stratum.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - load_snap_targets(session, years=[SNAP_YEAR]) - - national_strata = session.exec( - select(Stratum).where(Stratum.name == "US SNAP Recipients") - ).all() - - # Should only have one national stratum - assert len(national_strata) == 1 - - def test_state_stratum_has_parent(self, temp_db): - """State strata should have national stratum as parent.""" - with Session(temp_db) as session: - load_snap_targets(session, years=[SNAP_YEAR]) - - national = session.exec( - select(Stratum).where(Stratum.name == "US SNAP Recipients") - ).first() - - ca = session.exec( - select(Stratum).where(Stratum.name == "California SNAP Recipients") - ).first() - - assert ca.parent_id == national.id diff --git a/tests/test_etl_soi.py b/tests/test_etl_soi.py deleted file mode 100644 index 4a1f7a84..00000000 --- a/tests/test_etl_soi.py +++ /dev/null @@ -1,38 +0,0 @@ -from db.etl_soi import ( - available_soi_table_1_4_years, - available_soi_years, - load_soi_table_1_1_data, - load_soi_table_1_4_data, -) - - -def test_available_soi_years_include_latest_table_1_1_sources(): - assert available_soi_years() == [2020, 2021, 2022, 2023] - - -def test_load_soi_table_1_1_data_reads_packaged_2023_source(): - data = load_soi_table_1_1_data(2023) - - assert data["source_url"] == "https://www.irs.gov/pub/irs-soi/23in11si.xls" - assert data["total_returns"] == 160_602_107 - assert data["total_agi"] == 15_286_017_359_000 - assert data["total_income_tax"] == 2_147_909_818_000 - assert data["returns_by_agi_bracket"]["100k_to_200k"] == 27_602_755 - assert data["agi_by_bracket"]["100k_to_200k"] == 3_818_295_141_000 - assert data["income_tax_by_bracket"]["100k_to_200k"] == 409_532_689_000 - - -def test_available_soi_years_include_latest_table_1_4_sources(): - assert available_soi_table_1_4_years() == [2020, 2021, 2022, 2023] - - -def test_load_soi_table_1_4_data_reads_packaged_2023_wage_source(): - data = load_soi_table_1_4_data(2023) - - assert data["source_url"] == "https://www.irs.gov/pub/irs-soi/23in14ar.xls" - assert data["total_employment_income_returns"] == 128_591_050 - assert data["total_employment_income"] == 10_204_095_705_000 - assert ( - data["employment_income_returns_by_agi_bracket"]["100k_to_200k"] == 23_193_910 - ) - assert data["employment_income_by_agi_bracket"]["100k_to_200k"] == 2_774_550_975_000 diff --git a/tests/test_policyengine_chronicle_imports.py b/tests/test_policyengine_chronicle_imports.py index f186b123..0d19b658 100644 --- a/tests/test_policyengine_chronicle_imports.py +++ b/tests/test_policyengine_chronicle_imports.py @@ -84,13 +84,15 @@ def test__given_retired_profile_namespace__then_it_is_not_importable() -> None: importlib.import_module("policyengine_chronicle.target_profiles") -def test__given_public_chronicle_normalization__then_target_construction_is_hidden() -> ( +def test__given_public_chronicle_normalization__then_target_construction_is_gone() -> ( None ): with pytest.raises(AttributeError): getattr(chronicle_normalization, "as_target") with pytest.raises(AttributeError): getattr(chronicle_normalization, "target_kwargs") + with pytest.raises(AttributeError): + getattr(chronicle_normalization, "TargetBlueprint") def test__given_chronicle_help__then_cli_does_not_eager_load_legacy_clients( @@ -107,3 +109,5 @@ def test__given_chronicle_help__then_cli_does_not_eager_load_legacy_clients( output = capsys.readouterr().out assert "Usage: chronicle [options]" in output assert "validate-facts" in output + assert " load\n" not in output + assert " stats\n" not in output diff --git a/tests/test_supabase_client.py b/tests/test_supabase_client.py index c0448844..e66cf6c0 100644 --- a/tests/test_supabase_client.py +++ b/tests/test_supabase_client.py @@ -106,35 +106,3 @@ def test_query_sources_returns_list(self, mock_client): assert isinstance(sources, list) assert len(sources) == 1 assert sources[0]["jurisdiction"] == "us" - - def test_query_targets_filters_by_jurisdiction(self, mock_client): - """Querying targets filters by jurisdiction.""" - from db.supabase_client import query_targets - - mock_client.table.return_value.select.return_value.eq.return_value.execute.return_value.data = [ - {"id": "456", "variable": "tax_unit_count", "value": 1000000} - ] - - with patch("db.supabase_client.get_supabase_client", return_value=mock_client): - targets = query_targets(jurisdiction="us") - mock_client.table.assert_called() - assert isinstance(targets, list) - - def test_query_strata_with_constraints(self, mock_client): - """Querying strata includes constraints.""" - from db.supabase_client import query_strata - - mock_client.table.return_value.select.return_value.execute.return_value.data = [ - { - "id": "789", - "name": "California filers", - "jurisdiction": "us", - "constraints": [ - {"variable": "state_fips", "operator": "==", "value": "06"} - ], - } - ] - - with patch("db.supabase_client.get_supabase_client", return_value=mock_client): - strata = query_strata() - assert isinstance(strata, list)