Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Returns
-------
model : IV3SLS
Model instance
Notes
-----
At least one of exog or endog must be provided.
Utility function to simplify the construction of multivariate IV
models which all use the same regressors and instruments. Constructs
the dictionary of equations from the variables using the common
exogenous, endogenous and instrumental variables.
"""
equations = OrderedDict()
dependent = IVData(dependent, var_name='dependent')
if exog is None and endog is None:
raise ValueError('At least one of exog or endog must be provided')
exog = IVData(exog, var_name='exog')
endog = IVData(endog, var_name='endog', nobs=dependent.shape[0])
instr = IVData(instruments, var_name='instruments', nobs=dependent.shape[0])
for col in dependent.pandas:
equations[col] = (dependent.pandas[[col]], exog.pandas, endog.pandas, instr.pandas)
return cls(equations)
def _check_weights(self):
if self._weights is None:
nobs = self._dependent.shape[0]
self._is_weighted = False
self._weight_data = IVData(ones(nobs), 'weights')
else:
self._is_weighted = True
weights = IVData(self._weights).ndarray
weights = weights / nanmean(weights)
self._weight_data = IVData(weights, var_name='weights', nobs=self._nobs)
def __init__(self, dependent: ArrayLike, exog: OptionalArrayLike,
endog: OptionalArrayLike, instruments: OptionalArrayLike, *,
weights: OptionalArrayLike = None, fuller: Numeric = 0,
kappa: OptionalNumeric = None):
self.dependent = IVData(dependent, var_name='dependent')
nobs = self.dependent.shape[0] # type: int
self.exog = IVData(exog, var_name='exog', nobs=nobs)
self.endog = IVData(endog, var_name='endog', nobs=nobs)
self.instruments = IVData(instruments, var_name='instruments', nobs=nobs)
self._original_index = self.dependent.pandas.index
if weights is None:
weights = ones(self.dependent.shape)
weights = IVData(weights).ndarray
if any(weights <= 0):
raise ValueError('weights must be strictly positive.')
weights = weights / nanmean(weights)
self.weights = IVData(weights, var_name='weights', nobs=nobs)
self._drop_locs = self._drop_missing()
# dependent variable
w = sqrt(self.weights.ndarray)
self._y = self.dependent.ndarray
self._wy = self._y * w
def __init__(self, dependent: ArrayLike, exog: OptionalArrayLike = None, *,
absorb: InteractionVar = None,
interactions: Union[InteractionVar, Iterable[InteractionVar]] = None,
weights: OptionalArrayLike = None):
self._dependent = IVData(dependent, 'dependent')
self._nobs = nobs = self._dependent.shape[0]
self._exog = IVData(exog, 'exog', nobs=self._nobs)
self._absorb = absorb
if isinstance(absorb, DataFrame):
self._absorb_inter = Interaction.from_frame(absorb)
elif absorb is None:
self._absorb_inter = Interaction(None, None, nobs)
elif isinstance(absorb, Interaction):
self._absorb_inter = absorb
else:
raise TypeError('absorb must ba a DataFrame or an Interaction')
self._weights = weights
self._is_weighted = False
self._check_weights()
self._interactions = interactions
Model instance
Notes
-----
At least one of exog or endog must be provided.
Utility function to simplify the construction of multivariate IV
models which all use the same regressors and instruments. Constructs
the dictionary of equations from the variables using the common
exogenous, endogenous and instrumental variables.
"""
equations = OrderedDict()
dependent = IVData(dependent, var_name='dependent')
if exog is None and endog is None:
raise ValueError('At least one of exog or endog must be provided')
exog = IVData(exog, var_name='exog')
endog = IVData(endog, var_name='endog', nobs=dependent.shape[0])
instr = IVData(instruments, var_name='instruments', nobs=dependent.shape[0])
for col in dependent.pandas:
equations[col] = (dependent.pandas[[col]], exog.pandas, endog.pandas, instr.pandas)
return cls(equations)
dep_name = 'dependent_' + str(i)
exog_name = 'exog_' + str(i)
endog_name = 'endog_' + str(i)
instr_name = 'instr_' + str(i)
if isinstance(eq_data, (tuple, list)):
dep = IVData(eq_data[0], var_name=dep_name)
self._dependent.append(dep)
current_id = id(eq_data[1])
self._exog.append(IVData(eq_data[1], var_name=exog_name, nobs=dep.shape[0]))
endog = IVData(eq_data[2], var_name=endog_name, nobs=dep.shape[0])
if endog.shape[1] > 0:
current_id = (current_id, id(eq_data[2]))
ids.append(current_id)
self._endog.append(endog)
self._instr.append(IVData(eq_data[3], var_name=instr_name, nobs=dep.shape[0]))
if len(eq_data) == 5:
self._weights.append(IVData(eq_data[4]))
else:
dep = self._dependent[-1].ndarray
self._weights.append(IVData(np.ones_like(dep)))
elif isinstance(eq_data, (dict, Mapping)):
dep = IVData(eq_data['dependent'], var_name=dep_name)
self._dependent.append(dep)
exog = eq_data.get('exog', None)
self._exog.append(IVData(exog, var_name=exog_name, nobs=dep.shape[0]))
current_id = id(exog)
endog = eq_data.get('endog', None)
endog = IVData(endog, var_name=endog_name, nobs=dep.shape[0])
self._exog.append(IVData(eq_data[1], var_name=exog_name, nobs=dep.shape[0]))
endog = IVData(eq_data[2], var_name=endog_name, nobs=dep.shape[0])
if endog.shape[1] > 0:
current_id = (current_id, id(eq_data[2]))
ids.append(current_id)
self._endog.append(endog)
self._instr.append(IVData(eq_data[3], var_name=instr_name, nobs=dep.shape[0]))
if len(eq_data) == 5:
self._weights.append(IVData(eq_data[4]))
else:
dep = self._dependent[-1].ndarray
self._weights.append(IVData(np.ones_like(dep)))
elif isinstance(eq_data, (dict, Mapping)):
dep = IVData(eq_data['dependent'], var_name=dep_name)
self._dependent.append(dep)
exog = eq_data.get('exog', None)
self._exog.append(IVData(exog, var_name=exog_name, nobs=dep.shape[0]))
current_id = id(exog)
endog = eq_data.get('endog', None)
endog = IVData(endog, var_name=endog_name, nobs=dep.shape[0])
self._endog.append(endog)
if 'endog' in eq_data:
current_id = (current_id, id(eq_data['endog']))
ids.append(current_id)
instr = eq_data.get('instruments', None)
instr = IVData(instr, var_name=instr_name, nobs=dep.shape[0])
self._instr.append(instr)
def _check_weights(self):
if self._weights is None:
nobs = self._dependent.shape[0]
self._is_weighted = False
self._weight_data = IVData(ones(nobs), 'weights')
else:
self._is_weighted = True
weights = IVData(self._weights).ndarray
weights = weights / nanmean(weights)
self._weight_data = IVData(weights, var_name='weights', nobs=self._nobs)
def _check_data(self):
cat, cont = self._cat, self._cont
cat_nobs = getattr(cat, 'shape', (0,))[0]
cont_nobs = getattr(cont, 'shape', (0,))[0]
nobs = max(cat_nobs, cont_nobs)
if cat is None and cont is None:
if self._nobs is not None:
self._cont_data = self._cat_data = IVData(None, 'none', nobs=self._nobs)
else:
raise ValueError('nobs must be provided when cat and cont are None')
return
self._nobs = nobs
self._cat_data = IVData(cat, 'cat', nobs=nobs, convert_dummies=False)
self._cont_data = IVData(cont, 'cont', nobs=nobs, convert_dummies=False)
if self._cat_data.shape[1] == self._cont_data.shape[1] == 0:
raise ValueError('Both cat and cont are empty arrays')
cat_data = self._cat_data.pandas
convert = [col for col in cat_data if not (is_categorical(cat_data[col]))]
if convert:
cat_data = DataFrame({col: cat_data[col].astype('category') for col in cat_data})
self._cat_data = IVData(cat_data, 'cat', convert_dummies=False)
def _check_weights(self):
if self._weights is None:
nobs = self._dependent.shape[0]
self._is_weighted = False
self._weight_data = IVData(ones(nobs), 'weights')
else:
self._is_weighted = True
weights = IVData(self._weights).ndarray
weights = weights / nanmean(weights)
self._weight_data = IVData(weights, var_name='weights', nobs=self._nobs)