causal-inference

Causal Inference Skill

"Current deep learning is System 1: fast, intuitive, but easily fooled. We need System 2: slow, deliberate, causal." — Yoshua Bengio

Overview

Causal inference enables:

1. Interventional reasoning: What happens if I do X?
2. Counterfactual reasoning: What would have happened if...?
3. Transfer: Causal structure generalizes across domains
4. Robustness: Causal models resist distribution shift

Pearl's Causal Hierarchy

Level 3: Counterfactual (Imagining)
  "What would have happened if I had done X?"
  P(y_x | x', y')
          ▲
Level 2: Intervention (Doing)
  "What happens if I do X?"
  P(y | do(X))
          ▲
Level 1: Association (Seeing)
  "What does X tell me about Y?"
  P(y | x)

Structural Causal Models (SCM)

class StructuralCausalModel:
    """
    SCM: Variables, causal graph, structural equations.
    """
    
    def __init__(self, variables: List[str], graph: DAG, equations: Dict):
        self.variables = variables
        self.graph = graph  # Directed Acyclic Graph
        self.equations = equations  # X_i = f_i(parents(X_i), U_i)
    
    def intervene(self, intervention: Dict[str, float]) -> "SCM":
        """
        do(X = x): Replace equation for X with constant.
        
        This breaks incoming edges to X.
        """
        new_equations = self.equations.copy()
        for var, value in intervention.items():
            new_equations[var] = lambda *_: value
        
        new_graph = self.graph.remove_edges_to(intervention.keys())
        
        return StructuralCausalModel(
            self.variables, new_graph, new_equations
        )
    
    def counterfactual(self, evidence: Dict, intervention: Dict) -> Dict:
        """
        Counterfactual: What would Y be if X had been x, given we observed evidence?
        
        Three steps:
        1. Abduction: Infer noise terms from evidence
        2. Action: Apply intervention
        3. Prediction: Compute counterfactual outcome
        """
        # Step 1: Abduction - infer noise terms U
        noise_terms = self.abduct_noise(evidence)
        
        # Step 2: Action - apply intervention
        intervened_scm = self.intervene(intervention)
        
        # Step 3: Prediction - forward propagate with inferred noise
        counterfactual_world = intervened_scm.forward(noise_terms)
        
        return counterfactual_world


class CausalDiscovery:
    """
    Learn causal structure from data.
    """
    
    def __init__(self, data: pd.DataFrame):
        self.data = data
        
    def pc_algorithm(self) -> DAG:
        """
        PC Algorithm: Constraint-based causal discovery.
        
        1. Start with complete undirected graph
        2. Remove edges based on conditional independence tests
        3. Orient edges using v-structures and rules
        """
        from causallearn.search.ConstraintBased.PC import pc
        
        result = pc(self.data.values)
        return result.G
    
    def gflownet_discovery(self) -> Distribution[DAG]:
        """
        Use GFlowNet to sample DAGs proportional to likelihood.
        
        This gives a DISTRIBUTION over causal graphs,
        properly accounting for uncertainty.
        """
        from gflownet import CausalDAGGFlowNet
        
        gfn = CausalDAGGFlowNet(n_variables=len(self.data.columns))
        gfn.train(reward=lambda g: self.bayesian_score(g))
        
        # Sample multiple DAGs
        dag_samples = [gfn.sample() for _ in range(1000)]
        return dag_samples