Source code for dowhy.causal_estimators.regression_discontinuity_estimator

import numpy as np
import pandas as pd

from dowhy.causal_estimator import CausalEstimator
from dowhy.causal_estimators.instrumental_variable_estimator import InstrumentalVariableEstimator


[docs]class RegressionDiscontinuityEstimator(CausalEstimator): """Compute effect of treatment using the regression discontinuity method. Estimates effect by transforming the problem to an instrumental variables problem. Supports additional parameters that can be specified in the estimate_effect() method. * 'rd_variable_name': name of the variable on which the discontinuity occurs. This is the instrument. * 'rd_threshold_value': Threshold at which the discontinuity occurs. * 'rd_bandwidth': Distance from the threshold within which confounders can be considered the same between treatment and control. Considered band is (threshold +- bandwidth) """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.logger.info("Using Regression Discontinuity Estimator") self.rd_variable = self._data[self.rd_variable_name] self.symbolic_estimator = self.construct_symbolic_estimator(self._target_estimand) self.logger.info(self.symbolic_estimator) 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
[docs] def construct_symbolic_estimator(self, estimand): return ""