Drawing DAGs

Time	Block
0:00 – 0:40	14.1 What a DAG says, and what it doesn't
0:40 – 1:30	14.2 Tools: causalkit and the customer table
1:30 – 1:45	Break
1:45 – 2:30	14.3 NorthMart's targeting DAG
2:30 – 3:20	14.4 Three building blocks: chain, fork, collider
3:20 – 3:35	Break
3:35 – 4:45	14.5 Day 14 mini-project: let the data test your DAG
4:45 – 5:00	14.6 Save your work with Git

---

14.1 What a DAG says, and what it doesn't (40 min)

Concept. A DAG is a set of variables (nodes) and arrows (edges). An arrow A → B says "A is a direct cause of B" (for some units, by some amount). Directed: arrows have a direction. Acyclic: you can't follow arrows back to where you started; nothing causes itself.

The most important information in a DAG is the arrows that are missing. A missing arrow is a strong claim: "A has no direct effect on B". Missing arrows are what make causal effects estimable, and they're what data can sometimes refute (14.5).

Vocabulary, with NorthMart examples:

Term	Meaning	Example
parent / child	direct cause / direct effect	loyalty is a parent of email
ancestor / descendant	cause / effect through any chain	loyalty is an ancestor of post_orders
path	any route along edges, ignoring direction	email ← intensity → post_orders
latent node	a variable that matters but isn't in the data	intensity: how much a customer likes to shop

A DAG says nothing about the size or shape of effects (linear, multiplicative…), only which variables affect which.

dbt also draws a DAG (Day 7): there, arrows mean "is built from"; here they mean "causes". Same mathematics, different meaning.

---

14.2 Tools: causalkit and the customer table (50 min)

We'll use DAGs every day from now on, so we put the helpers in a small shared package, causalkit. Create the folder and an empty causalkit/__init__.py:

mkdir -p causalkit day14 day15 day16 day17
touch causalkit/__init__.py

causalkit/dags.py builds DAGs with networkx, draws them, and (from Day 15) finds backdoor paths and adjustment sets. Read the docstring of each function; you'll use them all this week.

"""
Small DAG helpers for Module 2, built on networkx. Written to be read, not just used.

    from causalkit.dags import make_dag, draw_dag, backdoor_paths, is_valid_adjustment, minimal_adjustment_sets
"""
from itertools import combinations
from pathlib import Path

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
import networkx as nx


def make_dag(edges: list[tuple[str, str]], latent: set[str] | None = None) -> nx.DiGraph:
    """Build a DAG from (cause, effect) pairs. Latent (unobserved) nodes are marked."""
    g = nx.DiGraph(edges)
    if not nx.is_directed_acyclic_graph(g):
        raise ValueError(f"Not a DAG: there is a cycle {nx.find_cycle(g)}")
    for node in g.nodes:
        g.nodes[node]["latent"] = node in (latent or set())
    return g


def observed(g: nx.DiGraph) -> set[str]:
    return {n for n in g.nodes if not g.nodes[n]["latent"]}


def draw_dag(g: nx.DiGraph, path: str, title: str = "", pos: dict | None = None,
             treatment: str | None = None, outcome: str | None = None) -> None:
    """Save a picture of the DAG. Latent nodes are dashed; treatment and outcome are coloured."""
    pos = pos or _layered_layout(g)
    fig, ax = plt.subplots(figsize=(7.5, 4.8))
    colours = []
    for n in g.nodes:
        colours.append("#3987e5" if n == treatment else "#d95926" if n == outcome else "#e9eef5")
    nx.draw_networkx_edges(g, pos, ax=ax, arrowsize=18, width=1.6, node_size=2600,
                           edge_color="#4a5568", connectionstyle="arc3,rad=0.05")
    for n, c in zip(g.nodes, colours):
        nx.draw_networkx_nodes(g, pos, nodelist=[n], node_color=c, node_size=2600, ax=ax,
                               edgecolors="#1f2937", linewidths=1.5,
                               node_shape="o")
        if g.nodes[n]["latent"]:
            ax.scatter(*pos[n], s=2600, facecolors="white", edgecolors="#1f2937",
                       linewidths=1.5, linestyle="--", zorder=3)
    labels = {n: n.replace("_", "\n") for n in g.nodes}
    nx.draw_networkx_labels(g, pos, labels, font_size=8.5, ax=ax,
                            font_color="#111827")
    ax.set_title(title, fontsize=11)
    ax.axis("off")
    ax.margins(0.15)
    fig.tight_layout()
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    fig.savefig(path, dpi=150)
    plt.close(fig)


def _layered_layout(g: nx.DiGraph) -> dict:
    """Left-to-right layout by topological generation (causes on the left)."""
    for layer, nodes in enumerate(nx.topological_generations(g)):
        for node in nodes:
            g.nodes[node]["layer"] = layer
    return nx.multipartite_layout(g, subset_key="layer", align="vertical", scale=2)


