Add nonfiler income calibration targets (#994)

Author: MaxGhenis

changelog.d/994.added.md

@@ -0,0 +1 @@
+Added nonfiler-inclusive income calibration targets.

policyengine_us_data/calibration/unified_matrix_builder.py

@@ -36,6 +36,9 @@
 )
 from policyengine_us_data.pipeline_metadata import pipeline_node
 from policyengine_us_data.pipeline_schema import PipelineNode
+from policyengine_us_data.utils.target_variables import (
+    target_variable_components,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -854,6 +857,50 @@ def _evaluate_constraints_standalone(
     return np.array([hh_mask.get(hid, False) for hid in household_ids])
 
 
+def _target_expression_entity_from_map(
+    target_variable: str,
+    variable_entity_map: dict,
+) -> str:
+    entities = set()
+    missing = []
+    for component in target_variable_components(target_variable):
+        if component not in variable_entity_map:
+            missing.append(component)
+        else:
+            entities.add(variable_entity_map[component])
+    if missing:
+        raise ValueError(
+            f"Target expression {target_variable!r} includes variables "
+            f"with unknown entities: {missing}"
+        )
+    if len(entities) != 1:
+        raise ValueError(
+            "Additive target expressions must use variables with one "
+            f"entity; got {target_variable!r} with entities {entities}"
+        )
+    return entities.pop()
+
+
+def _sum_target_expression_values(
+    values_by_variable: dict,
+    target_variable: str,
+) -> np.ndarray | None:
+    result = None
+    for component in target_variable_components(target_variable):
+        values = values_by_variable.get(component)
+        if values is None:
+            return None
+        result = values if result is None else result + values
+    return result
+
+
+def _target_variables_for_calculation(target_variables) -> set[str]:
+    variables = set()
+    for variable in target_variables:
+        variables.update(target_variable_components(str(variable)))
+    return variables
+
+
 def _calculate_target_values_standalone(
     target_variable: str,
     non_geo_constraints: list,
@@ -875,7 +922,14 @@ def _calculate_target_values_standalone(
     (picklable, unlike ``tax_benefit_system``).
     """
     is_count = target_variable.endswith("_count")
-    target_entity = variable_entity_map.get(target_variable, "household")
+    is_expression = len(target_variable_components(target_variable)) > 1
+    if is_expression:
+        target_entity = _target_expression_entity_from_map(
+            target_variable,
+            variable_entity_map,
+        )
+    else:
+        target_entity = variable_entity_map.get(target_variable, "household")
 
     if reform_id > 0:
         mask = _evaluate_constraints_standalone(
@@ -898,7 +952,11 @@ def _calculate_target_values_standalone(
             household_ids,
             n_households,
         )
-        vals = hh_vars.get(target_variable)
+        vals = (
+            _sum_target_expression_values(hh_vars, target_variable)
+            if is_expression
+            else hh_vars.get(target_variable)
+        )
         if vals is None:
             return np.zeros(n_households, dtype=np.float32)
         return (vals * mask).astype(np.float32)
@@ -922,7 +980,11 @@ def _calculate_target_values_standalone(
         return hh_mask.astype(np.float32)
 
     if not is_count:
-        entity_values = target_entity_vars.get(target_variable)
+        entity_values = (
+            _sum_target_expression_values(target_entity_vars, target_variable)
+            if is_expression
+            else target_entity_vars.get(target_variable)
+        )
         entity_hh_idx = entity_hh_idx_map.get(target_entity)
         person_to_entity_idx = person_to_entity_idx_map.get(target_entity)
         if (
@@ -2601,7 +2663,10 @@ def build_matrix(
                 )
             )
 
-        unique_variables = set(targets_df["variable"].values)
+        target_variables = [
+            str(targets_df.iloc[i]["variable"]) for i in range(n_targets)
+        ]
+        unique_variables = _target_variables_for_calculation(target_variables)
         reform_variables = {
             str(row["variable"])
             for _, row in targets_df.iterrows()
@@ -2613,6 +2678,16 @@ def build_matrix(
                 variable_entity_map[var] = sim.tax_benefit_system.variables[
                     var
                 ].entity.key
+        for var in target_variables:
+            if len(target_variable_components(var)) > 1:
+                variable_entity_map[var] = _target_expression_entity_from_map(
+                    var,
+                    variable_entity_map,
+                )
+            elif var in sim.tax_benefit_system.variables:
+                variable_entity_map[var] = sim.tax_benefit_system.variables[
+                    var
+                ].entity.key
 
         # 5a. Collect unique constraint variables
         unique_constraint_vars = set()
@@ -3315,7 +3390,10 @@ def build_matrix_chunked(
                 )
             )
 
-        unique_variables = set(targets_df["variable"].values)
+        target_variables = [
+            str(targets_df.iloc[i]["variable"]) for i in range(n_targets)
+        ]
+        unique_variables = _target_variables_for_calculation(target_variables)
         reform_variables = {
             str(row["variable"])
             for _, row in targets_df.iterrows()
@@ -3327,9 +3405,6 @@ def build_matrix_chunked(
                 unique_constraint_vars.add(constraint["variable"])
 
         base_entity_maps = build_household_entity_maps(sim)
-        target_variables = [
-            str(targets_df.iloc[i]["variable"]) for i in range(n_targets)
-        ]
 
         if chunk_dir is None:
             chunk_root = Path(tempfile.mkdtemp(prefix="matrix_chunks_"))

policyengine_us_data/db/etl_irs_soi.py

@@ -70,6 +70,7 @@
         name="qualified_business_income_deduction",
         breakdown=None,
     ),
+    dict(code="00200", name="irs_employment_income", breakdown=None),
     dict(code="00900", name="total_self_employment_income", breakdown=None),
     dict(
         code="01000",

policyengine_us_data/db/etl_national_targets.py

@@ -9,6 +9,7 @@
     StratumConstraint,
     Target,
 )
+from policyengine_us_data.db.create_field_valid_values import FieldValidValues
 from policyengine_us_data.storage.calibration_targets.soi_metadata import (
     RETIREMENT_CONTRIBUTION_TARGETS,
 )
@@ -25,6 +26,114 @@
     etl_argparser,
     get_geographic_strata,
 )
+from policyengine_us_data.utils.target_variables import (
+    target_variable_components,
+)
+
+BEA_NIPA_WAGES_AND_SALARIES_2024 = 12_387_929_000_000
+BEA_NIPA_PROPRIETORS_INCOME_2024 = 2_023_080_000_000
+BEA_NIPA_PERSONAL_INTEREST_INCOME_2024 = 1_926_644_000_000
+BEA_NIPA_PERSONAL_DIVIDEND_INCOME_2024 = 2_218_700_000_000
+
+NIPA_PROPRIETORS_INCOME_VARIABLE = (
+    "total_self_employment_income+farm_operations_income+partnership_s_corp_income"
+)
+NIPA_PERSONAL_INTEREST_INCOME_VARIABLE = "interest_income"
+TAXABLE_INTEREST_AND_ORDINARY_DIVIDENDS_VARIABLE = (
+    "taxable_interest_income+dividend_income"
+)
+
+CBO_INCOME_BY_SOURCE_TARGETS = [
+    {
+        "variable": "irs_employment_income",
+        "parameter": "employment_income",
+        "notes": (
+            "CBO detailed AGI-by-source employment income; restricted to "
+            "tax filers because this is an AGI tax-return concept"
+        ),
+    },
+    {
+        "variable": "self_employment_income",
+        "parameter": "self_employment_income",
+        "notes": (
+            "CBO detailed AGI-by-source self-employment income; restricted "
+            "to tax filers because this is an AGI tax-return concept"
+        ),
+    },
+    {
+        "variable": "taxable_pension_income",
+        "parameter": "taxable_pension_income",
+        "notes": (
+            "CBO detailed AGI-by-source taxable pension income; restricted "
+            "to tax filers because this is an AGI tax-return concept"
+        ),
+    },
+    {
+        "variable": "taxable_social_security",
+        "parameter": "taxable_social_security",
+        "notes": (
+            "CBO detailed AGI-by-source taxable Social Security; restricted "
+            "to tax filers because this is an AGI tax-return concept"
+        ),
+    },
+    {
+        "variable": "qualified_dividend_income",
+        "parameter": "qualified_dividend_income",
+        "notes": (
+            "CBO detailed AGI-by-source qualified dividends; restricted to "
+            "tax filers because this is an AGI tax-return concept"
+        ),
+    },
+    {
+        "variable": "loss_limited_net_capital_gains",
+        "parameter": "net_capital_gain",
+        "notes": (
+            "CBO detailed AGI-by-source net capital gains; restricted to "
+            "tax filers because this is an AGI tax-return concept"
+        ),
+    },
+    {
+        "variable": TAXABLE_INTEREST_AND_ORDINARY_DIVIDENDS_VARIABLE,
+        "parameter": "taxable_interest_and_ordinary_dividends",
+        "notes": (
+            "CBO detailed AGI-by-source taxable interest plus ordinary "
+            "dividends; restricted to tax filers because this is an AGI "
+            "tax-return concept"
+        ),
+    },
+]
+
+
+def _register_target_variable(session: Session, variable: str) -> None:
+    from policyengine_us.system import system
+
+    missing = [
+        component
+        for component in target_variable_components(variable)
+        if component not in system.variables
+    ]
+    if missing:
+        raise ValueError(
+            f"Target variable expression {variable!r} includes unknown "
+            f"policyengine-us variables: {missing}"
+        )
+
+    existing = session.exec(
+        select(FieldValidValues).where(
+            FieldValidValues.field_name == "variable",
+            FieldValidValues.valid_value == variable,
+        )
+    ).first()
+    if existing is None:
+        session.add(
+            FieldValidValues(
+                field_name="variable",
+                valid_value=variable,
+                description="Additive calibration target expression",
+            )
+        )
+        session.flush()
+
 
 WIC_NATIONAL_ANNUAL_SUMMARY_SOURCE = (
     "https://www.fns.usda.gov/sites/default/files/resource-files/wisummary-4.xlsx"
@@ -305,7 +414,76 @@ def extract_national_targets(year: int = DEFAULT_YEAR):
     ]
     tax_expenditure_targets = [{**target} for target in raw_tax_expenditure_targets]
 
+    income_by_source = tax_benefit_system.parameters(
+        time_period
+    ).calibration.gov.cbo.income_by_source
+    for target in CBO_INCOME_BY_SOURCE_TARGETS:
+        try:
+            value = income_by_source._children[target["parameter"]]
+            tax_filer_targets.append(
+                {
+                    "variable": target["variable"],
+                    "value": float(value),
+                    "source": "CBO Revenue Projections",
+                    "notes": target["notes"],
+                    "year": time_period,
+                }
+            )
+        except (KeyError, AttributeError) as e:
+            print(
+                "Warning: Could not extract CBO income-by-source "
+                f"{target['parameter']} target: {e}"
+            )
+
     direct_sum_targets = [
+        {
+            "variable": "employment_income_before_lsr",
+            "value": BEA_NIPA_WAGES_AND_SALARIES_2024,
+            "source": "BEA NIPA Table 2.1",
+            "notes": (
+                "Gross wages and salaries for all workers, including "
+                "nonfilers; FRED/BEA series A034RC1A027NBEA"
+            ),
+            "year": 2024,
+        },
+        {
+            "variable": NIPA_PROPRIETORS_INCOME_VARIABLE,
+            "value": BEA_NIPA_PROPRIETORS_INCOME_2024,
+            "source": "BEA NIPA Table 2.1",
+            "notes": (
+                "Proprietors' income with IVA and CCAdj for all persons, "
+                "including nonfilers; FRED/BEA series A041RC1A027NBEA. "
+                "Mapped to the closest additive PolicyEngine aggregate: "
+                "total self-employment, farm operations, and "
+                "partnership/S-corp income."
+            ),
+            "year": 2024,
+        },
+        {
+            "variable": NIPA_PERSONAL_INTEREST_INCOME_VARIABLE,
+            "value": BEA_NIPA_PERSONAL_INTEREST_INCOME_2024,
+            "source": "BEA NIPA Table 2.1",
+            "notes": (
+                "Personal interest income for all persons, including "
+                "nonfilers; FRED/BEA series A064RC1A027NBEA. NIPA also "
+                "includes imputed interest, so this is a macro benchmark "
+                "rather than a pure tax concept."
+            ),
+            "year": 2024,
+        },
+        {
+            "variable": "dividend_income",
+            "value": BEA_NIPA_PERSONAL_DIVIDEND_INCOME_2024,
+            "source": "BEA NIPA Table 2.1",
+            "notes": (
+                "Personal dividend income for all persons, including "
+                "nonfilers; FRED/BEA series B703RC1A027NBEA. NIPA "
+                "includes dividends received through pension funds and "
+                "private trusts, so this is a macro benchmark rather than "
+                "a pure tax concept."
+            ),
+            "year": 2024,
+        },
         {
             "variable": "medicaid",
             "value": 871.7e9,
@@ -701,6 +879,7 @@ def load_national_targets(
         # Process direct sum targets
         for _, target_data in direct_targets_df.iterrows():
             target_year = target_data["year"]
+            _register_target_variable(session, target_data["variable"])
             # Check if target already exists
             existing_target = session.exec(
                 select(Target).where(
@@ -767,6 +946,7 @@ def load_national_targets(
             # Add tax-related targets to filer stratum
             for _, target_data in tax_filer_df.iterrows():
                 target_year = target_data["year"]
+                _register_target_variable(session, target_data["variable"])
                 # Check if target already exists
                 existing_target = session.exec(
                     select(Target).where(

policyengine_us_data/db/validate_database.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 
 import pandas as pd
+from policyengine_us_data.utils.target_variables import target_variable_is_valid
 
 
 DEFAULT_DB_PATH = (
@@ -34,7 +35,7 @@ def validate_database(db_path: str | Path = DEFAULT_DB_PATH) -> None:
         targets_df = pd.read_sql("SELECT * FROM targets", conn)
 
         for var_name in set(targets_df["variable"]):
-            if var_name not in system.variables:
+            if not target_variable_is_valid(var_name, system.variables):
                 raise ValueError(f"{var_name} not a policyengine-us variable")
 
         for var_name in set(stratum_constraints_df["constraint_variable"]):