Fix stale calibration targets by deriving time_period from dataset (#505)

* Fix stale calibration targets by deriving time_period from dataset

- Remove hardcoded CBO_YEAR and TREASURY_YEAR constants
- Add --dataset CLI argument to etl_national_targets.py
- Derive time_period from sim.default_calculation_period
- Default to HuggingFace production dataset

The dataset itself is now the single source of truth for the
calibration year, preventing future drift when updating to new
base years.

Closes #503

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Use income_tax_positive for CBO calibration in loss.py

The CBO income_tax parameter represents positive-only receipts (refundable
credit payments in excess of liability are classified as outlays, not
negative receipts). Using income_tax_positive matches this definition.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Add --dataset argument to all database ETL scripts

All ETL scripts now derive their target year from the dataset's
default_calculation_period instead of hardcoding years. This ensures
all calibration targets stay synchronized when updating to a new
base year annually.

Updated scripts:
- create_initial_strata.py
- etl_age.py
- etl_irs_soi.py (with configurable --lag for IRS data delay)
- etl_medicaid.py
- etl_snap.py
- etl_state_income_tax.py

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Add 119th Congress district code support for 2024 ACS data

- Update parse_ucgid to recognize both 5001800US (118th) and 5001900US (119th Congress)
- Expand Puerto Rico and territory filters to handle both Congress code formats
- Update TERRITORY_UCGIDS and NON_VOTING_GEO_IDS with 119th Congress codes

This ensures consistent redistricting alignment: 2024 ACS data uses 119th Congress
codes natively, and IRS SOI data is converted via the 116th→119th mapping matrix.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

* Remove seed-related changes to reduce PR scope

Revert deterministic hash-based medicaid/SSI seed logic in cps.py,
update Makefile seed to 3526.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Upgrade policyengine-us to 1.550.1 in uv.lock

Needed for income_tax_positive variable used in loss.py.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Cherry-pick ACA PTC targets from PR #508 and update changelog

Adds aca_ptc ingestion from IRS SOI data (code 85530) to etl_irs_soi.py
and updates DATABASE_GUIDE.md to reflect stratum_group_id 119.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Split local area publish into build+stage and promote phases

Prevents silent no-op promotes by detecting when HF commits don't
change HEAD. Adds separate promote workflow for manual gate before
pushing staging files to production. Also bumps calibration epochs
from 200 to 250.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: baogorek

.github/workflows/local_area_promote.yaml

@@ -0,0 +1,43 @@
+name: Promote Local Area H5 Files
+
+on:
+  workflow_dispatch:
+    inputs:
+      version:
+        description: 'Version to promote (e.g. 1.23.0)'
+        required: true
+        type: string
+      branch:
+        description: 'Branch to use for repo setup'
+        required: false
+        default: 'main'
+        type: string
+
+jobs:
+  promote-local-area:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    env:
+      HUGGING_FACE_TOKEN: ${{ secrets.HUGGING_FACE_TOKEN }}
+      MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }}
+      MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }}
+
+    steps:
+      - name: Checkout repo
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.13'
+
+      - name: Install Modal CLI
+        run: pip install modal
+
+      - name: Promote staged files to production
+        run: |
+          VERSION="${{ github.event.inputs.version }}"
+          BRANCH="${{ github.event.inputs.branch }}"
+          echo "Promoting version ${VERSION} from branch ${BRANCH}"
+          modal run modal_app/local_area.py::main_promote --version="${VERSION}" --branch="${BRANCH}"

.github/workflows/local_area_publish.yaml

@@ -49,7 +49,7 @@ jobs:
       - name: Install Modal CLI
         run: pip install modal
 
-      - name: Run local area publishing on Modal
+      - name: Run local area build and stage on Modal
         run: |
           NUM_WORKERS="${{ github.event.inputs.num_workers || '8' }}"
           SKIP_UPLOAD="${{ github.event.inputs.skip_upload || 'false' }}"
@@ -63,3 +63,13 @@ jobs:
 
           echo "Running: $CMD"
           $CMD