def backdoor_paths(g: nx.DiGraph, treatment: str, outcome: str) -> list[list[str]]:
    """Every path from treatment to outcome that starts with an arrow INTO the treatment."""
    skeleton = g.to_undirected()
    paths = []
    for p in nx.all_simple_paths(skeleton, treatment, outcome):
        if g.has_edge(p[1], p[0]):            # first step goes against an arrow: p[1] -> treatment
            paths.append(p)
    return paths


def path_is_blocked(g: nx.DiGraph, path: list[str], z: set[str]) -> bool:
    """d-separation for one path: blocked by a non-collider in Z, or a collider not in Z (nor its descendants)."""
    for a, m, b in zip(path, path[1:], path[2:]):
        collider = g.has_edge(a, m) and g.has_edge(b, m)
        if collider:
            if m not in z and not (nx.descendants(g, m) & z):
                return True
        elif m in z:
            return True
    return False


def describe_path(g: nx.DiGraph, path: list[str]) -> str:
    out = path[0]
    for a, b in zip(path, path[1:]):
        out += f" -> {b}" if g.has_edge(a, b) else f" <- {b}"
    return out


def is_valid_adjustment(g: nx.DiGraph, treatment: str, outcome: str, z: set[str]) -> bool:
    """Backdoor criterion: Z has no descendant of the treatment and blocks every backdoor path."""
    if nx.descendants(g, treatment) & z:
        return False
    return all(path_is_blocked(g, p, z) for p in backdoor_paths(g, treatment, outcome))


def minimal_adjustment_sets(g: nx.DiGraph, treatment: str, outcome: str) -> list[set[str]]:
    """All smallest valid adjustment sets made only of OBSERVED variables (brute force: fine for small DAGs)."""
    candidates = sorted(observed(g) - {treatment, outcome} - nx.descendants(g, treatment))
    found: list[set[str]] = []
    for k in range(len(candidates) + 1):
        for combo in combinations(candidates, k):
            z = set(combo)
            if any(f <= z for f in found):
                continue                      # a smaller valid set is already inside
            if is_valid_adjustment(g, treatment, outcome, z):
                found.append(z)
    return found


def implied_independencies(g: nx.DiGraph, max_conditioning: int = 1) -> list[tuple[str, str, frozenset]]:
    """Testable claims of the DAG: pairs of OBSERVED, non-adjacent variables that are
    d-separated given some small set of observed variables."""
    obs = sorted(observed(g))
    claims = []
    for i, a in enumerate(obs):
        for b in obs[i + 1:]:
            if g.has_edge(a, b) or g.has_edge(b, a):
                continue
            others = [v for v in obs if v not in (a, b)]
            for k in range(max_conditioning + 1):
                hit = next((frozenset(z) for z in combinations(others, k)
                            if nx.is_d_separator(g, {a}, {b}, set(z))), None)
                if hit is not None:
                    claims.append((a, b, hit))
                    break
    return claims

causalkit/data.py builds one row per customer from Module 1's published mart: pre-period features plus post-period outcomes (weeks 9–16):

"""One row per customer for Module 2, built from Module 1's published Causal Data Mart."""
from pathlib import Path

import polars as pl

PUBLISHED = Path("data/published")


def load_customers() -> pl.DataFrame:
    """Pre-period features + post-period outcomes (weeks 9-16) for every customer."""
    features = pl.read_parquet(PUBLISHED / "customer_features.parquet")
    post = (pl.read_parquet(PUBLISHED / "panel_customer_week.parquet")
            .filter(pl.col("post") == 1)
            .group_by("customer_id")
            .agg(post_orders=pl.col("n_orders").sum(),
                 post_revenue=pl.col("revenue").sum().round(2),
                 post_active_weeks=(pl.col("n_orders") > 0).sum()))
    return (features.join(post, on="customer_id")
            .rename({"treated": "email"})
            .sort("customer_id"))

Module 2 reads only data/published/. If it's missing, run the Module 1 flow first: python pipeline/northmart_flow.py (with the API running).

---

14.3 NorthMart's targeting DAG (45 min)

Concept. Write the DAG from knowledge, not from data: how did marketing choose recipients? What drives orders? Here's what NorthMart's marketing team told us (and the Module 1 generator confirms):

• Loyalty members shop more, and marketing favoured them.
• Marketing also favoured heavy shoppers, judged by recent activity.
• How much someone likes to shop (intensity) drives orders both before and after the email. Nobody measures intensity directly.

Create day14/first_dag.py:

# day14/first_dag.py   (run with:  python -m day14.first_dag)
"""Write down how NorthMart's email was targeted, as a DAG."""
import networkx as nx

