Stage 5 Increment 3 shadow residual-capacity fallback

macromodel/agents/households/func/residual_capacity_fallback.py

@@ -0,0 +1,206 @@
+"""Stage 5 Increment 3: shadow residual-capacity fallback.
+
+This helper caps the preferred Stage 5 branch from Increment 2 using a
+provisional consumer-credit DSTI proxy and feasible illiquid-asset
+liquidation capacity. It is still shadow-only: it does not mutate household
+stocks, debt service, or bank-clearing state.
+
+The DSTI proxy intentionally uses the new-loan annuity cap with mortgage
+service only:
+
+``psi * Y - a^m``
+
+where ``psi`` is the calibrated debt-service-to-income ceiling, ``Y`` is
+gross household income, and ``a^m`` is scheduled mortgage service. That
+proxy is a Stage 5 approximation only; settled remodulated consumer-debt
+service belongs to Stage 6.
+"""
+
+from dataclasses import dataclass
+
+import numpy as np
+
+from macromodel.agents.households.func.borrow_vs_sell import (
+    PREFERRED_MARGIN_BORROW,
+    PREFERRED_MARGIN_SELL,
+)
+
+_RESIDUAL_CAPACITY_EPSILON = 1e-12
+
+
+@dataclass(frozen=True)
+class ResidualCapacityFallbackResult:
+    """Per-household Stage 5 residual-capacity diagnostics for one period."""
+
+    dsti_headroom: np.ndarray
+    dsti_maximum_loan_size: np.ndarray
+    dsti_cap_binding: np.ndarray
+    borrow_planned: np.ndarray
+    liquidation_planned: np.ndarray
+    shadow_credit_requested: np.ndarray
+    forced_liquidation_amount: np.ndarray
+    residual_shortfall_after_caps: np.ndarray
+
+
+def _annuity_payment_factor(period_rate: np.ndarray, loan_maturity: np.ndarray) -> np.ndarray:
+    """Return the fixed-payment annuity factor used by the credit market."""
+    period_rate = np.asarray(period_rate, dtype=float)
+    loan_maturity = np.asarray(loan_maturity, dtype=float)
+
+    valid = np.isfinite(period_rate) & np.isfinite(loan_maturity) & (loan_maturity > 0.0)
+    safe_rate = np.where(valid & (period_rate > 0.0), period_rate, 0.0)
+    safe_maturity = np.where(valid, loan_maturity, 1.0)
+
+    zero_rate_factor = 1.0 / safe_maturity
+    with np.errstate(over="ignore", divide="ignore", invalid="ignore"):
+        positive_rate_factor = safe_rate / (1.0 - np.power(1.0 + safe_rate, -safe_maturity))
+    factor = np.where(safe_rate > 0.0, positive_rate_factor, zero_rate_factor)
+    return np.where(valid & np.isfinite(factor) & (factor > 0.0), factor, 0.0)
+
+
+def compute_residual_capacity_fallback(
+    preferred_margin_after_lfa: np.ndarray,
+    preferred_margin_amount: np.ndarray,
+    income: np.ndarray,
+    scheduled_mortgage_payment: np.ndarray,
+    r_b: np.ndarray,
+    consumer_loan_maturity: np.ndarray | int,
+    dsti_limit: np.ndarray | float,
+    current_ifa: np.ndarray,
+) -> ResidualCapacityFallbackResult:
+    """Apply the shadow DSTI and liquidation caps to the preferred Stage 5 branch.
+
+    Invalid or non-finite inputs collapse to a conservative zero-capacity
+    fallback. Positive residuals are never driven negative by this helper.
+    """
+    preferred_margin_after_lfa = np.asarray(preferred_margin_after_lfa, dtype=float)
+    preferred_margin_amount = np.asarray(preferred_margin_amount, dtype=float)
+    income = np.asarray(income, dtype=float)
+    scheduled_mortgage_payment = np.asarray(scheduled_mortgage_payment, dtype=float)
+    r_b = np.asarray(r_b, dtype=float)
+    consumer_loan_maturity = np.asarray(consumer_loan_maturity, dtype=float)
+    dsti_limit = np.asarray(dsti_limit, dtype=float)
+    current_ifa = np.asarray(current_ifa, dtype=float)
+
+    (
+        preferred_margin_after_lfa,
+        preferred_margin_amount,
+        income,
+        scheduled_mortgage_payment,
+        r_b,
+        consumer_loan_maturity,
+        dsti_limit,
+        current_ifa,
+    ) = np.broadcast_arrays(
+        preferred_margin_after_lfa,
+        preferred_margin_amount,
+        income,
+        scheduled_mortgage_payment,
+        r_b,
+        consumer_loan_maturity,
+        dsti_limit,
+        current_ifa,
+    )
+
+    positive_amount = np.isfinite(preferred_margin_amount) & (preferred_margin_amount > 0.0)
+    borrow_branch = positive_amount & (preferred_margin_after_lfa == PREFERRED_MARGIN_BORROW)
+    sell_branch = positive_amount & (preferred_margin_after_lfa == PREFERRED_MARGIN_SELL)
+    valid_margin = borrow_branch | sell_branch
+
+    proxy_valid = (
+        valid_margin
+        & np.isfinite(income)
+        & (income >= 0.0)
+        & np.isfinite(scheduled_mortgage_payment)
+        & (scheduled_mortgage_payment >= 0.0)
+        & np.isfinite(consumer_loan_maturity)
+        & np.isfinite(dsti_limit)
+        & np.isfinite(current_ifa)
+        & (current_ifa >= 0.0)
+        & (consumer_loan_maturity > 0.0)
+        & (dsti_limit > 0.0)
+    )
+
+    dsti_headroom = np.where(
+        proxy_valid,
+        np.maximum(dsti_limit * income - scheduled_mortgage_payment, 0.0),
+        0.0,
+    )
+    safe_rate = np.where(proxy_valid & (r_b > 0.0), r_b, 0.0)
+    annuity_factor = _annuity_payment_factor(safe_rate, consumer_loan_maturity)
+    dsti_maximum_loan_size = np.where(
+        proxy_valid & (annuity_factor > 0.0),
+        dsti_headroom / annuity_factor,
+        0.0,
+    )
+    dsti_maximum_loan_size = np.where(
+        np.isfinite(dsti_maximum_loan_size) & (dsti_maximum_loan_size > 0.0),
+        dsti_maximum_loan_size,
+        0.0,
+    )
+
+    liquidation_capacity = np.where(
+        proxy_valid & np.isfinite(current_ifa) & (current_ifa > 0.0),
+        current_ifa,
+        0.0,
+    )
+
+    borrow_planned = np.zeros_like(preferred_margin_amount, dtype=float)
+    liquidation_planned = np.zeros_like(preferred_margin_amount, dtype=float)
+    forced_liquidation_amount = np.zeros_like(preferred_margin_amount, dtype=float)
+
+    borrow_request = np.where(borrow_branch, preferred_margin_amount, 0.0)
+    borrow_planned = np.where(
+        borrow_branch,
+        np.minimum(borrow_request, dsti_maximum_loan_size),
+        borrow_planned,
+    )
+    borrow_shortfall = np.maximum(borrow_request - borrow_planned, 0.0)
+
+    liquidation_planned = np.where(
+        borrow_branch,
+        np.minimum(borrow_shortfall, liquidation_capacity),
+        liquidation_planned,
+    )
+    forced_liquidation_amount = np.where(borrow_branch, liquidation_planned, forced_liquidation_amount)
+
+    sell_request = np.where(sell_branch, preferred_margin_amount, 0.0)
+    sell_liquidation = np.where(sell_branch, np.minimum(sell_request, liquidation_capacity), 0.0)
+    liquidation_planned = np.where(sell_branch, sell_liquidation, liquidation_planned)
+    sell_shortfall = np.maximum(sell_request - sell_liquidation, 0.0)
+    sell_borrow = np.where(sell_branch, np.minimum(sell_shortfall, dsti_maximum_loan_size), 0.0)
+    borrow_planned = np.where(sell_branch, sell_borrow, borrow_planned)
+
+    shadow_credit_requested = borrow_planned.copy()
+
+    branch_residual = np.where(
+        borrow_branch,
+        preferred_margin_amount - borrow_planned - liquidation_planned,
+        np.where(sell_branch, preferred_margin_amount - liquidation_planned - borrow_planned, preferred_margin_amount),
+    )
+    residual_shortfall_after_caps = np.where(
+        positive_amount & np.isfinite(branch_residual),
+        np.maximum(branch_residual, 0.0),
+        0.0,
+    )
+
+    dsti_cap_binding = np.where(
+        borrow_branch,
+        borrow_planned < (borrow_request - _RESIDUAL_CAPACITY_EPSILON),
+        np.where(
+            sell_branch,
+            sell_borrow < (sell_shortfall - _RESIDUAL_CAPACITY_EPSILON),
+            False,
+        ),
+    )
+
+    return ResidualCapacityFallbackResult(
+        dsti_headroom=dsti_headroom,
+        dsti_maximum_loan_size=dsti_maximum_loan_size,
+        dsti_cap_binding=dsti_cap_binding,
+        borrow_planned=borrow_planned,
+        liquidation_planned=liquidation_planned,
+        shadow_credit_requested=shadow_credit_requested,
+        forced_liquidation_amount=forced_liquidation_amount,
+        residual_shortfall_after_caps=residual_shortfall_after_caps,
+    )

macromodel/agents/households/households.py

@@ -43,6 +43,10 @@
 from macromodel.agents.households.func.portfolio_target_share import (
     compute_household_head_covariates,
 )
+from macromodel.agents.households.func.residual_capacity_fallback import (
+    ResidualCapacityFallbackResult,
+    compute_residual_capacity_fallback,
+)
 from macromodel.agents.households.household_properties import HouseholdType
 from macromodel.agents.households.households_ts import create_households_timeseries
 from macromodel.agents.households.income_belief_learning import (
@@ -118,6 +122,14 @@
     "borrow_vs_sell_spread": 0.0,
     "borrow_vs_sell_l_tilde": 0.0,
     "borrow_vs_sell_comparison_valid_flag": False,
+    "dsti_headroom": 0.0,
+    "dsti_maximum_loan_size": 0.0,
+    "dsti_cap_binding": False,
+    "borrow_planned": 0.0,
+    "liquidation_planned": 0.0,
+    "shadow_credit_requested": 0.0,
+    "forced_liquidation_amount": 0.0,
+    "residual_shortfall_after_caps": 0.0,
 }
 
 
@@ -1129,6 +1141,71 @@ def compute_and_record_borrow_vs_sell_choice(
             self.ts.borrow_vs_sell_comparison_valid_flag.append(result.comparison_valid_flag)
         return result.preferred_margin, result.preferred_amount
 
+    def compute_and_record_residual_capacity_fallback(
+        self,
+        preferred_margin_after_lfa: np.ndarray,
+        preferred_margin_amount: np.ndarray,
+        banks: Banks,
+        income: np.ndarray,
+        scheduled_mortgage_payment: np.ndarray,
+        consumer_loan_maturity: int,
+        dsti_limit: float,
+        current_ifa: np.ndarray,
+        replace_current: bool = False,
+    ) -> ResidualCapacityFallbackResult:
+        """Compute and persist the Stage 5 Increment 3 shadow residual-capacity fallback.
+
+        This uses the provisional DSTI proxy only. It records the shadow plan
+        but does not touch live balances, debt service, or market-clearing
+        state.
+        """
+        bank_rates = np.asarray(banks.ts.current("interest_rates_on_household_consumption_loans"), dtype=float)
+        corresponding_bank_ids = np.asarray(self.states["Corresponding Bank ID"], dtype=int)
+        if bank_rates.ndim == 0:
+            bank_rates = np.full(self.ts.current("n_households"), float(bank_rates), dtype=float)
+        elif bank_rates.ndim == 1:
+            if bank_rates.shape[0] == self.ts.current("n_households"):
+                pass
+            elif bank_rates.shape[0] == banks.ts.current("n_banks"):
+                bank_rates = bank_rates[corresponding_bank_ids]
+            else:
+                bank_rates = np.resize(bank_rates, banks.ts.current("n_banks"))[corresponding_bank_ids]
+        else:
+            raise ValueError(
+                "interest_rates_on_household_consumption_loans must be a scalar, one rate per household, "
+                "or one rate per bank."
+            )
+
+        result = compute_residual_capacity_fallback(
+            preferred_margin_after_lfa=np.asarray(preferred_margin_after_lfa, dtype=float),
+            preferred_margin_amount=np.asarray(preferred_margin_amount, dtype=float),
+            income=np.asarray(income, dtype=float),
+            scheduled_mortgage_payment=np.asarray(scheduled_mortgage_payment, dtype=float),
+            r_b=bank_rates,
+            consumer_loan_maturity=consumer_loan_maturity,
+            dsti_limit=dsti_limit,
+            current_ifa=np.asarray(current_ifa, dtype=float),
+        )
+        if replace_current:
+            self.ts.override_current("dsti_headroom", result.dsti_headroom)
+            self.ts.override_current("dsti_maximum_loan_size", result.dsti_maximum_loan_size)
+            self.ts.override_current("dsti_cap_binding", result.dsti_cap_binding)
+            self.ts.override_current("borrow_planned", result.borrow_planned)
+            self.ts.override_current("liquidation_planned", result.liquidation_planned)
+            self.ts.override_current("shadow_credit_requested", result.shadow_credit_requested)
+            self.ts.override_current("forced_liquidation_amount", result.forced_liquidation_amount)
+            self.ts.override_current("residual_shortfall_after_caps", result.residual_shortfall_after_caps)
+        else:
+            self.ts.dsti_headroom.append(result.dsti_headroom)
+            self.ts.dsti_maximum_loan_size.append(result.dsti_maximum_loan_size)
+            self.ts.dsti_cap_binding.append(result.dsti_cap_binding)
+            self.ts.borrow_planned.append(result.borrow_planned)
+            self.ts.liquidation_planned.append(result.liquidation_planned)
+            self.ts.shadow_credit_requested.append(result.shadow_credit_requested)
+            self.ts.forced_liquidation_amount.append(result.forced_liquidation_amount)
+            self.ts.residual_shortfall_after_caps.append(result.residual_shortfall_after_caps)
+        return result
+
     def current_stage4_handoff_for_stage5(
         self,
         *,

macromodel/configurations/country_configuration.py

@@ -1,6 +1,6 @@
-from typing import Optional
+from typing import Any, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from .bank_configuration import BanksConfiguration
 from .central_bank_configuration import CentralBankConfiguration
@@ -35,6 +35,7 @@ class CountryConfiguration(BaseModel):
     housing_market: HousingMarketConfiguration = HousingMarketConfiguration()
     credit_market: CreditMarketConfiguration = CreditMarketConfiguration()
     goods_market: GoodsMarketConfiguration = GoodsMarketConfiguration()
+    consumer_credit: dict[str, Any] = Field(default_factory=dict)
 
     forecasting_window: int = 60
     assume_zero_growth: bool = False