+
+      - name: Post-build summary
+        if: success()
+        run: |
+          echo "## Build + Stage Complete" >> $GITHUB_STEP_SUMMARY
+          echo "" >> $GITHUB_STEP_SUMMARY
+          echo "Files have been uploaded to GCS and staged on HuggingFace." >> $GITHUB_STEP_SUMMARY
+          echo "" >> $GITHUB_STEP_SUMMARY
+          echo "### Next step: Promote to production" >> $GITHUB_STEP_SUMMARY
+          echo "Trigger the **Promote Local Area H5 Files** workflow with the version from the build output." >> $GITHUB_STEP_SUMMARY

Makefile

@@ -87,7 +87,7 @@ data: download
 	python policyengine_us_data/datasets/cps/extended_cps.py
 	python policyengine_us_data/datasets/cps/enhanced_cps.py
 	python policyengine_us_data/datasets/cps/small_enhanced_cps.py
-	python policyengine_us_data/datasets/cps/local_area_calibration/create_stratified_cps.py 10500
+	python policyengine_us_data/datasets/cps/local_area_calibration/create_stratified_cps.py 12000 --top=99.5 --seed=3526
 
 publish-local-area:
 	python policyengine_us_data/datasets/cps/local_area_calibration/publish_local_area.py

changelog_entry.yaml

@@ -0,0 +1,6 @@
+- date: 2026-02-02
+  type: fixed
+  description: Fix stale calibration targets by deriving time_period from dataset across all ETL scripts, using income_tax_positive for CBO calibration, and adding 119th Congress district code support for consistent redistricting alignment
+- date: 2026-02-07
+  type: added
+  description: Add ACA Premium Tax Credit targets from IRS SOI data (cherry-picked from PR #508)

modal_app/local_area.py

@@ -259,14 +259,11 @@ def validate_staging(branch: str, version: str) -> Dict:
     memory=8192,
     timeout=14400,
 )
-def atomic_upload(branch: str, version: str, manifest: Dict) -> str:
+def upload_to_staging(branch: str, version: str, manifest: Dict) -> str:
     """
-    Upload files using staging approach for atomic deployment.
+    Upload files to GCS (production) and HuggingFace (staging only).
 
-    1. Upload to GCS (direct, overwrites existing)
-    2. Upload to HuggingFace staging/ folder
-    3. Atomically promote staging/ to production paths
-    4. Clean up staging/
+    Promote must be run separately via promote_publish.
     """
     setup_gcp_credentials()
     setup_repo(branch)
@@ -286,8 +283,6 @@ def atomic_upload(branch: str, version: str, manifest: Dict) -> str:
 from policyengine_us_data.utils.data_upload import (
     upload_local_area_file,
     upload_to_staging_hf,
-    promote_staging_to_production_hf,
-    cleanup_staging_hf,
 )
 
 manifest = json.loads('''{manifest_json}''')
@@ -306,11 +301,9 @@ def atomic_upload(branch: str, version: str, manifest: Dict) -> str:
 print(f"Verified {{verification['verified']}} files")
 
 files_with_paths = []
-rel_paths = []
 for rel_path in manifest["files"].keys():
     local_path = version_dir / rel_path
     files_with_paths.append((local_path, rel_path))
-    rel_paths.append(rel_path)
 
 # Upload to GCS (direct to production paths)
 print(f"Uploading {{len(files_with_paths)}} files to GCS...")
@@ -331,12 +324,73 @@ def atomic_upload(branch: str, version: str, manifest: Dict) -> str:
 hf_count = upload_to_staging_hf(files_with_paths, version)
 print(f"Uploaded {{hf_count}} files to HuggingFace staging/")
 
