# imports
import numpy as np
from taxcalc import Policy, Records, Calculator, GrowFactors
from ccc.utils import DEFAULT_START_YEAR, TC_LAST_YEAR, RECORDS_START_YEAR
[docs]
def get_calculator(
calculator_start_year,
baseline_policy=None,
reform=None,
data="cps",
gfactors=None,
weights=None,
records_start_year=RECORDS_START_YEAR,
):
"""
This function creates the tax calculator object for the microsim
model.
Note: gfactors and weights are only used if provide custom data
path or file with those gfactors and weights. Otherwise, the
model defaults to those gfactors and weights from Tax-Calculator.
Args:
calculator_start_year (integer): first year of budget window
baseline_policy (dictionary): IIT baseline parameters
reform (dictionary): IIT reform parameters
data (string or Pandas DataFrame): path to file or DataFrame
for Tax-Calculator Records object (optional)
gfactors (str or DataFrame): grow factors to extrapolate data
weights (DataFrame): weights DataFrame for Tax-Calculator
Records object (optional)
records_start_year (integer): the start year for the data and
weights dfs
Returns:
calc1 (Tax Calculator Calculator object): TC Calculator object
with a current_year equal to calculator_start_year
"""
# create a calculator
policy1 = Policy()
if data is not None and "cps" in data:
print("Using CPS")
records1 = Records.cps_constructor()
# impute short and long term capital gains if using CPS data
# in 2012 SOI data 6.587% of CG as short-term gains
records1.p22250 = 0.06587 * records1.e01100
records1.p23250 = (1 - 0.06587) * records1.e01100
# set total capital gains to zero
records1.e01100 = np.zeros(records1.e01100.shape[0])
elif data is None or "puf" in data: # pragma: no cover
print("Using PUF")
records1 = Records()
elif data is not None and "tmd" in data: # pragma: no cover
print("Using TMD")
records1 = Records.tmd_constructor("tmd.csv.gz")
elif data is not None: # pragma: no cover
print("Data is ", data)
print("Weights are ", weights)
print("Records start year is ", records_start_year)
records1 = Records(
data=data,
gfactors=gfactors,
weights=weights,
start_year=records_start_year,
) # pragma: no cover
else: # pragma: no cover
raise ValueError("Please provide data or use CPS, PUF, or TMD.")
if baseline_policy: # if something other than current law policy baseline
update_policy(policy1, baseline_policy)
if reform: # if there is a reform
update_policy(policy1, reform)
# the default set up increments year to 2013
calc1 = Calculator(records=records1, policy=policy1)
print("Calculator initial year = ", calc1.current_year)
# this increment_year function extrapolates all PUF variables to
# the next year so this step takes the calculator to the start_year
if calculator_start_year > TC_LAST_YEAR:
raise RuntimeError("Start year is beyond data extrapolation.")
while calc1.current_year < calculator_start_year:
calc1.increment_year()
return calc1
[docs]
def get_rates(
start_year=DEFAULT_START_YEAR,
baseline_policy=None,
reform={},
data="cps",
gfactors=None,
weights=None,
records_start_year=RECORDS_START_YEAR,
):
"""
This function computes weighted average marginal tax rates using
micro data from the tax calculator
Args:
start_year (integer): start year for the simulations
baseline_policy (dict): baseline parameters
reform (dict): reform parameters
data (string or Pandas DataFrame): path to file or DataFrame
for Tax-Calculator Records object (optional)
gfactors (Tax-Calculator GrowFactors object): grow factors
weights (str): path to weights file for Tax-Calculator
Records object
records_start_year (integer): the start year for the microdata
Returns:
individual_rates (dict): individual income (IIT+payroll)
marginal tax rates
"""
calc1 = get_calculator(
calculator_start_year=start_year,
baseline_policy=baseline_policy,
reform=reform,
data=data,
gfactors=None,
weights=None,
records_start_year=records_start_year,
)
# running all the functions and calculates taxes
calc1.calc_all()
# Loop over years in window of calculations
end_year = start_year
array_size = end_year - start_year + 1
rates_dict = {
"tau_div": "e00650",
"tau_int": "e00300",
"tau_scg": "p22250",
"tau_lcg": "p23250",
}
individual_rates = {
"tau_pt": np.zeros(array_size),
"tau_div": np.zeros(array_size),
"tau_int": np.zeros(array_size),
"tau_scg": np.zeros(array_size),
"tau_lcg": np.zeros(array_size),
"tau_td": np.zeros(array_size),
"tau_h": np.zeros(array_size),
}
for year in range(start_year, end_year + 1):
print("Calculator year = ", calc1.current_year)
calc1.advance_to_year(year)
print("year: ", str(calc1.current_year))
# Compute mtrs
# Sch C
[mtr_fica_schC, mtr_iit_schC, mtr_combined_schC] = calc1.mtr("e00900p")
# Sch E - includes partnership and s corp income
[mtr_fica_schE, mtr_iit_schE, mtr_combined_schE] = calc1.mtr("e02000")
# Partnership and s corp income
[mtr_fica_PT, mtr_iit_PT, mtr_combined_PT] = calc1.mtr("e26270")
# pension distributions
# does PUF have e01500? Do we want IRA distributions here?
# Weird - I see e01500 in PUF, but error when try to call it
[mtr_fica_pension, mtr_iit_pension, mtr_combined_pension] = calc1.mtr(
"e01700"
)
# mortgage interest and property tax deductions
# do we also want mtg ins premiums here?
# mtg interest
[mtr_fica_mtg, mtr_iit_mtg, mtr_combined_mtg] = calc1.mtr("e19200")
# prop tax
[mtr_fica_prop, mtr_iit_prop, mtr_combined_prop] = calc1.mtr("e18500")
pos_ti = calc1.array("c04800") > 0
individual_rates["tau_pt"][year - start_year] = (
(
(mtr_iit_schC * np.abs(calc1.array("e00900p")))
+ (
mtr_iit_schE
* np.abs(calc1.array("e02000") - calc1.array("e26270"))
)
+ (mtr_iit_PT * np.abs(calc1.array("e26270")))
)
* pos_ti
* calc1.array("s006")
).sum() / (
(
np.abs(calc1.array("e00900p"))
+ np.abs(calc1.array("e02000") - calc1.array("e26270"))
+ np.abs(calc1.array("e26270"))
)
* pos_ti
* calc1.array("s006")
).sum()
individual_rates["tau_td"][year - start_year] = (
mtr_iit_pension
* calc1.array("e01500")
* pos_ti
* calc1.array("s006")
).sum() / (calc1.array("e01500") * pos_ti * calc1.array("s006")).sum()
individual_rates["tau_h"][year - start_year] = -1 * (
(
(mtr_iit_mtg * calc1.array("e19200"))
+ (mtr_iit_prop * calc1.array("e18500"))
* pos_ti
* calc1.array("s006")
).sum()
/ (
(calc1.array("e19200"))
+ (calc1.array("e18500")) * pos_ti * calc1.array("s006")
).sum()
)
# Loop over MTRs that have only one income source
for k, v in rates_dict.items():
[mtr_fica, mtr_iit, mtr_combined] = calc1.mtr(v)
individual_rates[k][year - start_year] = (
mtr_iit * calc1.array(v) * pos_ti * calc1.array("s006")
).sum() / (calc1.array(v) * pos_ti * calc1.array("s006")).sum()
print(individual_rates)
return individual_rates
def update_policy(policy_obj, reform, **kwargs):
"""
Convenience method that updates the Policy object with the reform
dict using the appropriate method, given the reform format.
"""
if is_paramtools_format(reform):
policy_obj.adjust(reform, **kwargs)
else:
policy_obj.implement_reform(reform, **kwargs)
def is_paramtools_format(reform):
"""
Check first item in reform to determine if it is using the ParamTools
adjustment or the Tax-Calculator reform format.
If first item is a dict, then it is likely be a Tax-Calculator reform:
{
param: {2020: 1000}
}
Otherwise, it is likely to be a ParamTools format.
Returns:
format (bool): True if reform is likely to be in PT format.
"""
for _, data in reform.items():
if isinstance(data, dict):
return False # taxcalc reform
else:
# Not doing a specific check to see if the value is a list
# since it could be a list or just a scalar value.
return True
