"""
Tax-Calculator tax-filing-unit Records class.
"""
# CODING-STYLE CHECKS:
# pycodestyle records.py
# pylint --disable=locally-disabled records.py
import os
from pathlib import Path
import numpy as np
import pandas as pd
from taxcalc.data import Data
from taxcalc.growfactors import GrowFactors
from taxcalc.utils import read_egg_csv
[docs]
class Records(Data):
"""
Records is a subclass of the abstract Data class, and therefore,
inherits its methods (none of which are shown here).
Constructor for the tax-filing-unit Records class.
Parameters
----------
data: string or Pandas DataFrame or None
string describes CSV file in which records data reside;
DataFrame already contains records data;
default value is None.
NOTE: when using custom data, set this argument to a DataFrame.
NOTE: to use your own data for a specific year with Tax-Calculator,
be sure to read the documentation on creating your own data file and
then construct a Records object like this:
mydata = pd.read_csv(<mydata.csv>)
myrec = Records(data=mydata, start_year=<mydata_year>,
gfactors=None, weights=None)
NOTE: data=None is allowed but the returned instance contains only
the data variable information in the specified VARINFO file.
start_year: integer or None
specifies calendar year of the input data;
default value is None.
Note that if specifying your own data (see above NOTE) as being
a custom data set, be sure to explicitly set start_year to the
custom data's calendar year.
gfactors: GrowFactors class instance or None
containing record data growth (or extrapolation) factors.
default value is None.
weights: Pandas DataFrame or None
DataFrame contains data weights;
None creates empty weights DataFrame;
default value is None
NOTE: when using custom weights, set this argument to a DataFrame.
NOTE: see weights_scale documentation below.
adjust_ratios: Pandas DataFrame or None
DataFrame contains transposed/no-index adjustment ratios;
None creates empty adjustment-ratios DataFrame;
default value is None.
NOTE: when using custom ratios, set this argument to a DataFrame.
NOTE: if specifying a DataFrame, set adjust_ratios to my_df defined as:
my_df = pd.read_csv('<my_ratios.csv>', index_col=0).transpose()
exact_calculations: boolean
specifies whether or not exact tax calculations are done without
any smoothing of stair-step provisions in income tax law;
default value is false.
weights_scale: float
specifies the weights scaling factor used to convert contents
of weights file into the s006 variable. PUF and CPS input data
generated in the taxdata repository use a weights_scale of 0.01,
while TMD input data generated in the tax-microdata repository
use a 1.0 weights_scale value.
default value is 0.01.
Raises
------
ValueError:
if data is not the appropriate type.
if taxpayer and spouse variables do not add up to filing-unit total.
if dividends is less than qualified dividends.
if gfactors is not None or a GrowFactors class instance.
if start_year is not an integer.
if files cannot be found.
Returns
-------
class instance: Records
Notes
-----
Use Records.cps_constructor() to get a Records object instantiated
with CPS input data developed in the taxdata repository.
Use Records.puf_constructor() to get a Records object instantiated
with PUF input data developed in the taxdata repository.
Use Records.tmd_constructor() to get a Records object instantiated
with TMD input data developed in the tax-microdata repository.
"""
# suppress pylint warning about constructor having too many arguments:
# pylint: disable=too-many-arguments
# suppress pylint warnings about uppercase variable names:
# pylint: disable=invalid-name
# suppress pylint warnings about too many class instance attributes:
# pylint: disable=too-many-instance-attributes
PUFCSV_YEAR = 2011
CPSCSV_YEAR = 2014
TMDCSV_YEAR = 2021
CODE_PATH = os.path.abspath(os.path.dirname(__file__))
VARINFO_FILE_NAME = 'records_variables.json'
VARINFO_FILE_PATH = CODE_PATH
def __init__(self,
data=None,
start_year=None,
gfactors=None,
weights=None,
adjust_ratios=None,
exact_calculations=False,
weights_scale=0.01):
# pylint: disable=too-many-positional-arguments
# pylint: disable=no-member,too-many-branches
if isinstance(weights, str):
weights = os.path.join(Records.CODE_PATH, weights)
super().__init__(data, start_year, gfactors, weights, weights_scale)
if data is None:
return # because there are no data
# read adjustment ratios
self.ADJ = None
self._read_ratios(adjust_ratios)
# specify exact value based on exact_calculations
self.exact[:] = np.where(exact_calculations is True, 1, 0)
# specify FLPDYR value based on start_year
self.FLPDYR.fill(start_year)
# check for valid MARS values
if not np.all(np.logical_and(np.greater_equal(self.MARS, 1),
np.less_equal(self.MARS, 5))):
raise ValueError('not all MARS values in [1,5] range')
# create variables derived from MARS, which is in MUST_READ_VARS
self.num[:] = np.where(self.MARS == 2, 2, 1)
self.sep[:] = np.where(self.MARS == 3, 2, 1)
# check for valid EIC values
if not np.all(np.logical_and(np.greater_equal(self.EIC, 0),
np.less_equal(self.EIC, 3))):
raise ValueError('not all EIC values in [0,3] range')
# check that three sets of split-earnings variables have valid values
msg = 'expression "{0} == {0}p + {0}s" is not true for every record'
tol = 0.020001 # handles "%.2f" rounding errors
if not np.allclose(self.e00200, (self.e00200p + self.e00200s),
rtol=0.0, atol=tol):
raise ValueError(msg.format('e00200'))
if not np.allclose(self.e00900, (self.e00900p + self.e00900s),
rtol=0.0, atol=tol):
raise ValueError(msg.format('e00900'))
if not np.allclose(self.e02100, (self.e02100p + self.e02100s),
rtol=0.0, atol=tol):
raise ValueError(msg.format('e02100'))
# check that spouse income variables have valid values
nospouse = self.MARS != 2
zeros = np.zeros_like(self.MARS[nospouse])
msg = '{} is not always zero for non-married filing unit'
if not np.allclose(self.e00200s[nospouse], zeros):
raise ValueError(msg.format('e00200s'))
if not np.allclose(self.e00900s[nospouse], zeros):
raise ValueError(msg.format('e00900s'))
if not np.allclose(self.e02100s[nospouse], zeros):
raise ValueError(msg.format('e02100s'))
if not np.allclose(self.k1bx14s[nospouse], zeros):
raise ValueError(msg.format('k1bx14s'))
# check that ordinary dividends are no less than qualified dividends
other_dividends = np.maximum(0., self.e00600 - self.e00650)
if not np.allclose(self.e00600, self.e00650 + other_dividends,
rtol=0.0, atol=tol):
msg = 'expression "e00600 >= e00650" is not true for every record'
raise ValueError(msg)
del other_dividends
# check that total pension income is no less than taxable pension inc
nontaxable_pensions = np.maximum(0., self.e01500 - self.e01700)
if not np.allclose(self.e01500, self.e01700 + nontaxable_pensions,
rtol=0.0, atol=tol):
msg = 'expression "e01500 >= e01700" is not true for every record'
raise ValueError(msg)
del nontaxable_pensions
# check that PT_SSTB_income has valid value
if not np.all(np.logical_and(np.greater_equal(self.PT_SSTB_income, 0),
np.less_equal(self.PT_SSTB_income, 1))):
raise ValueError('not all PT_SSTB_income values are 0 or 1')
[docs]
@staticmethod
def cps_constructor(
data=None,
gfactors=GrowFactors(),
exact_calculations=False
):
"""
Static method returns a Records object instantiated with CPS
input data. This is a convenience method that eliminates the
need to specify all the details of the CPS input data.
"""
if data is None:
data = os.path.join(Records.CODE_PATH, 'cps.csv.gz')
if gfactors is None:
weights = None
else:
weights = os.path.join(Records.CODE_PATH, 'cps_weights.csv.gz')
return Records(
data=data,
start_year=Records.CPSCSV_YEAR,
gfactors=gfactors,
weights=weights,
adjust_ratios=None,
exact_calculations=exact_calculations,
weights_scale=0.01,
)
@staticmethod
def puf_constructor(
data='puf.csv',
gfactors=GrowFactors(),
weights='puf_weights.csv.gz',
ratios='puf_ratios.csv',
exact_calculations=False
): # pragma: no cover
"""
Static method returns a Records object instantiated with PUF
input data. This is a convenience method that eliminates the
need to specify all the details of the PUF input data.
"""
assert isinstance(data, str)
assert isinstance(gfactors, GrowFactors)
assert isinstance(weights, str)
assert isinstance(ratios, str)
return Records(
data=pd.read_csv(data),
start_year=Records.PUFCSV_YEAR,
gfactors=gfactors,
weights=pd.read_csv(weights),
adjust_ratios=pd.read_csv(ratios, index_col=0).transpose(),
exact_calculations=exact_calculations,
weights_scale=0.01,
)
@staticmethod
def tmd_constructor(
data_path: Path,
weights_path: Path,
growfactors: Path | GrowFactors,
exact_calculations=False,
): # pragma: no cover
"""
Static method returns a Records object instantiated with TMD
input data. This is a convenience method that eliminates the
need to specify all the details of the TMD input data.
"""
assert isinstance(data_path, Path)
assert isinstance(weights_path, Path)
if isinstance(growfactors, Path):
growfactors = GrowFactors(growfactors_filename=str(growfactors))
else:
assert isinstance(growfactors, GrowFactors)
return Records(
data=pd.read_csv(data_path),
start_year=Records.TMDCSV_YEAR,
weights=pd.read_csv(weights_path),
gfactors=growfactors,
adjust_ratios=None,
exact_calculations=exact_calculations,
weights_scale=1.0,
)
[docs]
def increment_year(self):
"""
Add one to current year, and also does
extrapolation, reweighting, adjusting for new current year.
"""
super().increment_year()
self.FLPDYR.fill(self.current_year) # pylint: disable=no-member
# apply variable adjustment ratios
self._adjust(self.current_year)
[docs]
@staticmethod
def read_cps_data():
"""
Return data in cps.csv.gz as a Pandas DataFrame.
"""
fname = os.path.join(Records.CODE_PATH, 'cps.csv.gz')
if os.path.isfile(fname):
cpsdf = pd.read_csv(fname)
else: # find file in conda package
cpsdf = read_egg_csv(fname) # pragma: no cover
return cpsdf
# ----- begin private methods of Records class -----
[docs]
def _adjust(self, year):
"""
Adjust value of PUF income variables to match SOI distributions
Note: adjustment must leave variables as numpy.ndarray type
"""
# pylint: disable=no-member
if self.ADJ.size > 0: # pragma: no cover
# Interest income
self.e00300 *= self.ADJ[f'INT{year}'].iloc[self.agi_bin].values
[docs]
def _read_ratios(self, ratios):
"""
Read Records PUF-related adjustment ratios using
specified transposed/no-index DataFrame as ratios or
create empty DataFrame if ratios is None.
"""
assert ratios is None or isinstance(ratios, pd.DataFrame)
if ratios is None:
setattr(self, 'ADJ', pd.DataFrame({'nothing': []}))
return
if isinstance(ratios, pd.DataFrame): # pragma: no cover
assert 'INT2013' in ratios.columns # check for transposed
assert ratios.index.name is None # check for no-index
ADJ = ratios
self.ADJ = pd.DataFrame()
if ADJ.index.name != 'agi_bin':
ADJ.index.name = 'agi_bin'
setattr(self, 'ADJ', ADJ.astype(np.float32))