from causalkit.dags import draw_dag, make_dag

g = make_dag(
    [
        ("loyalty", "intensity"),     # loyalty members shop more
        ("loyalty", "email"),         # marketing favoured loyalty members
        ("intensity", "email"),       # ...and heavy shoppers
        ("intensity", "pre_orders"),  # heavy shoppers ordered more before
        ("intensity", "post_orders"), # ...and after
        ("email", "post_orders"),     # the effect we want
    ],
    latent={"intensity"},             # nobody observes "shopping intensity" directly
)

print("Nodes:", sorted(g.nodes))
print("Parents of email:      ", sorted(g.predecessors("email")))
print("Children of intensity: ", sorted(g.successors("intensity")))
print("Ancestors of post_orders:", sorted(nx.ancestors(g, "post_orders")))
print("Descendants of email:    ", sorted(nx.descendants(g, "email")))

draw_dag(g, "figures/day14_northmart_dag.png", "NorthMart: how the SPRING26 email was targeted",
         treatment="email", outcome="post_orders")
print("Saved figures/day14_northmart_dag.png")

python -m day14.first_dag

Nodes: ['email', 'intensity', 'loyalty', 'post_orders', 'pre_orders']
Parents of email:       ['intensity', 'loyalty']
Children of intensity:  ['email', 'post_orders', 'pre_orders']
Ancestors of post_orders: ['email', 'intensity', 'loyalty']
Descendants of email:     ['post_orders']
Saved figures/day14_northmart_dag.png

Open the figure. The text version:

                 ┌──────────► pre_orders
                 │
  loyalty ──► intensity ─────────────────► post_orders
     │           │                            ▲
     │           ▼                            │
     └───────► email ─────────────────────────┘

intensity is dashed in the figure: it's latent. Notice that pre_orders is a child of intensity, not a cause of the email: marketing didn't email people because of their pre-period order count, but because of how much they shop, which pre-period orders reflect imperfectly. That distinction will matter tomorrow.

---

14.4 Three building blocks: chain, fork, collider (50 min)

Concept. Every path in every DAG is built from three-node pieces. What happens to the association between the ends when you control for (condition on) the middle node?

Structure	Picture	x and y associated?	After controlling for m
Chain	x → m → y	yes (x causes y through m)	blocked
Fork	x ← m → y	yes (common cause)	blocked
Collider	x → m ← y	no	opened: spurious association

The first two match intuition. The collider is the surprise, and the source of many published mistakes.

Create day14/three_structures.py:

# day14/three_structures.py   (python -m day14.three_structures)
"""Chain, fork, collider: what happens to an association when you condition on the middle node."""
import numpy as np
import polars as pl
import statsmodels.formula.api as smf

rng = np.random.default_rng(14)
n = 20_000


def slope(df: pl.DataFrame, formula: str, var: str = "x") -> float:
    return smf.ols(formula, data=df.to_pandas()).fit().params[var]


# Chain:    x -> m -> y      (email -> site visits -> orders)
x = rng.normal(size=n); m = x + rng.normal(size=n); y = m + rng.normal(size=n)
chain = pl.DataFrame({"x": x, "m": m, "y": y})

# Fork:     x <- m -> y      (email <- intensity -> orders), no arrow from x to y
m = rng.normal(size=n); x = m + rng.normal(size=n); y = m + rng.normal(size=n)
fork = pl.DataFrame({"x": x, "m": m, "y": y})

# Collider: x -> m <- y      (email -> "ordered in March" <- intensity), x and y unrelated
x = rng.normal(size=n); y = rng.normal(size=n); m = x + y + rng.normal(size=n)
collider = pl.DataFrame({"x": x, "m": m, "y": y})

rows = []
for name, df in [("chain    x -> m -> y", chain), ("fork     x <- m -> y", fork), ("collider x -> m <- y", collider)]:
    rows.append({"structure": name,
                 "y on x": round(slope(df, "y ~ x"), 3),
                 "y on x, controlling m": round(slope(df, "y ~ x + m"), 3)})
with pl.Config(fmt_str_lengths=30):
    print(pl.DataFrame(rows))

│ structure            ┆ y on x ┆ y on x, controlling m │
│ chain    x -> m -> y ┆ 1.023  ┆ 0.021                 │
│ fork     x <- m -> y ┆ 0.497  ┆ 0.004                 │
│ collider x -> m <- y ┆ 0.001  ┆ -0.5                  │

• Chain: controlling for the mediator removes x's effect (it all went through m).
• Fork: controlling for the common cause removes the spurious association. This is why we control for confounders.
• Collider: x and y are independent, but among units with the same m, a high x must be offset by a low y, so a strong negative association appears out of nothing.