-# Atomically promote staging to production
-print("Promoting staging/ to production (atomic commit)...")
+print(f"Staged version {{version}} for promotion")
+""",
+        ],
+        text=True,
+        env=os.environ.copy(),
+    )
+
+    if result.returncode != 0:
+        raise RuntimeError(f"Upload failed: {result.stderr}")
+
+    return (
+        f"Staged version {version} with {len(manifest['files'])} files. "
+        f"Run promote workflow to publish to HuggingFace production."
+    )
+
+
+@app.function(
+    image=image,
+    secrets=[hf_secret],
+    volumes={VOLUME_MOUNT: staging_volume},
+    memory=4096,
+    timeout=3600,
+)
+def promote_publish(branch: str = "main", version: str = "") -> str:
+    """
+    Promote staged files from HF staging/ to production paths, then cleanup.
+
+    Reads the manifest from the Modal staging volume to determine which
+    files to promote.
+    """
+    setup_repo(branch)
+
+    staging_dir = Path(VOLUME_MOUNT)
+    staging_volume.reload()
+
+    manifest_path = staging_dir / version / "manifest.json"
+    if not manifest_path.exists():
+        raise RuntimeError(
+            f"No manifest found at {manifest_path}. "
+            f"Run build+stage workflow first."
+        )
+
+    with open(manifest_path) as f:
+        manifest = json.load(f)
+
+    rel_paths_json = json.dumps(list(manifest["files"].keys()))
+
+    result = subprocess.run(
+        [
+            "uv",
+            "run",
+            "python",
+            "-c",
+            f"""
+import json
+from policyengine_us_data.utils.data_upload import (
+    promote_staging_to_production_hf,
+    cleanup_staging_hf,
+)
+
+rel_paths = json.loads('''{rel_paths_json}''')
+version = "{version}"
+
+print(f"Promoting {{len(rel_paths)}} files from staging/ to production...")
 promoted = promote_staging_to_production_hf(rel_paths, version)
 print(f"Promoted {{promoted}} files to production")
 
-# Clean up staging
 print("Cleaning up staging/...")
 cleaned = cleanup_staging_hf(rel_paths, version)
 print(f"Cleaned up {{cleaned}} files from staging/")
@@ -349,9 +403,9 @@ def atomic_upload(branch: str, version: str, manifest: Dict) -> str:
     )
 
     if result.returncode != 0:
-        raise RuntimeError(f"Upload failed: {result.stderr}")
+        raise RuntimeError(f"Promote failed: {result.stderr}")
 
