in dowhy/causal_identifier.py [0:0]
def identify_ate_effect(self, optimize_backdoor):
estimands_dict = {}
mediation_first_stage_confounders = None
mediation_second_stage_confounders = None
### 1. BACKDOOR IDENTIFICATION
# First, checking if there are any valid backdoor adjustment sets
if optimize_backdoor == False:
backdoor_sets = self.identify_backdoor(self.treatment_name, self.outcome_name)
else:
from dowhy.causal_identifiers.backdoor import Backdoor
path = Backdoor(self._graph._graph, self.treatment_name, self.outcome_name)
backdoor_sets = path.get_backdoor_vars()
estimands_dict, backdoor_variables_dict = self.build_backdoor_estimands_dict(
self.treatment_name,
self.outcome_name,
backdoor_sets,
estimands_dict)
# Setting default "backdoor" identification adjustment set
default_backdoor_id = self.get_default_backdoor_set_id(backdoor_variables_dict)
if len(backdoor_variables_dict) > 0:
estimands_dict["backdoor"] = estimands_dict.get(str(default_backdoor_id), None)
backdoor_variables_dict["backdoor"] = backdoor_variables_dict.get(str(default_backdoor_id), None)
else:
estimands_dict["backdoor"] = None
### 2. INSTRUMENTAL VARIABLE IDENTIFICATION
# Now checking if there is also a valid iv estimand
instrument_names = self._graph.get_instruments(self.treatment_name,
self.outcome_name)
self.logger.info("Instrumental variables for treatment and outcome:" +
str(instrument_names))
if len(instrument_names) > 0:
iv_estimand_expr = self.construct_iv_estimand(
self.estimand_type,
self._graph.treatment_name,
self._graph.outcome_name,
instrument_names
)
self.logger.debug("Identified expression = " + str(iv_estimand_expr))
estimands_dict["iv"] = iv_estimand_expr
else:
estimands_dict["iv"] = None
### 3. FRONTDOOR IDENTIFICATION
# Now checking if there is a valid frontdoor variable
frontdoor_variables_names = self.identify_frontdoor()
self.logger.info("Frontdoor variables for treatment and outcome:" +
str(frontdoor_variables_names))
if len(frontdoor_variables_names) >0:
frontdoor_estimand_expr = self.construct_frontdoor_estimand(
self.estimand_type,
self._graph.treatment_name,
self._graph.outcome_name,
frontdoor_variables_names
)
self.logger.debug("Identified expression = " + str(frontdoor_estimand_expr))
estimands_dict["frontdoor"] = frontdoor_estimand_expr
mediation_first_stage_confounders = self.identify_mediation_first_stage_confounders(self.treatment_name, frontdoor_variables_names)
mediation_second_stage_confounders = self.identify_mediation_second_stage_confounders(frontdoor_variables_names, self.outcome_name)
else:
estimands_dict["frontdoor"] = None
# Finally returning the estimand object
estimand = IdentifiedEstimand(
self,
treatment_variable=self._graph.treatment_name,
outcome_variable=self._graph.outcome_name,
estimand_type=self.estimand_type,
estimands=estimands_dict,
backdoor_variables=backdoor_variables_dict,
instrumental_variables=instrument_names,
frontdoor_variables=frontdoor_variables_names,
mediation_first_stage_confounders=mediation_first_stage_confounders,
mediation_second_stage_confounders=mediation_second_stage_confounders,
default_backdoor_id = default_backdoor_id
)
return estimand