def _estimate_effect()

in dowhy/causal_estimators/two_stage_regression_estimator.py [0:0]

• 73 lines of code
• 7 McCabe index (conditional complexity)

    def _estimate_effect(self):
        #first_stage_features = self.build_first_stage_features()
        #fs_model = self.first_stage_model()
        #if self._target_estimand.identifier_method=="frontdoor":
        #    first_stage_outcome = self._frontdoor_variables
        #elif self._target_estimand.identifier_method=="mediation":
        #    first_stage_outcome = self._mediators
        #fs_model.fit(first_stage_features, self._frontdoor_variables)
        #self.logger.debug("Coefficients of the fitted model: " +
        #                  ",".join(map(str, fs_model.coef_)))
        #residuals = self._frontdoor_variables - fs_model.predict(first_stage_features)
        #self._data["residual"] = residuals
        estimate_value = None
        # First stage
        modified_target_estimand = copy.deepcopy(self._target_estimand)
        modified_target_estimand.identifier_method="backdoor"
        modified_target_estimand.backdoor_variables = self._target_estimand.mediation_first_stage_confounders
        if self._target_estimand.identifier_method=="frontdoor":
            modified_target_estimand.outcome_variable = parse_state(self._frontdoor_variables_names)
        elif self._target_estimand.identifier_method=="mediation":
            modified_target_estimand.outcome_variable = parse_state(self._mediators_names)

        first_stage_estimate = self.first_stage_model(self._data,
                 modified_target_estimand,
                 self._treatment_name,
                 parse_state(modified_target_estimand.outcome_variable),
                 control_value=self._control_value,
                 treatment_value=self._treatment_value,
                 test_significance=self._significance_test,
                 evaluate_effect_strength=self._effect_strength_eval,
                 confidence_intervals = self._confidence_intervals,
                 target_units=self._target_units,
                 effect_modifiers=self._effect_modifier_names,
                 params=self.method_params)._estimate_effect()

        # Second Stage
        modified_target_estimand = copy.deepcopy(self._target_estimand)
        modified_target_estimand.identifier_method="backdoor"
        modified_target_estimand.backdoor_variables = self._target_estimand.mediation_second_stage_confounders
        if self._target_estimand.identifier_method=="frontdoor":
            modified_target_estimand.treatment_variable = parse_state(self._frontdoor_variables_names)
        elif self._target_estimand.identifier_method=="mediation":
            modified_target_estimand.treatment_variable = parse_state(self._mediators_names)

        second_stage_estimate = self.second_stage_model(self._data,
                 modified_target_estimand,
                 parse_state(modified_target_estimand.treatment_variable),
                 parse_state(self._outcome_name), # to convert it to array before passing to causal estimator
                 control_value=self._control_value,
                 treatment_value=self._treatment_value,
                 test_significance=self._significance_test,
                 evaluate_effect_strength=self._effect_strength_eval,
                 confidence_intervals = self._confidence_intervals,
                 target_units=self._target_units,
                 effect_modifiers=self._effect_modifier_names,
                 params=self.method_params)._estimate_effect()
        # Combining the two estimates
        natural_indirect_effect = first_stage_estimate.value * second_stage_estimate.value
        # This same estimate is valid for frontdoor as well as mediation (NIE)
        estimate_value = natural_indirect_effect
        self.symbolic_estimator = self.construct_symbolic_estimator(
                first_stage_estimate.realized_estimand_expr,
                second_stage_estimate.realized_estimand_expr,
                estimand_type=CausalIdentifier.NONPARAMETRIC_NIE)
        if self._target_estimand.estimand_type == CausalIdentifier.NONPARAMETRIC_NDE:
            # Total  effect of treatment
            modified_target_estimand = copy.deepcopy(self._target_estimand)
            modified_target_estimand.identifier_method="backdoor"

            total_effect_estimate = self.second_stage_model(self._data,
                     modified_target_estimand,
                     self._treatment_name,
                     parse_state(self._outcome_name),
                     control_value=self._control_value,
                     treatment_value=self._treatment_value,
                     test_significance=self._significance_test,
                     evaluate_effect_strength=self._effect_strength_eval,
                     confidence_intervals = self._confidence_intervals,
                     target_units=self._target_units,
                     effect_modifiers=self._effect_modifier_names,
                     params=self.method_params)._estimate_effect()
            natural_direct_effect = total_effect_estimate.value - natural_indirect_effect
            estimate_value = natural_direct_effect
            self.symbolic_estimator = self.construct_symbolic_estimator(
                    first_stage_estimate.realized_estimand_expr,
                    second_stage_estimate.realized_estimand_expr,
                    total_effect_estimate.realized_estimand_expr,
                    estimand_type=self._target_estimand.estimand_type)
        return CausalEstimate(estimate=estimate_value,
                              control_value=self._control_value,
                              treatment_value=self._treatment_value,
                              target_estimand=self._target_estimand,
                              realized_estimand_expr=self.symbolic_estimator)