Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def custom_data_average_treatment_effect_test(self, data):
model = CausalModel(
data=data['df'],
treatment=data["treatment_name"],
outcome=data["outcome_name"],
graph=data["gml_graph"],
proceed_when_unidentifiable=True,
test_significance=None
)
target_estimand = model.identify_effect()
estimator_ate = self._Estimator(
data['df'],
identified_estimand=target_estimand,
treatment=data["treatment_name"],
outcome=data["outcome_name"],
test_significance=None
)
true_ate = data["ate"]
def null_refutation_test(self, data=None, dataset="linear", beta=10,
num_common_causes=1, num_instruments=1, num_samples=100000,
treatment_is_binary=True):
# Supports user-provided dataset object
if data is None:
data = dowhy.datasets.linear_dataset(beta=beta,
num_common_causes=num_common_causes,
num_instruments=num_instruments,
num_samples=num_samples,
treatment_is_binary=treatment_is_binary)
model = CausalModel(
data=data['df'],
treatment=data["treatment_name"],
outcome=data["outcome_name"],
graph=data["gml_graph"],
proceed_when_unidentifiable=True,
test_significance=None
)
target_estimand = model.identify_effect()
ate_estimate = model.estimate_effect(
identified_estimand=target_estimand,
method_name=self.estimator_method,
test_significance=None
)
true_ate = data["ate"]
# To test if there are any exceptions
def test_graph_input(self, beta, num_instruments, num_samples, num_treatments):
num_common_causes = 5
data = dowhy.datasets.linear_dataset(beta=beta,
num_common_causes=num_common_causes,
num_instruments=num_instruments,
num_samples=num_samples,
num_treatments = num_treatments,
treatment_is_binary=True)
model = CausalModel(
data=data['df'],
treatment=data["treatment_name"],
outcome=data["outcome_name"],
graph=data["gml_graph"],
proceed_when_unidentifiable=True,
test_significance=None
)
# removing two common causes
gml_str = 'graph[directed 1 node[ id "{0}" label "{0}"]node[ id "{1}" label "{1}"]node[ id "Unobserved Confounders" label "Unobserved Confounders"]edge[source "{0}" target "{1}"]edge[source "Unobserved Confounders" target "{0}"]edge[source "Unobserved Confounders" target "{1}"]node[ id "X0" label "X0"] edge[ source "X0" target "{0}"] node[ id "X1" label "X1"] edge[ source "X1" target "{0}"] node[ id "X2" label "X2"] edge[ source "X2" target "{0}"] edge[ source "X0" target "{1}"] edge[ source "X1" target "{1}"] edge[ source "X2" target "{1}"] node[ id "Z0" label "Z0"] edge[ source "Z0" target "{0}"]]'.format(data["treatment_name"][0], data["outcome_name"])
print(gml_str)
model = CausalModel(
data=data['df'],
treatment=data["treatment_name"],
outcome=data["outcome_name"],
graph=gml_str,
proceed_when_unidentifiable=True,
num_samples=num_samples,
num_treatments = num_treatments,
treatment_is_binary=True)
model = CausalModel(
data=data['df'],
treatment=data["treatment_name"],
outcome=data["outcome_name"],
graph=data["gml_graph"],
proceed_when_unidentifiable=True,
test_significance=None
)
# removing two common causes
gml_str = 'graph[directed 1 node[ id "{0}" label "{0}"]node[ id "{1}" label "{1}"]node[ id "Unobserved Confounders" label "Unobserved Confounders"]edge[source "{0}" target "{1}"]edge[source "Unobserved Confounders" target "{0}"]edge[source "Unobserved Confounders" target "{1}"]node[ id "X0" label "X0"] edge[ source "X0" target "{0}"] node[ id "X1" label "X1"] edge[ source "X1" target "{0}"] node[ id "X2" label "X2"] edge[ source "X2" target "{0}"] edge[ source "X0" target "{1}"] edge[ source "X1" target "{1}"] edge[ source "X2" target "{1}"] node[ id "Z0" label "Z0"] edge[ source "Z0" target "{0}"]]'.format(data["treatment_name"][0], data["outcome_name"])
print(gml_str)
model = CausalModel(
data=data['df'],
treatment=data["treatment_name"],
outcome=data["outcome_name"],
graph=gml_str,
proceed_when_unidentifiable=True,
test_significance=None,
missing_nodes_as_confounders=True
)
assert all(node_name in model._common_causes for node_name in ["X1", "X2"])
num_common_causes=1, num_instruments=1,
num_effect_modifiers=0, num_treatments=1,
num_samples=100000,
treatment_is_binary=True,
outcome_is_binary=False,
method_params=None):
data = dowhy.datasets.linear_dataset(beta=beta,
num_common_causes=num_common_causes,
num_instruments=num_instruments,
num_effect_modifiers = num_effect_modifiers,
num_treatments = num_treatments,
num_samples=num_samples,
treatment_is_binary=treatment_is_binary,
outcome_is_binary = outcome_is_binary)
model = CausalModel(
data=data['df'],
treatment=data["treatment_name"],
outcome=data["outcome_name"],
graph=data["gml_graph"],
proceed_when_unidentifiable=True,
test_significance=None
)
target_estimand = model.identify_effect()
estimator_ate = self._Estimator(
data['df'],
identified_estimand=target_estimand,
treatment=data["treatment_name"],
outcome=data["outcome_name"],
control_value = 0,
treatment_value = 1,
test_significance=None,