from pathlib import Path
import os
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from sklearn.linear_model import LogisticRegression, LinearRegression
from sklearn.model_selection import KFold
def propensity_elastic_net(X: np.ndarray, treatment: np.ndarray) -> np.ndarray:
model = LogisticRegression(
penalty="elasticnet", solver="saga", l1_ratio=0.5, max_iter=2000
)
model.fit(X, treatment)
return model.predict_proba(X)[:, 1]
class SLearner:
def __init__(self, learner: str = "ranger"):
self.learner = learner
def fit(self, X, t, y):
Xt = np.column_stack([X, t])
if self.learner == "lm":
self.model_ = LinearRegression().fit(Xt, y)
else:
self.model_ = RandomForestRegressor(
n_estimators=100, random_state=42, n_jobs=-1
).fit(Xt, y)
return self
def predict(self, X):
y0 = self.model_.predict(np.column_stack([X, np.zeros(len(X))]))
y1 = self.model_.predict(np.column_stack([X, np.ones(len(X))]))
return y1 - y0
class TLearner:
def __init__(self, learner: str = "ranger"):
self.learner = learner
def fit(self, X, t, y):
t = t.astype(int)
if self.learner == "lm":
self.m0_ = LinearRegression().fit(X[t == 0], y[t == 0])
self.m1_ = LinearRegression().fit(X[t == 1], y[t == 1])
else:
self.m0_ = RandomForestRegressor(
n_estimators=100, random_state=42, n_jobs=-1
).fit(X[t == 0], y[t == 0])
self.m1_ = RandomForestRegressor(
n_estimators=100, random_state=42, n_jobs=-1
).fit(X[t == 1], y[t == 1])
return self
def predict(self, X):
return self.m1_.predict(X) - self.m0_.predict(X)
class XLearner:
def __init__(self, learner: str = "ranger"):
self.learner = learner
def _reg(self):
if self.learner == "lm":
return LinearRegression()
return RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1)
def fit(self, X, t, y, p=None):
t = t.astype(int)
self.propensity_model_ = LogisticRegression(
penalty="elasticnet", solver="saga", l1_ratio=0.5, max_iter=2000
).fit(X, t)
m0, m1 = self._reg(), self._reg()
m0.fit(X[t == 0], y[t == 0])
m1.fit(X[t == 1], y[t == 1])
d1 = y[t == 1] - m0.predict(X[t == 1])
d0 = m1.predict(X[t == 0]) - y[t == 0]
tau1 = self._reg().fit(X[t == 1], d1)
tau0 = self._reg().fit(X[t == 0], d0)
self.tau0_, self.tau1_ = tau0, tau1
return self
def predict(self, X):
p = self.propensity_model_.predict_proba(X)[:, 1]
return p * self.tau0_.predict(X) + (1 - p) * self.tau1_.predict(X)
class RLearner:
def __init__(self, learner: str = "ranger", n_fold: int = 3):
self.learner = learner
self.n_fold = n_fold
def _reg(self):
if self.learner == "lm":
return LinearRegression()
return RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1)
def fit(self, X, t, y, p=None):
t = t.astype(int).astype(float)
p = propensity_elastic_net(X, t) if p is None else p
e = t - p
m = self._reg()
kf = KFold(n_splits=self.n_fold, shuffle=True, random_state=42)
y_hat = np.zeros_like(y, dtype=float)
for tr, va in kf.split(X):
m.fit(X[tr], y[tr])
y_hat[va] = m.predict(X[va])
resid = y - y_hat
tau = self._reg()
tau.fit(X, resid / (e + 1e-6), sample_weight=e**2)
self.tau_ = tau
return self
def predict(self, X):
return self.tau_.predict(X)
def load_ihdp(replications: int = 2, random_state: int = 1):
base_url = "https://raw.githubusercontent.com/uber/causalml/master/docs/examples/data"
cols = ["treatment", "y_factual", "y_cfactual", "mu0", "mu1"] + [f"X{i}" for i in range(25)]
parts = []
for i in range(1, 10):
url = f"{base_url}/ihdp_npci_{i}.csv"
tmp = pd.read_csv(url, header=None)
tmp.columns = cols[: tmp.shape[1]]
parts.append(tmp)
df = pd.concat(parts, ignore_index=True)
if replications > 1:
df = pd.concat([df] * replications, ignore_index=True)
binfeats = [f"X{i}" for i in range(6, 25)]
contfeats = [f"X{i}" for i in range(6)]
xcols = binfeats + contfeats
X = df[xcols].to_numpy(dtype=float)
treatment = df["treatment"].to_numpy(dtype=int)
y = df["y_factual"].to_numpy(dtype=float)
tau = np.where(
treatment == 1,
df["y_factual"] - df["y_cfactual"],
df["y_cfactual"] - df["y_factual"],
)
n = len(X)
rng = np.random.default_rng(random_state)
val_idx = rng.choice(n, size=int(0.2 * n), replace=False)
train_idx = np.setdiff1d(np.arange(n), val_idx)
return df, X, treatment, y, tau, train_idx, val_idx
def simulate_hidden_confounder(n=2000, p=5, sigma=1.0, adj=0, random_state=42):
"""Synthetic DGP with latent confounder (CausalML-style benchmark)."""
rng = np.random.default_rng(random_state)
z = rng.normal(size=n)
X = np.column_stack([z + rng.normal(scale=sigma, size=n) for _ in range(p)])
e = 1 / (1 + np.exp(-(adj + 0.5 * z)))
w = rng.binomial(1, e)
b = 2 * z + rng.normal(scale=sigma, size=n)
tau = 0.5 + 0.3 * z
y = b + w * tau + rng.normal(scale=sigma, size=n)
return {"X": X, "y": y, "w": w, "tau": tau, "b": b, "e": e}
def naive_ate(t, y):
return y[t == 1].mean() - y[t == 0].mean()
def ate_table(estimates, true_ate, cis=None):
"""Summarize ATE estimates against ground truth."""
rows = []
for name, ate in estimates.items():
row = {
"Method": name,
"ATE": ate,
"AbsError": abs(ate - true_ate),
"PctError": abs(ate / true_ate - 1) * 100 if true_ate != 0 else np.nan,
}
if cis and name in cis:
lo, hi = cis[name]
row["CI_lo"] = lo
row["CI_hi"] = hi
row["CoversTruth"] = lo <= true_ate <= hi
rows.append(row)
rows.append({
"Method": "True ATE",
"ATE": true_ate,
"AbsError": 0.0,
"PctError": 0.0,
})
return pd.DataFrame(rows)
def plot_ate_comparison(estimates, true_ate, cis=None, title="ATE comparison"):
"""Forest-style plot of ATE estimates with optional confidence intervals."""
methods = list(estimates.keys())
vals = [estimates[m] for m in methods]
y_pos = np.arange(len(methods))
fig, ax = plt.subplots(figsize=(8, max(4, len(methods) * 0.6)))
ax.axvline(true_ate, color="black", linestyle="--", linewidth=1.5, label="True ATE")
ax.scatter(vals, y_pos, color="steelblue", s=60, zorder=3)
if cis:
for i, m in enumerate(methods):
if m in cis:
lo, hi = cis[m]
ax.plot([lo, hi], [y_pos[i], y_pos[i]], color="steelblue", linewidth=2)
ax.set_yticks(y_pos)
ax.set_yticklabels(methods)
ax.set_xlabel("Estimated ATE")
ax.set_title(title)
ax.legend(loc="lower right")
plt.tight_layout()
plt.show()
from pydeepcausalml.effect import NeuralDML
RUN_FAST = os.getenv("PYDEEPCAUSALML_FAST_RENDER", "true").lower() == "true"
set_seed(42)