Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import copy
import numpy as np
from dowhy.causal_refuter import CausalRefutation
from dowhy.causal_refuter import CausalRefuter
class PlaceboTreatmentRefuter(CausalRefuter):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._placebo_type = kwargs["placebo_type"]
def refute_estimate(self):
num_rows = self._data.shape[0]
if self._placebo_type == "permute":
new_treatment = self._data[self._treatment_name].sample(frac=1).values
else:
new_treatment = np.random.randn(num_rows)
new_data = self._data.assign(placebo=new_treatment)
self.logger.debug(new_data[0:10])
estimator_class = self._estimate.params['estimator_class']
identified_estimand = copy.deepcopy(self._target_estimand)
def get_class_object(method_name, *args, **kwargs):
# from https://www.bnmetrics.com/blog/factory-pattern-in-python3-simple-version
try:
module_name = method_name
class_name = string.capwords(method_name, "_").replace("_", "")
refuter_module = import_module('.' + module_name,
package="dowhy.causal_refuters")
refuter_class = getattr(refuter_module, class_name)
assert issubclass(refuter_class, CausalRefuter)
except (AttributeError, AssertionError, ImportError):
raise ImportError('{} is not an existing causal refuter.'.format(method_name))
return refuter_class
import copy
import numpy as np
import pandas as pd
from dowhy.causal_refuter import CausalRefutation
from dowhy.causal_refuter import CausalRefuter
class AddUnobservedCommonCause(CausalRefuter):
"""Add an unobserved confounder for refutation.
TODO: Needs scaled version of the parameters and an interpretation module
(e.g., in comparison to biggest effect of known confounder)
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.effect_on_t = kwargs["confounders_effect_on_treatment"] if "confounders_effect_on_treatment" in kwargs else "binary_flip"
self.effect_on_y = kwargs["confounders_effect_on_outcome"] if "confounders_effect_on_outcome" in kwargs else "linear"
self.kappa_t = kwargs["effect_strength_on_treatment"]
self.kappa_y = kwargs["effect_strength_on_outcome"]
def refute_estimate(self):
import copy
import numpy as np
from dowhy.causal_refuter import CausalRefutation
from dowhy.causal_refuter import CausalRefuter
class RandomCommonCause(CausalRefuter):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def refute_estimate(self):
num_rows = self._data.shape[0]
new_data = self._data.assign(w_random=np.random.randn(num_rows))
self.logger.debug(new_data[0:10])
new_backdoor_variables = self._target_estimand.backdoor_variables + ['w_random']
estimator_class = self._estimate.params['estimator_class']
identified_estimand = copy.deepcopy(self._target_estimand)
# Adding a new backdoor variable to the identified estimand
identified_estimand.backdoor_variables = new_backdoor_variables
new_estimator = estimator_class(
new_data,
identified_estimand,
from dowhy.causal_refuter import CausalRefuter, CausalRefutation
class DataSubsetRefuter(CausalRefuter):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._subset_fraction = kwargs["subset_fraction"]
def refute_estimate(self):
new_data = self._data.sample(frac=self._subset_fraction)
new_estimator = self.get_estimator_object(new_data, self._target_estimand, self._estimate)
new_effect = new_estimator.estimate_effect()
refute = CausalRefutation(
self._estimate.value,
new_effect.value,
refutation_type="Refute: Use a subset of data"
)