joint_generator = TCMGenerator(
engine="polars",
min_length=384,
max_length=384,
freq="h",
multivariate=True,
n_vars_range=(4, 6),
max_lag_range=(1, 12),
edge_probability_range=(0.15, 0.3),
seed=1,
)
joint_df = joint_generator.generate(n_series=4)
# Edges in a temporal SCM are lagged, so dependence between nodes shows up in
# lagged cross-correlations rather than the contemporaneous correlation matrix.
def max_lagged_xcorr(x: np.ndarray, y: np.ndarray, max_lag: int) -> float:
"""Largest |corr(x_t, y_{t-k})| over k = -max_lag..max_lag."""
best = abs(np.corrcoef(x, y)[0, 1])
for k in range(1, max_lag + 1):
best = max(
best,
abs(np.corrcoef(x[k:], y[: len(y) - k])[0, 1]),
abs(np.corrcoef(y[k:], x[: len(x) - k])[0, 1]),
)
return best
wide = joint_df.pivot(on="unique_id", index="ds", values="y").sort("ds")
nodes = [column for column in wide.columns if column != "ds"]
values = {node: wide[node].to_numpy() for node in nodes}
# Baseline: the same statistic on independent SCM draws (no shared graph).
independent_baseline_df = TCMGenerator(
engine="polars", min_length=384, max_length=384, freq="h", seed=101
).generate(n_series=4)
wide_ind = independent_baseline_df.pivot(on="unique_id", index="ds", values="y").sort(
"ds"
)
ind_nodes = [column for column in wide_ind.columns if column != "ds"]
ind_values = {node: wide_ind[node].to_numpy() for node in ind_nodes}
print("Max |cross-correlation| over lags 0..12 per node pair:\n")
print(" shared SCM (one causal graph):")
for i in range(len(nodes)):
for j in range(i + 1, len(nodes)):
xc = max_lagged_xcorr(values[nodes[i]], values[nodes[j]], max_lag=12)
print(f" {nodes[i]} vs {nodes[j]}: {xc:.2f}")
print("\n independent draws (baseline noise level):")
for i in range(len(ind_nodes)):
for j in range(i + 1, len(ind_nodes)):
xc = max_lagged_xcorr(ind_values[ind_nodes[i]], ind_values[ind_nodes[j]], 12)
print(f" {ind_nodes[i]} vs {ind_nodes[j]}: {xc:.2f}")
joint_df.head()
