Source code for ogcore.parameters

import os
import numpy as np
import scipy.interpolate as si
import paramtools
import ogcore
from ogcore import elliptical_u_est
from ogcore.utils import (
    rate_conversion,
    extrapolate_array,
    extrapolate_nested_list,
)
from ogcore.constants import BASELINE_DIR

CURRENT_PATH = os.path.abspath(os.path.dirname(__file__))


[docs] class Specifications(paramtools.Parameters): """ Inherits ParamTools Parameters abstract base class. """ defaults = os.path.join(CURRENT_PATH, "default_parameters.json") array_first = True def __init__( self, output_base=BASELINE_DIR, baseline_dir=BASELINE_DIR, baseline=False, num_workers=1, ): super().__init__() self.output_base = output_base self.baseline_dir = baseline_dir self.baseline = baseline self.num_workers = num_workers # put OG-Core version in parameters to save for reference self.ogcore_version = ogcore.__version__ # does cheap calculations to find parameter values self.initialize() self.parameter_warnings = "" self.parameter_errors = "" self._ignore_errors = False
[docs] def initialize(self): """ ParametersBase reads JSON file and sets attributes to self Next call self.compute_default_params for further initialization Args: None Returns: None """ self.compute_default_params()
[docs] def compute_default_params(self): """ Does cheap calculations to return parameter values Args: None Returns: None """ # Catch error if baseline_spending=True and baseline=True if self.baseline_spending and self.baseline: err_msg = ( "Parameter baseline_spending=True cannot coincide with " + "baseline=True." ) raise ValueError(err_msg) # reshape lambdas self.lambdas = self.lambdas.reshape(self.lambdas.shape[0], 1) # cast integers as integers self.S = int(self.S) self.T = int(self.T) self.J = len(self.lambdas) # get parameters of elliptical utility function self.b_ellipse, self.upsilon = elliptical_u_est.estimation( self.frisch, self.ltilde ) # determine length of budget window from individual income tax # parameters passed in self.BW = len(self.etr_params) # Find number of economically active periods of life self.E = int( self.starting_age * (self.S / (self.ending_age - self.starting_age)) ) # Find rates in model periods from annualized rates self.beta = 1 / ( rate_conversion( 1 / self.beta_annual - 1, self.starting_age, self.ending_age, self.S, ) + 1 ) self.delta = -1 * rate_conversion( -1 * self.delta_annual, self.starting_age, self.ending_age, self.S ) self.delta_g = -1 * rate_conversion( -1 * self.delta_g_annual, self.starting_age, self.ending_age, self.S, ) self.g_y = rate_conversion( self.g_y_annual, self.starting_age, self.ending_age, self.S ) # set fraction of income taxes from payroll to zero initially # will be updated when read in tax function parameters if self.T + self.S > self.BW: self.frac_tax_payroll = np.append( self.frac_tax_payroll, np.ones(self.T + self.S - self.BW) * self.frac_tax_payroll[-1], ) # Interpolate chi_n and create omega_SS_80 if necessary if self.S < 80 and self.chi_n.shape[-1] == 80: self.age_midp_80 = np.linspace(20.5, 99.5, 80) reshape_chi_n = np.zeros((self.T + self.S, self.S)) for t in range(self.T + self.S): if self.chi_n.ndim == 1: self.chi_n_interp = si.interp1d( self.age_midp_80, np.squeeze(self.chi_n), kind="cubic", ) else: self.chi_n_interp = si.interp1d( self.age_midp_80, np.squeeze(self.chi_n[t, :]), kind="cubic", ) self.newstep = 80.0 / self.S self.age_midp_S = np.linspace( 20 + 0.5 * self.newstep, 100 - 0.5 * self.newstep, self.S ) reshape_chi_n[t, :] = self.chi_n_interp(self.age_midp_S) self.chi_n = reshape_chi_n # Extend parameters that may vary over the time path tp_param_list = [ "alpha_G", "alpha_T", "alpha_I", "alpha_bs_G", "alpha_bs_T", "alpha_bs_I", "world_int_rate_annual", "adjustment_factor_for_cit_receipts", "tau_bq", "tau_payroll", "h_wealth", "m_wealth", "p_wealth", "retirement_age", "replacement_rate_adjust", "zeta_D", "zeta_K", "r_gov_scale", "r_gov_shift", "g_RM", ] for item in tp_param_list: param_in = getattr(self, item) param_out = extrapolate_array( param_in, dims=(self.T + self.S,), item=item ) setattr(self, item, param_out) # Deal with parameters that vary across industry and over time tp_param_list2 = [ "Z", "delta_tau_annual", "cit_rate", "inv_tax_credit", ] for item in tp_param_list2: param_in = getattr(self, item) param_out = extrapolate_array( param_in, dims=(self.T + self.S, self.M), item=item ) setattr(self, item, param_out) # Deal with parameters that vary across consumption good and over time tp_param_list3 = ["tau_c"] for item in tp_param_list3: param_in = getattr(self, item) param_out = extrapolate_array( param_in, dims=(self.T + self.S, self.I), item=item ) setattr(self, item, param_out) # Deal with parameters that vary across J and over time tp_param_list3 = [ "labor_income_tax_noncompliance_rate", "capital_income_tax_noncompliance_rate", ] for item in tp_param_list3: param_in = getattr(self, item) param_out = extrapolate_array( param_in, dims=(self.T + self.S, self.J), item=item ) setattr(self, item, param_out) # Deal with parameters that vary across age and over time tp_param_list4 = [ "rho", ] for item in tp_param_list4: param_in = getattr(self, item) param_out = extrapolate_array( param_in, dims=(self.T + self.S, self.S), item=item ) setattr(self, item, param_out) # Deal with tax parameters that maybe age and time specific tax_params_to_TP = [ "etr_params", "mtrx_params", "mtry_params", ] for item in tax_params_to_TP: tax_to_set_in = getattr(self, item) try: len(tax_to_set_in[0][0]) except TypeError: print( "please give a " + item + " that is a nested lists of" + " lists that is three lists deep" ) assert False tax_to_set_out = extrapolate_nested_list( tax_to_set_in, dims=(self.T, self.S, len(tax_to_set_in[0][0])) ) setattr(self, item, tax_to_set_out) # Try to deal with size of eta and eta_RM. They may vary by S, J, T, # but want to allow user to enter one that varies by only S, S and J, # S and T, or T and S and J. eta_params_to_TP = [ "eta", "eta_RM", ] for item in eta_params_to_TP: param_in = getattr(self, item) param_out = extrapolate_array( param_in, dims=(self.T + self.S, self.S, self.J), item=item ) setattr(self, item, param_out) # extrapolate lifetime ability e matrix over time dimension param_in = getattr(self, "e") param_out = extrapolate_array( param_in, dims=(self.T, self.S, self.J), item="e" ) setattr(self, "e", param_out) # Extrapolate chi_n over T + S param_in = getattr(self, "chi_n") param_out = extrapolate_array( param_in, dims=(self.T + self.S, self.S), item="chi_n" ) setattr(self, "chi_n", param_out) # make sure zeta matrix sums to one (e.g., default off due to rounding) self.zeta = self.zeta / self.zeta.sum() # open economy parameters self.world_int_rate = rate_conversion( self.world_int_rate_annual, self.starting_age, self.ending_age, self.S, ) # set period of retirement self.retire = ( np.round( ((self.retirement_age - self.starting_age) * self.S) / 80.0 ) - 1 ).astype(int) # Calculations for business income taxes # at some point, we will want to make Cost of Capital Calculator # a dependency to compute tau_b # this adjustment factor has as the numerator CIT receipts/GDP # and as the denominator CIT receipts/GDP from the # model with baseline parameterization and no adjustment to the # CIT_rate self.tau_b = ( self.cit_rate * self.c_corp_share_of_assets * self.adjustment_factor_for_cit_receipts.reshape( self.adjustment_factor_for_cit_receipts.shape[0], 1 ) ) self.delta_tau = -1 * rate_conversion( -1 * self.delta_tau_annual, self.starting_age, self.ending_age, self.S, ) # for constant demographics if self.constant_demographics: self.g_n_ss = 0.0 self.g_n = np.zeros(self.T + self.S) surv_rate = np.ones_like(self.rho) - self.rho surv_rate1 = np.ones((self.S,)) # prob start at age S surv_rate1[1:] = np.cumprod(surv_rate[-1, :-1], dtype=float) # number of each age alive at any time omega_SS = np.ones(self.S) * surv_rate1 self.omega_SS = omega_SS / omega_SS.sum() self.imm_rates = np.zeros((self.T + self.S, self.S)) self.omega = np.tile( np.reshape(self.omega_SS, (1, self.S)), (self.T + self.S, 1) ) self.omega_S_preTP = self.omega_SS # Create time series of stationarized UBI transfers self.ubi_nom_array = self.get_ubi_nom_objs()
[docs] def get_ubi_nom_objs(self): """ Generate time series of nominal SxJ UBI household matrix and aggregate UBI expenditure over necessary time periods. Also generate steady-state versions Args: self: OG-Core Specifications class object Returns: ubi_nom_array (array): T+S x S x J array time series of UBI transfers in dollars for each type-j age-s household in every period t """ # Get matrices of number of children 0-17, number of dependents 18-20, # number of adults 21-64, and number of seniors >= 65 from # OG-Core-Calibration package ubi_num_017_mat = 1.1 * np.ones((self.S, self.J)) ubi_num_1864_mat = 0.85 * np.ones((self.S, self.J)) ubi_num_65p_mat = 0.15 * np.ones((self.S, self.J)) # Calculate the UBI transfers to each household type in the first # period t=0 ubi_nom_init = np.tile( np.reshape( np.minimum( ( self.ubi_nom_017 * ubi_num_017_mat + self.ubi_nom_1864 * ubi_num_1864_mat + self.ubi_nom_65p * ubi_num_65p_mat ), self.ubi_nom_max, ), (1, self.S, self.J), ), (self.T + self.S, 1, 1), ) # Calculate steady-state and transition path of stationary individual # household UBI payments and stationary aggregate UBI outlays if self.ubi_growthadj or self.g_y_annual == 0: # If ubi_growthadj=True or if g_y_annual<0, then ubi_arr is # just a copy of the initial UBI matrix for T periods. ubi_nom_array = ubi_nom_init else: # If ubi_growthadj=False, and g_y_annual>=0, then must divide # by e^{g_y t} every period, then set the steady-state matrix # to its value close to zero at t=T ubi_nom_array = np.zeros_like(ubi_nom_init) discount_factor = np.exp( self.g_y * np.linspace(0, self.T, self.T + 1) ) ubi_nom_array[: self.T + 1, :, :] = ubi_nom_init[ : self.T + 1, :, : ] / discount_factor.reshape(discount_factor.shape[0], 1, 1) ubi_nom_array[self.T + 1 :, :, :] = ubi_nom_array[self.T, :, :] return ubi_nom_array
[docs] def update_specifications(self, revision, raise_errors=True): """ Updates parameter specification with values in revision dictionary. Args: revision (dict): dictionary or JSON string with one or more `PARAM: VALUE` pairs raise_errors (boolean): if True (the default), raises ValueError when `parameter_errors` exists; if False, does not raise ValueError when `parameter_errors` exists and leaves error handling to caller of the update_specifications method. Returns: None Raises: ValueError: if raise_errors is True AND `_validate_parameter_names_types` generates errors OR `_validate_parameter_values` generates errors. Notes: Given a reform dictionary, typical usage of the Specifications class is as follows:: >>> specs = Specifications() >>> specs.update_specifications(revision) An example of a multi-parameter specification is as follows:: >>> revision = { frisch: [0.03] } """ if not (isinstance(revision, dict) or isinstance(revision, str)): raise ValueError("ERROR: revision is not a dictionary or string") # Skip over the adjust method if the tax parameters passed in # are functions (e.g., in the case of tax_func_type = mono) tax_update_dict = {} tax_func_params_functions = False try: if revision["tax_func_type"] in ["mono", "mono2D"]: tax_func_params_functions = True except (KeyError, TypeError): pass if self.tax_func_type in ["mono", "mono2D"]: tax_func_params_functions = True if tax_func_params_functions: try: for item in ["etr_params", "mtrx_params", "mtry_params"]: if item in revision.keys(): tax_update_dict[item] = revision[item] del revision[item] except (KeyError, TypeError): pass self.adjust(revision, raise_errors=raise_errors) # put tax values skipped over in the adjust method back in so # they are in the parameters class. if tax_update_dict != {}: for key, value in tax_update_dict.items(): setattr(self, key, value) self.compute_default_params()
[docs] def revision_warnings_errors(spec_revision): """ Generate warnings and errors for OG-Core parameter specifications Args: spec_revision (dict): dictionary suitable for use with the `Specifications.update_specifications method`. Returns: rtn_dict (dict): with endpoint specific warning and error messages """ rtn_dict = {"warnings": "", "errors": ""} spec = Specifications() spec.update_specifications(spec_revision, raise_errors=False) if spec._errors: rtn_dict["errors"] = spec._errors return rtn_dict