Enable cliff impacts

changelog_entry.yaml

@@ -0,0 +1,4 @@
+- bump: patch
+  changes:
+    fixed:
+    - Added cliff impacts.

policyengine/outputs/macro/comparison/calculate_economy_comparison.py

@@ -775,6 +775,16 @@ def uk_constituency_breakdown(
     return UKConstituencyBreakdownWithValues(**output)
 
 
+class CliffImpactInSimulation(BaseModel):
+    cliff_gap: float
+    cliff_share: float
+
+
+class CliffImpact(BaseModel):
+    baseline: CliffImpactInSimulation
+    reform: CliffImpactInSimulation
+
+
 class EconomyComparison(BaseModel):
     country_package_version: str
     budget: BudgetaryImpact
@@ -789,6 +799,7 @@ class EconomyComparison(BaseModel):
     intra_wealth_decile: IntraWealthDecileImpact
     labor_supply_response: LaborSupplyResponse
     constituency_impact: UKConstituencyBreakdown
+    cliff_impact: CliffImpact | None
 
 
 def calculate_economy_comparison(
@@ -802,51 +813,54 @@ def calculate_economy_comparison(
     reform: SingleEconomy = simulation.calculate_single_economy(reform=True)
     options = simulation.options
     country_id = options.country
-    if baseline.type == "general":
-        budgetary_impact_data = budgetary_impact(baseline, reform)
-        detailed_budgetary_impact_data = detailed_budgetary_impact(
-            baseline, reform, country_id
-        )
-        decile_impact_data = decile_impact(baseline, reform)
-        inequality_impact_data = inequality_impact(baseline, reform)
-        poverty_impact_data = poverty_impact(baseline, reform)
-        poverty_by_gender_data = poverty_gender_breakdown(baseline, reform)
-        poverty_by_race_data = poverty_racial_breakdown(baseline, reform)
-        intra_decile_impact_data = intra_decile_impact(baseline, reform)
-        labor_supply_response_data = labor_supply_response(baseline, reform)
-        constituency_impact_data: UKConstituencyBreakdown = (
-            uk_constituency_breakdown(baseline, reform, country_id)
-        )
-        wealth_decile_impact_data = wealth_decile_impact(
-            baseline, reform, country_id
-        )
-        intra_wealth_decile_impact_data = intra_wealth_decile_impact(
-            baseline, reform, country_id
-        )
-
-        return EconomyComparison(
-            country_package_version=get_country_package_version(country_id),
-            budget=budgetary_impact_data,
-            detailed_budget=detailed_budgetary_impact_data,
-            decile=decile_impact_data,
-            inequality=inequality_impact_data,
-            poverty=poverty_impact_data,
-            poverty_by_gender=poverty_by_gender_data,
-            poverty_by_race=poverty_by_race_data,
-            intra_decile=intra_decile_impact_data,
-            wealth_decile=wealth_decile_impact_data,
-            intra_wealth_decile=intra_wealth_decile_impact_data,
-            labor_supply_response=labor_supply_response_data,
-            constituency_impact=constituency_impact_data,
-        )
-    elif baseline.type == "cliff":
-        return dict(
-            baseline=dict(
+    budgetary_impact_data = budgetary_impact(baseline, reform)
+    detailed_budgetary_impact_data = detailed_budgetary_impact(
+        baseline, reform, country_id
+    )
+    decile_impact_data = decile_impact(baseline, reform)
+    inequality_impact_data = inequality_impact(baseline, reform)
+    poverty_impact_data = poverty_impact(baseline, reform)
+    poverty_by_gender_data = poverty_gender_breakdown(baseline, reform)
+    poverty_by_race_data = poverty_racial_breakdown(baseline, reform)
+    intra_decile_impact_data = intra_decile_impact(baseline, reform)
+    labor_supply_response_data = labor_supply_response(baseline, reform)
+    constituency_impact_data: UKConstituencyBreakdown = (
+        uk_constituency_breakdown(baseline, reform, country_id)
+    )
+    wealth_decile_impact_data = wealth_decile_impact(
+        baseline, reform, country_id
+    )
+    intra_wealth_decile_impact_data = intra_wealth_decile_impact(
+        baseline, reform, country_id
+    )
+
+    if simulation.options.include_cliffs:
+        cliff_impact = CliffImpact(
+            baseline=CliffImpactInSimulation(
                 cliff_gap=baseline.cliff_gap,
                 cliff_share=baseline.cliff_share,
             ),
-            reform=dict(
+            reform=CliffImpactInSimulation(
                 cliff_gap=reform.cliff_gap,
                 cliff_share=reform.cliff_share,
             ),
         )
+    else:
+        cliff_impact = None
+
+    return EconomyComparison(
+        country_package_version=get_country_package_version(country_id),
+        budget=budgetary_impact_data,
+        detailed_budget=detailed_budgetary_impact_data,
+        decile=decile_impact_data,
+        inequality=inequality_impact_data,
+        poverty=poverty_impact_data,
+        poverty_by_gender=poverty_by_gender_data,
+        poverty_by_race=poverty_by_race_data,
+        intra_decile=intra_decile_impact_data,
+        wealth_decile=wealth_decile_impact_data,
+        intra_wealth_decile=intra_wealth_decile_impact_data,
+        labor_supply_response=labor_supply_response_data,
+        constituency_impact=constituency_impact_data,
+        cliff_impact=cliff_impact,
+    )

policyengine/outputs/macro/single/calculate_single_economy.py

@@ -10,6 +10,8 @@
 from policyengine_core.simulations import Microsimulation
 from typing import Dict
 from dataclasses import dataclass
+from typing import Literal
+from microdf import MicroSeries
 
 
 class SingleEconomy(BaseModel):
@@ -47,8 +49,10 @@ class SingleEconomy(BaseModel):
     weekly_hours: float | None
     weekly_hours_income_effect: float | None
     weekly_hours_substitution_effect: float | None
-    type: str
+    type: Literal["general", "cliff"]
     programs: Dict[str, float] | None
+    cliff_gap: float | None = None
+    cliff_share: float | None = None
 
 
 @dataclass
@@ -327,10 +331,27 @@ def calculate_uk_programs(self) -> Dict[str, float]:
             for program in UKPrograms.PROGRAMS
         }
 
+    def calculate_cliffs(self):
+        cliff_gap: MicroSeries = self.simulation.calculate("cliff_gap")
+        is_on_cliff: MicroSeries = self.simulation.calculate("is_on_cliff")
+        total_cliff_gap: float = cliff_gap.sum()
+        total_adults: float = self.simulation.calculate("is_adult").sum()
+        cliff_share: float = is_on_cliff.sum() / total_adults
+        return CliffImpactInSimulation(
+            cliff_gap=total_cliff_gap,
+            cliff_share=cliff_share,
+        )
+
+
+class CliffImpactInSimulation(BaseModel):
+    cliff_gap: float
+    cliff_share: float
+
 
 def calculate_single_economy(
     simulation: Simulation, reform: bool = False
 ) -> Dict:
+    include_cliffs = simulation.options.include_cliffs
     task_manager = GeneralEconomyTask(
         (
             simulation.baseline_simulation
@@ -382,6 +403,14 @@ def calculate_single_economy(
         except:
             total_state_tax = 0
 
+    if include_cliffs:
+        cliffs = task_manager.calculate_cliffs()
+        cliff_gap = cliffs.cliff_gap
+        cliff_share = cliffs.cliff_share
+    else:
+        cliff_gap = None
+        cliff_share = None
+
     return SingleEconomy(
         **{
             "total_net_income": total_net_income,
@@ -414,7 +443,9 @@ def calculate_single_economy(
             "age": age,
             **labor_supply_responses,
             **lsr_working_hours,
-            "type": "general",
+            "type": "general" if not include_cliffs else "cliff",
             "programs": uk_programs,
+            "cliff_gap": cliff_gap if include_cliffs else None,
+            "cliff_share": cliff_share if include_cliffs else None,
         }
     )

policyengine/simulation.py

@@ -58,6 +58,10 @@ class SimulationOptions(BaseModel):
         "[Analysis title]",
         description="The title of the analysis (for charts). If not provided, a default title will be generated.",
     )
+    include_cliffs: bool | None = Field(
+        False,
+        description="Whether to include tax-benefit cliffs in the simulation analyses. If True, cliffs will be included.",
+    )
 
 
 class Simulation: