Add markup-based bank interest-rate pricing

macro_data/processing/synthetic_banks/default_synthetic_banks.py

@@ -98,6 +98,10 @@ def __init__(
         hh_mortgage_passthrough: float,
         hh_mortgage_ect: float,
         hh_mortgage_rate: float,
+        firm_short_spread: float,
+        firm_long_spread: float,
+        household_consumption_spread: float,
+        mortgage_spread: float,
         proxy_country: Optional[Country] = None,
     ):
         """Initialize the banking system data container.
@@ -117,6 +121,10 @@ def __init__(
             hh_mortgage_passthrough (float): Estimated mortgage rate parameter
             hh_mortgage_ect (float): Estimated mortgage ECT parameter
             hh_mortgage_rate (float): Initial mortgage rate
+            firm_short_spread (float): Mean pre-start short-term firm loan spread
+            firm_long_spread (float): Mean pre-start long-term firm loan spread
+            household_consumption_spread (float): Mean pre-start household consumption spread
+            mortgage_spread (float): Mean pre-start mortgage spread
             proxy_country (Optional[Country]): Country to use as proxy when missing data
         """
         super().__init__(
@@ -134,6 +142,10 @@ def __init__(
             hh_mortgage_passthrough=hh_mortgage_passthrough,
             hh_mortgage_ect=hh_mortgage_ect,
             hh_mortgage_rate=hh_mortgage_rate,
+            firm_short_spread=firm_short_spread,
+            firm_long_spread=firm_long_spread,
+            household_consumption_spread=household_consumption_spread,
+            mortgage_spread=mortgage_spread,
         )
 
     @classmethod
@@ -220,7 +232,11 @@ def from_readers(
             hh_mortgage_ect,
             hh_mortgage_passthrough,
             hh_mortgage_rate,
-        ) = cls.initialise_rates(country_name, inflation_data, proxy_eu_country, quarter, readers, year)
+            firm_short_spread,
+            firm_long_spread,
+            household_consumption_spread,
+            mortgage_spread,
+        ) = cls.initialise_rates(country_name, inflation_data, proxy_eu_country, quarter, readers, year, time_unit)
 
         initial_central_bank_policy_rate = (
             readers.policy_rates.get_policy_rates(country_name).loc[f"{year}-Q{quarter}", "Policy Rate"].values[0]
@@ -243,6 +259,10 @@ def from_readers(
             hh_mortgage_passthrough=hh_mortgage_passthrough,
             hh_mortgage_ect=hh_mortgage_ect,
             hh_mortgage_rate=hh_mortgage_rate,
+            firm_short_spread=firm_short_spread,
+            firm_long_spread=firm_long_spread,
+            household_consumption_spread=household_consumption_spread,
+            mortgage_spread=mortgage_spread,
             quarter=quarter,
             proxy_country=proxy_eu_country,
         )
@@ -331,7 +351,11 @@ def from_readers_compustat(
             hh_mortgage_ect,
             hh_mortgage_passthrough,
             hh_mortgage_rate,
-        ) = cls.initialise_rates(country_name, inflation_data, proxy_eu_country, quarter, readers, year)
+            firm_short_spread,
+            firm_long_spread,
+            household_consumption_spread,
+            mortgage_spread,
+        ) = cls.initialise_rates(country_name, inflation_data, proxy_eu_country, quarter, readers, year, time_unit)
 
         initial_central_bank_policy_rate = (
             readers.policy_rates.get_policy_rates(country_name).loc[f"{year}-Q{quarter}", "Policy Rate"].values[0]
@@ -354,11 +378,83 @@ def from_readers_compustat(
             hh_mortgage_passthrough=hh_mortgage_passthrough,
             hh_mortgage_ect=hh_mortgage_ect,
             hh_mortgage_rate=hh_mortgage_rate,
+            firm_short_spread=firm_short_spread,
+            firm_long_spread=firm_long_spread,
+            household_consumption_spread=household_consumption_spread,
+            mortgage_spread=mortgage_spread,
             quarter=quarter,
         )
 
     @classmethod
-    def initialise_rates(cls, country_name, inflation_data, proxy_eu_country, quarter, readers, year):
+    @staticmethod
+    def _periods_per_year(time_unit: int) -> int:
+        """Return the number of model periods in one calendar year."""
+        if time_unit <= 0 or 12 % time_unit != 0:
+            raise ValueError("Bank-rate preprocessing requires `time_unit` to be a positive divisor of 12.")
+        return 12 // time_unit
+
+    @classmethod
+    def _convert_annual_rate_to_period(cls, annual_rate: float, time_unit: int) -> float:
+        """Convert an annual quoted rate to the model's per-period convention."""
+        return annual_rate / cls._periods_per_year(time_unit)
+
+    @classmethod
+    def _convert_annual_rate_series_to_period(
+        cls, annual_rates: Optional[pd.Series], time_unit: int
+    ) -> Optional[pd.Series]:
+        """Convert a quoted annual rate series to per-period units."""
+        if annual_rates is None:
+            return None
+        return annual_rates.astype(float) / cls._periods_per_year(time_unit)
+
+    @classmethod
+    def _convert_annual_rate_frame_to_period(
+        cls,
+        annual_rates: pd.DataFrame,
+        time_unit: int,
+        column: str,
+    ) -> pd.DataFrame:
+        """Convert a quoted annual-rate column in a DataFrame to per-period units."""
+        period_rates = annual_rates.copy()
+        period_rates[column] = period_rates[column].astype(float) / cls._periods_per_year(time_unit)
+        return period_rates
+
+    @staticmethod
+    def _initial_policy_rate_quarter_key(year: int, quarter: int) -> str:
+        """Return the pre-start quarter key used for initial policy-rate anchoring."""
+        start_period = pd.Period(year=year, quarter=quarter, freq="Q")
+        initial_period = start_period - 1
+        return f"{initial_period.year}-Q{initial_period.quarter}"
+
+    @staticmethod
+    def _pre_start_mask(index: pd.Index, year: int, quarter: int) -> pd.Series:
+        """Return the mask for quarterly observations strictly before the start quarter."""
+        return index < pd.Timestamp(f"{year}Q{quarter}")
+
+    @classmethod
+    def _mean_pre_start_spread(
+        cls,
+        product_rates: Optional[pd.Series],
+        policy_rates: pd.DataFrame,
+        year: int,
+        quarter: int,
+    ) -> float:
+        """Compute the mean quarterly pre-start spread between a product rate and policy."""
+        if product_rates is None:
+            return 0.0
+
+        df = pd.DataFrame({"product_rate": product_rates}).sort_index()
+        df = pd.merge_asof(df, policy_rates[["Policy Rate"]].sort_index(), left_index=True, right_index=True)
+        df["Policy Rate"] = df["Policy Rate"].ffill()
+        df = df.loc[cls._pre_start_mask(df.index, year, quarter)].dropna(subset=["product_rate", "Policy Rate"])
+
+        if df.empty:
+            return 0.0
+
+        return float((df["product_rate"] - df["Policy Rate"]).mean())
+
+    @classmethod
+    def initialise_rates(cls, country_name, inflation_data, proxy_eu_country, quarter, readers, year, time_unit):
         """Preprocess and estimate initial interest rate parameters.
 
         This method:
@@ -381,7 +477,7 @@ def initialise_rates(cls, country_name, inflation_data, proxy_eu_country, quarte
             year (int): Reference year
 
         Returns:
-            tuple: Nine estimated parameters:
+            tuple: Thirteen estimated parameters:
                 - firm_ect: Estimated error correction for firm rates
                 - firm_passthrough: Estimated adjustment for firm rates
                 - firm_rate: Initial firm rate
@@ -391,17 +487,38 @@ def initialise_rates(cls, country_name, inflation_data, proxy_eu_country, quarte
                 - hh_mortgage_ect: Estimated mortgage ECT
                 - hh_mortgage_passthrough: Estimated mortgage adjustment
                 - hh_mortgage_rate: Initial mortgage rate
+                - firm_short_spread: Mean pre-start short-term firm spread
+                - firm_long_spread: Mean pre-start long-term firm spread
+                - household_consumption_spread: Mean pre-start household consumption spread
+                - mortgage_spread: Mean pre-start mortgage spread
         """
         if country_name.is_eu_country:
             data_country = country_name
         else:
             if proxy_eu_country is None:
                 raise ValueError("Proxy EU country is required for non-EU countries.")
             data_country = proxy_eu_country
-        firm_rate = readers.ecb_reader.get_firm_rates(data_country)
-        household_consumption_rate = readers.ecb_reader.get_household_consumption_rates(data_country)
-        household_mortgage_rates = readers.ecb_reader.get_household_mortgage_rates(data_country)
-        policy_rates = readers.policy_rates.get_policy_rates(data_country)
+        firm_rate = cls._convert_annual_rate_series_to_period(
+            readers.ecb_reader.get_firm_rates(data_country), time_unit
+        )
+        household_consumption_rate = cls._convert_annual_rate_series_to_period(
+            readers.ecb_reader.get_household_consumption_rates(data_country),
+            time_unit,
+        )
+        household_mortgage_rates = cls._convert_annual_rate_series_to_period(
+            readers.ecb_reader.get_household_mortgage_rates(data_country),
+            time_unit,
+        )
+        policy_rates = cls._convert_annual_rate_frame_to_period(
+            readers.policy_rates.get_policy_rates(data_country),
+            time_unit,
+            "Policy Rate",
+        )
+        firm_spread = cls._mean_pre_start_spread(firm_rate, policy_rates, year, quarter)
+        household_consumption_spread = cls._mean_pre_start_spread(
+            household_consumption_rate, policy_rates, year, quarter
+        )
+        mortgage_spread = cls._mean_pre_start_spread(household_mortgage_rates, policy_rates, year, quarter)
         npl_rates = readers.world_bank.get_npl_ratios(data_country)
         if any(
             [
@@ -447,6 +564,10 @@ def initialise_rates(cls, country_name, inflation_data, proxy_eu_country, quarte
             household_mortgages_ect,
             hh_mortgage_passthrough,
             household_mortgages_rate,
+            firm_spread,
+            firm_spread,
+            household_consumption_spread,
+            mortgage_spread,
         )
 
     def set_bank_equity(self, bank_equity: float) -> None:

macro_data/processing/synthetic_banks/synthetic_banks.py

@@ -87,6 +87,10 @@ class SyntheticBanks(ABC):
         hh_mortgage_passthrough (float): Estimated mortgage rate parameter
         hh_mortgage_ect (float): Estimated mortgage ECT parameter
         hh_mortgage_rate (float): Initial mortgage rate
+        firm_short_spread (float): Mean pre-start short-term firm loan spread over policy
+        firm_long_spread (float): Mean pre-start long-term firm loan spread over policy
+        household_consumption_spread (float): Mean pre-start household consumption loan spread
+        mortgage_spread (float): Mean pre-start mortgage spread over policy
     """
 
     @abstractmethod
@@ -106,6 +110,10 @@ def __init__(
         hh_mortgage_passthrough: float,
         hh_mortgage_ect: float,
         hh_mortgage_rate: float,
+        firm_short_spread: float,
+        firm_long_spread: float,
+        household_consumption_spread: float,
+        mortgage_spread: float,
     ) -> None:
         """Initialize a synthetic banking system.
 
@@ -124,6 +132,10 @@ def __init__(
             hh_mortgage_passthrough (float): Rate adjustment for mortgages
             hh_mortgage_ect (float): Error correction for mortgage rates
             hh_mortgage_rate (float): Base mortgage rate
+            firm_short_spread (float): Mean pre-start short-term firm loan spread
+            firm_long_spread (float): Mean pre-start long-term firm loan spread
+            household_consumption_spread (float): Mean pre-start household consumption spread
+            mortgage_spread (float): Mean pre-start mortgage spread
         """
         # Parameters
         self.country_name = country_name
@@ -147,6 +159,11 @@ def __init__(
         self.hh_mortgage_ect = hh_mortgage_ect
         self.hh_mortgage_rate = hh_mortgage_rate
 
+        self.firm_short_spread = firm_short_spread
+        self.firm_long_spread = firm_long_spread
+        self.household_consumption_spread = household_consumption_spread
+        self.mortgage_spread = mortgage_spread
+
     def initialise_deposits_and_loans(
         self, synthetic_population: SyntheticPopulation, firm_deposits: np.ndarray, firm_debt: np.ndarray
     ) -> None:
@@ -293,20 +310,8 @@ def initialise_rates_profits_liabilities(
             mortgage_markup (float): Markup for mortgages
             household_overdraft_markup (float): Markup for household overdrafts
         """
-        bank_markup_interest_rate_short_term_firm_loans = risk_premium
-        bank_markup_interest_rate_long_term_firm_loans = risk_premium
-        bank_markup_interest_rate_household_payday_loans = risk_premium
-        bank_markup_interest_rate_overdraft_firm = risk_premium
-
         self.set_initial_interest_rates(
             central_bank_policy_rate=policy_rate,
-            bank_markup_interest_rate_short_term_firm_loans=bank_markup_interest_rate_short_term_firm_loans,
-            bank_markup_interest_rate_long_term_firm_loans=bank_markup_interest_rate_long_term_firm_loans,
-            bank_markup_interest_rate_household_payday_loans=bank_markup_interest_rate_household_payday_loans,
-            bank_markup_interest_rate_household_consumption_loans=consumption_loans_markup,
-            bank_markup_interest_rate_mortgages=mortgage_markup,
-            bank_markup_interest_rate_overdraft_firm=bank_markup_interest_rate_overdraft_firm,
-            bank_markup_interest_rate_overdraft_household=household_overdraft_markup,
         )
 
         self.set_interest_received_from_loans()
@@ -325,13 +330,6 @@ def initialise_rates_profits_liabilities(
     def set_initial_interest_rates(
         self,
         central_bank_policy_rate: float,
-        bank_markup_interest_rate_short_term_firm_loans: float,
-        bank_markup_interest_rate_long_term_firm_loans: float,
-        bank_markup_interest_rate_household_payday_loans: float,
-        bank_markup_interest_rate_household_consumption_loans: float,
-        bank_markup_interest_rate_mortgages: float,
-        bank_markup_interest_rate_overdraft_firm: float,
-        bank_markup_interest_rate_overdraft_household: float,
     ) -> None:
         """Set initial interest rates for all bank products.
 
@@ -343,13 +341,6 @@ def set_initial_interest_rates(
 
         Args:
             central_bank_policy_rate (float): Base rate from central bank
-            bank_markup_interest_rate_short_term_firm_loans (float): Markup for short-term firm loans
-            bank_markup_interest_rate_long_term_firm_loans (float): Markup for long-term firm loans
-            bank_markup_interest_rate_household_payday_loans (float): Markup for payday loans
-            bank_markup_interest_rate_household_consumption_loans (float): Markup for consumer loans
-            bank_markup_interest_rate_mortgages (float): Markup for mortgages
-            bank_markup_interest_rate_overdraft_firm (float): Markup for firm overdrafts
-            bank_markup_interest_rate_overdraft_household (float): Markup for household overdrafts
         """
         # Short-term interest rates for firm loans
         self.bank_data["Short-Term Interest Rates on Firm Loans"] = self.firm_rate

macromodel/agents/banks/banks.py

@@ -134,6 +134,20 @@ def from_pickled_agent(
         parameters = configuration.parameters
         functions = functions_from_model(model=configuration.functions, loc="macromodel.agents.banks")
 
+        def get_spread(
+            attribute_names: list[str],
+            column_names: list[str],
+        ) -> float | np.ndarray:
+            for attribute_name in attribute_names:
+                if hasattr(synthetic_banks, attribute_name):
+                    spread = getattr(synthetic_banks, attribute_name)
+                    if spread is not None:
+                        return spread
+            for column_name in column_names:
+                if column_name in synthetic_banks.bank_data.columns:
+                    return synthetic_banks.bank_data[column_name].values
+            return 0.0
+
         data = synthetic_banks.bank_data.drop(columns=["Corresponding Firms ID", "Corresponding Households ID"])
         ts = create_banks_timeseries(
             bank_data=data,
@@ -150,6 +164,22 @@ def from_pickled_agent(
             "Household Consumption ECT": synthetic_banks.hh_consumption_ect,
             "Household Mortgage Pass Through": synthetic_banks.hh_mortgage_passthrough,
             "Household Mortgage ECT": synthetic_banks.hh_mortgage_ect,
+            "Firm Short Spread": get_spread(
+                attribute_names=["firm_short_spread", "short_term_firm_spread", "firm_spread"],
+                column_names=["Firm Short Spread", "Short-Term Firm Spread", "Firm Spread"],
+            ),
+            "Firm Long Spread": get_spread(
+                attribute_names=["firm_long_spread", "long_term_firm_spread", "firm_spread"],
+                column_names=["Firm Long Spread", "Long-Term Firm Spread", "Firm Spread"],
+            ),
+            "Household Consumption Spread": get_spread(
+                attribute_names=["hh_consumption_spread", "household_consumption_spread"],
+                column_names=["Household Consumption Spread"],
+            ),
+            "Mortgage Spread": get_spread(
+                attribute_names=["mortgage_spread", "hh_mortgage_spread", "household_mortgage_spread"],
+                column_names=["Mortgage Spread", "Household Mortgage Spread"],
+            ),
         }
 
         return cls(
@@ -225,6 +255,7 @@ def set_interest_rates(self, central_bank_policy_rate: float) -> None:
                 prev_interest_rates_on_short_term_firm_loans=self.ts.current("interest_rates_on_short_term_firm_loans"),
                 firm_pt=self.states["Firm Pass Through"],
                 firm_ect=self.states["Firm ECT"],
+                firm_short_spread=self.states["Firm Short Spread"],
             )
         )
         self.ts.average_interest_rates_on_short_term_firm_loans.append(
@@ -236,6 +267,7 @@ def set_interest_rates(self, central_bank_policy_rate: float) -> None:
                 prev_interest_rates_on_long_term_firm_loans=self.ts.current("interest_rates_on_long_term_firm_loans"),
                 firm_pt=self.states["Firm Pass Through"],
                 firm_ect=self.states["Firm ECT"],
+                firm_long_spread=self.states["Firm Long Spread"],
             )
         )
         self.ts.average_interest_rates_on_long_term_firm_loans.append(
@@ -249,6 +281,7 @@ def set_interest_rates(self, central_bank_policy_rate: float) -> None:
                 ),
                 hh_cons_pt=self.states["Household Consumption Pass Through"],
                 hh_cons_ect=self.states["Household Consumption ECT"],
+                hh_consumption_spread=self.states["Household Consumption Spread"],
             )
         )
         self.ts.average_interest_rates_on_household_consumption_loans.append(
@@ -260,6 +293,7 @@ def set_interest_rates(self, central_bank_policy_rate: float) -> None:
                 prev_interest_rate_on_mortgages=self.ts.current("interest_rates_on_mortgages"),
                 hh_mortgage_pt=self.states["Household Mortgage Pass Through"],
                 hh_mortgage_ect=self.states["Household Mortgage ECT"],
+                mortgage_spread=self.states["Mortgage Spread"],
             )
         )
         self.ts.average_interest_rates_on_mortgages.append([self.ts.current("interest_rates_on_mortgages").mean()])