-    return f"Successfully published version {version} with {len(manifest['files'])} files"
+    return f"Successfully promoted version {version} with {len(manifest['files'])} files"
 
 
 @app.function(
@@ -544,10 +598,24 @@ def coordinate_publish(
             f"WARNING: Expected {expected_total} files, found {actual_total}"
         )
 
-    print("\nStarting atomic upload...")
-    result = atomic_upload.remote(
+    print("\nStarting upload to staging...")
+    result = upload_to_staging.remote(
         branch=branch, version=version, manifest=manifest
     )
+    print(result)
+
+    print("\n" + "=" * 60)
+    print("BUILD + STAGE COMPLETE")
+    print("=" * 60)
+    print(
+        f"To promote to HuggingFace production, run the "
+        f"'Promote Local Area H5 Files' workflow with version={version}"
+    )
+    print(
+        "Or run manually: modal run modal_app/local_area.py::main_promote "
+        f"--version={version}"
+    )
+    print("=" * 60)
 
     return result
 
@@ -565,3 +633,15 @@ def main(
         skip_upload=skip_upload,
     )
     print(result)
+
+
+@app.local_entrypoint()
+def main_promote(
+    version: str = "",
+    branch: str = "main",
+):
+    """Promote staged files to HuggingFace production."""
+    if not version:
+        raise ValueError("--version is required")
+    result = promote_publish.remote(branch=branch, version=version)
+    print(result)

policyengine_us_data/datasets/cps/enhanced_cps.py

@@ -196,7 +196,7 @@ def generate(self):
                 loss_matrix_clean,
                 targets_array_clean,
                 log_path="calibration_log.csv",
-                epochs=200,
+                epochs=250,
                 seed=1456,
             )
             data["household_weight"][year] = optimised_weights

policyengine_us_data/datasets/cps/local_area_calibration/calibration_utils.py

@@ -252,39 +252,24 @@ def get_pseudo_input_variables(sim) -> set:
     """
     Identify pseudo-input variables that should NOT be saved to H5 files.
 
-    A pseudo-input is a variable that:
-    - Appears in sim.input_variables (has stored values)
-    - Has 'adds' or 'subtracts' attribute
-    - At least one component has a formula (is calculated)
-
-    These variables have stale pre-computed values that corrupt calculations
-    when reloaded, because the stored value overrides the formula.
+    NOTE: This function currently returns an empty set. The original logic
+    excluded variables with 'adds' or 'subtracts' attributes, but analysis
+    showed that in CPS data, these variables contain authoritative stored
+    data that does NOT match their component variables:
+
+    - pre_tax_contributions: components are all 0, aggregate has imputed values
+    - tax_exempt_pension_income: aggregate has 135M, components only 20M
+    - taxable_pension_income: aggregate has 82M, components only 29M
+    - interest_deduction: aggregate has 41M, components are 0
+
+    The 'adds' attribute defines how to CALCULATE these values, but in CPS
+    data the stored values are the authoritative source. Excluding them and
+    recalculating from components produces incorrect results.
+
+    For geo-stacking, entity ID reindexing preserves within-entity
+    relationships, so aggregation within a person or tax_unit remains valid.
     """
-    tbs = sim.tax_benefit_system
-    pseudo_inputs = set()
-
-    for var_name in sim.input_variables:
-        var = tbs.variables.get(var_name)
-        if not var:
-            continue
-
-        adds = getattr(var, "adds", None)
-        if adds and isinstance(adds, list):
-            for component in adds:
-                comp_var = tbs.variables.get(component)
-                if comp_var and len(getattr(comp_var, "formulas", {})) > 0:
-                    pseudo_inputs.add(var_name)
-                    break
-
-        subtracts = getattr(var, "subtracts", None)
-        if subtracts and isinstance(subtracts, list):
-            for component in subtracts:
-                comp_var = tbs.variables.get(component)
-                if comp_var and len(getattr(comp_var, "formulas", {})) > 0:
-                    pseudo_inputs.add(var_name)
-                    break
-
-    return pseudo_inputs
+    return set()
 
 
 def apply_op(values: np.ndarray, op: str, val: str) -> np.ndarray:

policyengine_us_data/db/DATABASE_GUIDE.md

@@ -110,7 +110,7 @@ The `stratum_group_id` field categorizes strata:
 | 5 | Medicaid | Medicaid enrollment strata |
 | 6 | EITC | EITC recipients by qualifying children |
 | 7 | State Income Tax | State-level income tax collections (Census STC) |
-| 100-118 | IRS Conditional | Each IRS variable paired with conditional count constraints |
+| 100-119 | IRS Conditional | Each IRS variable paired with conditional count constraints (includes ACA PTC at 119) |
 
 ### Conditional Strata (IRS SOI)
 

policyengine_us_data/db/create_initial_strata.py

@@ -1,3 +1,4 @@
+import argparse
 import logging
 from typing import Dict
 
@@ -6,6 +7,8 @@
 from sqlmodel import Session, create_engine
 
 from policyengine_us_data.storage import STORAGE_FOLDER
+
+DEFAULT_DATASET = "hf://policyengine/policyengine-us-data/calibration/stratified_extended_cps.h5"
 from policyengine_us_data.db.create_database_tables import (
     Stratum,
     StratumConstraint,
@@ -68,6 +71,28 @@ def fetch_congressional_districts(year):
 
 
 def main():
+    parser = argparse.ArgumentParser(
+        description="Create initial geographic strata for calibration"
+    )
+    parser.add_argument(
+        "--dataset",
+        default=DEFAULT_DATASET,
+        help=(
+            "Source dataset (local path or HuggingFace URL). "
+            "The year for Census API calls is derived from the dataset's "
+            "default_calculation_period. Default: %(default)s"
+        ),
+    )
+    args = parser.parse_args()
+
+    # Derive year from dataset
+    from policyengine_us import Microsimulation
+
+    print(f"Loading dataset: {args.dataset}")
+    sim = Microsimulation(dataset=args.dataset)
+    year = int(sim.default_calculation_period)
+    print(f"Derived year from dataset: {year}")
+
     # State FIPS to name/abbreviation mapping
     STATE_NAMES = {
         1: "Alabama (AL)",
@@ -123,8 +148,7 @@ def main():
         56: "Wyoming (WY)",
     }
 
-    # Fetch congressional district data for year 2023
-    year = 2023
+    # Fetch congressional district data
     cd_df = fetch_congressional_districts(year)
 
     DATABASE_URL = (