These three rules are d-separation. A path is blocked if it contains a chain or fork whose middle is controlled for, or a collider whose middle (and all its descendants) is not controlled for. path_is_blocked in causalkit implements exactly these rules.

---

14.5 Day 14 mini-project: let the data test your DAG (70 min)

Concept. A DAG can't be proven by data, but it can be refuted. Every missing arrow implies that some variables are independent, possibly given others. If the data show a clear association where the DAG says there is none, the DAG is wrong.

implied_independencies (end of causalkit/dags.py) lists every such claim between observed variables. We test each with a regression: under the claim, the coefficient should be zero.

We compare two candidate DAGs: ours (A), and the one a hurried analyst implicitly assumes when comparing groups directly (B: "the email was sent at random"). Both include age, which we claim affects nothing here.

Create day14/test_the_dag.py:

# day14/test_the_dag.py   (python -m day14.test_the_dag)
"""Mini-project: a DAG makes testable claims. Check them against NorthMart's data."""
import polars as pl
import statsmodels.formula.api as smf

from causalkit.dags import implied_independencies, make_dag
from causalkit.data import load_customers

df = load_customers().with_columns(pl.col("email").cast(pl.Int8), pl.col("is_loyalty").alias("loyalty")).to_pandas()

CANDIDATES = {
    "A: targeted email (our DAG)": make_dag(
        [("loyalty", "intensity"), ("loyalty", "email"), ("intensity", "email"),
         ("intensity", "pre_orders"), ("intensity", "post_orders"), ("email", "post_orders")],
        latent={"intensity"}),
    "B: 'the email was random'": make_dag(
        [("loyalty", "intensity"), ("intensity", "pre_orders"), ("intensity", "post_orders"),
         ("email", "post_orders")],
        latent={"intensity"}),
}
for g in CANDIDATES.values():
    g.add_node("age", latent=False)            # age is in the data but (we claim) affects nothing here


def test(a: str, b: str, z: frozenset) -> tuple[float, float]:
    """Regress b on a (+ z). Under the claim 'a independent of b given z', a's coefficient is 0."""
    formula = f"{b} ~ {a}" + "".join(f" + {v}" for v in sorted(z))
    fit = smf.ols(formula, data=df).fit(cov_type="HC2")
    return fit.params[a], fit.pvalues[a]


for name, g in CANDIDATES.items():
    print(f"\n{name}")
    for a, b, z in implied_independencies(g):
        coef, p = test(a, b, z)
        claim = f"{a} _||_ {b}" + (f" | {', '.join(sorted(z))}" if z else "")
        verdict = "consistent" if p > 0.01 else "REJECTED by the data"
        print(f"  {claim:<28} coef {coef:+.4f}  p = {p:.3f}  -> {verdict}")

python -m day14.test_the_dag

A: targeted email (our DAG)
  age _||_ email               coef +0.0005  p = 0.268  -> consistent
  age _||_ loyalty             coef +0.0001  p = 0.795  -> consistent
  age _||_ post_orders         coef +0.0013  p = 0.522  -> consistent
  age _||_ pre_orders          coef +0.0029  p = 0.150  -> consistent

B: 'the email was random'
  ...
  email _||_ loyalty           coef +0.2830  p = 0.000  -> REJECTED by the data
  email _||_ pre_orders        coef +1.3822  p = 0.000  -> REJECTED by the data

The "random email" DAG is refuted: emailed customers had 1.38 more pre-period orders. Our DAG survives, but notice how few claims it makes. Because intensity is latent, most of its independences involve intensity and can't be tested. A DAG with a latent confounder is hard to refute and hard to use: tomorrow shows the second half of that sentence.

Try it. Add an arrow age → email to DAG A (as if marketing targeted by age). Which claim disappears, and does the data say anything about whether the arrow is needed?

Solution

Add ("age", "email") to A's edge list. Two claims disappear, and only age _||_ loyalty and age _||_ pre_orders remain:

• age _||_ email goes, obviously: there is now an arrow.
• age _||_ post_orders goes too: age now affects orders through the email.
• You might expect age _||_ post_orders | email to appear instead. It doesn't, because email is a collider on the path age → email ← intensity → post_orders: controlling for it opens that path.

The data had p = 0.268 for age _||_ email: no evidence that marketing used age. Adding arrows never makes a DAG easier to refute; it only removes claims. Keep arrows out unless knowledge or data says they belong.

---

14.6 Save your work with Git (15 min)

git add .
git commit -m "Day 14: causalkit, NorthMart DAG, d-separation, testable implications"
git push

Day 14 checklist

• I can read a DAG: parents, children, ancestors, descendants, paths, latent nodes
• I can write a DAG from domain knowledge and justify each arrow (and each missing arrow)
• I can state what controlling for the middle of a chain, fork or collider does
• I can list and test a DAG's implied independences, and explain why a test can refute but not prove a DAG