Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
[(0.4, InstrumentalVariableEstimator, [0,1], [1,2], [0,], [1, 2], [False, True], [False,]),])
def test_average_treatment_effect(self, error_tolerance, Estimator,
num_common_causes, num_instruments, num_effect_modifiers,
num_treatments, treatment_is_binary, outcome_is_binary
):
estimator_tester = TestEstimator(error_tolerance, Estimator)
# Not using testsuite from .base/TestEstimtor, custom code below
args_dict = {
'num_common_causes': num_common_causes,
'num_instruments': num_instruments,
'num_effect_modifiers': num_effect_modifiers,
'num_treatments': num_treatments,
'treatment_is_binary': treatment_is_binary,
'outcome_is_binary': outcome_is_binary
}
keys, values = zip(*args_dict.items())
configs = [dict(zip(keys, v)) for v in itertools.product(*values)]
def _estimate_effect(self):
upper_limit = self.rd_threshold_value + self.rd_bandwidth
lower_limit = self.rd_threshold_value - self.rd_bandwidth
rows_filter = np.s_[(self.rd_variable >= lower_limit) & (self.rd_variable <= upper_limit)]
local_rd_variable = self.rd_variable[rows_filter]
local_treatment_variable = self._treatment[self._treatment_name[0]][rows_filter] # indexing by treatment name again since this method assumes a single-dimensional treatment
local_outcome_variable = self._outcome[rows_filter]
local_df = pd.DataFrame(data={
'local_rd_variable': local_rd_variable,
'local_treatment': local_treatment_variable,
'local_outcome': local_outcome_variable
})
print(local_df)
iv_estimator = InstrumentalVariableEstimator(
local_df,
self._target_estimand,
['local_treatment'],
['local_outcome'],
test_significance=self._significance_test,
params={'iv_instrument_name': 'local_rd_variable'}
)
est = iv_estimator.estimate_effect()
return est