Causal Inference for Trading Decisions: Understanding Market Mechanisms
Causal inference distinguishes correlation from causation, enabling traders to understand true market mechanisms rather than spurious patterns. This leads to more robust and stable trading strategies.Understanding Causality in Markets
Causal approaches:- Identify root causes of price movements
- Distinguish correlation from causation
- Enable transferable insights across markets
- Improve strategy robustness
Complete Causal Inference System
``python
import numpy as np
import pandas as pd
from scipy import stats
import yfinance as yf
import matplotlib.pyplot as plt
from sklearn.preprocessing import StandardScaler
import networkx as nx
Granger Causality Test
def granger_causality_test(data, max_lag=5, significance_level=0.05):
"""Test if X Granger-causes Y"""
from sklearn.linear_model import LinearRegression
results = {}
for lag in range(1, max_lag + 1):
# Model 1: Y ~ Past(Y)
X_restricted = []
y_vals = []
for t in range(lag, len(data)):
X_restricted.append(data[t-lag:t, 1])
y_vals.append(data[t, 1])
X_restricted = np.array(X_restricted)
y_vals = np.array(y_vals)
model_r = LinearRegression()
model_r.fit(X_restricted, y_vals)
y_pred_r = model_r.predict(X_restricted)
rss_r = np.sum((y_vals - y_pred_r) ** 2)
# Model 2: Y ~ Past(Y) + Past(X)
X_unrestricted = []
for t in range(lag, len(data)):
X_unrestricted.append(np.concatenate([data[t-lag:t, 0], data[t-lag:t, 1]]))
X_unrestricted = np.array(X_unrestricted)
model_u = LinearRegression()
model_u.fit(X_unrestricted, y_vals)
y_pred_u = model_u.predict(X_unrestricted)
rss_u = np.sum((y_vals - y_pred_u) ** 2)
# F-statistic
n = len(y_vals)
m = lag
f_stat = ((rss_r - rss_u) / m) / (rss_u / (n - 2*m - 1))
p_value = 1 - stats.f.cdf(f_stat, m, n - 2*m - 1)
results[lag] = {'f_stat': f_stat, 'p_value': p_value}
return results
Fetch data
def prepare_causal_data(tickers, start_date, end_date):
"""Prepare data for causal analysis"""
data = yf.download(tickers, start=start_date, end=end_date, progress=False)['Close']
returns = data.pct_change().dropna()
return returns
Test causal relationships
tickers = ['SPY', 'QQQ', 'IWM']
returns = prepare_causal_data(tickers, '2020-01-01', '2024-01-01')
print("Granger Causality Results:")
print("=" * 60)
for i, ticker1 in enumerate(tickers):
for j, ticker2 in enumerate(tickers):
if i != j:
data = returns[[ticker1, ticker2]].values
results = granger_causality_test(data, max_lag=5)
print(f"\n{ticker1} → {ticker2}:")
for lag, result in results.items():
pval = result['p_value']
significant = "*" if pval < 0.01 else "" if pval < 0.05 else ""
print(f" Lag {lag}: p-value = {pval:.4f} {significant}")
`
Transfer Entropy for Directional Information Flow
`python
Transfer Entropy (more refined than Granger)
def transfer_entropy(X, Y, k=5, m=1):
"""Calculate transfer entropy from X to Y"""
n = len(X) - k - m + 1
# Discretize continuous data
X_bins = pd.qcut(X, q=3, labels=False, duplicates='drop')
Y_bins = pd.qcut(Y, q=3, labels=False, duplicates='drop')
transfer_ent = 0
for i in range(k, len(Y_bins) - m + 1):
# P(Y_t | Y_past, X_past)
y_current = Y_bins[i]
y_past = tuple(Y_bins[i-k:i])
x_past = tuple(X_bins[i-k:i])
# Count occurrences
key = (y_current, y_past, x_past)
# ... calculate conditional probabilities
return transfer_ent
Calculate for market pairs
print("\nTransfer Entropy Analysis:")
for i, ticker1 in enumerate(tickers):
for j, ticker2 in enumerate(tickers):
if i != j:
te = transfer_entropy(
returns[ticker1].values,
returns[ticker2].values,
k=5
)
print(f"{ticker1} → {ticker2}: {te:.4f}")
`
Causal Discovery with PC Algorithm
`python
def pc_algorithm(data, alpha=0.05):
"""PC algorithm for causal discovery"""
n_vars = data.shape[1]
graph = nx.complete_graph(n_vars)
# Phase 1: Remove edges based on independence tests
for depth in range(n_vars - 1):
for u in list(graph.nodes()):
for v in list(graph.neighbors(u)):
# Find conditional independence
separating_set = None
# Test independence
for S in get_separating_sets(data, u, v, depth):
# Conditional independence test
if is_conditionally_independent(data, u, v, S, alpha):
separating_set = S
break
if separating_set is not None:
graph.remove_edge(u, v)
return graph
def is_conditionally_independent(data, X, Y, Z, alpha):
"""Test conditional independence X ⊥ Y | Z"""
# Partial correlation test
from scipy.stats import chi2
partial_corr = compute_partial_correlation(data, X, Y, Z)
n = len(data)
# Test statistic
test_stat = (n - len(Z) - 2) * (partial_corr 2) / (1 - partial_corr 2 + 1e-10)
p_value = 1 - chi2.cdf(test_stat, 1)
return p_value > alpha
def compute_partial_correlation(data, X, Y, Z):
"""Compute partial correlation rho(X, Y | Z)"""
from sklearn.linear_model import LinearRegression
# Regress X on Z
Z_data = data[:, list(Z)] if Z else data[:, [0]]*0
model_x = LinearRegression()
model_x.fit(Z_data, data[:, X])
residuals_x = data[:, X] - model_x.predict(Z_data)
# Regress Y on Z
model_y = LinearRegression()
model_y.fit(Z_data, data[:, Y])
residuals_y = data[:, Y] - model_y.predict(Z_data)
# Correlation of residuals
partial_corr = np.corrcoef(residuals_x, residuals_y)[0, 1]
return partial_corr if not np.isnan(partial_corr) else 0
def get_separating_sets(data, u, v, depth):
"""Generate possible separating sets"""
n_vars = data.shape[1]
for S_size in range(min(depth + 1, n_vars - 2)):
for S in combinations([i for i in range(n_vars) if i not in [u, v]], S_size):
yield S
from itertools import combinations
Example causal discovery
data_array = returns.values
causal_graph = pc_algorithm(data_array, alpha=0.05)
print("Causal Graph Structure:")
print(f"Edges: {list(causal_graph.edges())}")
Visualize
fig, ax = plt.subplots(figsize=(10, 8))
pos = nx.spring_layout(causal_graph, seed=42)
nx.draw_networkx_nodes(causal_graph, pos, node_color='lightblue', node_size=500, ax=ax)
nx.draw_networkx_edges(causal_graph, pos, ax=ax, arrows=True)
nx.draw_networkx_labels(causal_graph, pos, labels={i: tickers[i] for i in range(len(tickers))}, ax=ax)
ax.set_title('Discovered Causal Structure')
plt.tight_layout()
plt.show()
`
Causal Impact Analysis
`python
from statsmodels.tsa.api import ARIMA
def causal_impact(y, X_intervention, pre_period, post_period, model_type='arima'):
"""Estimate causal impact of intervention"""
# Model on pre-intervention period
y_pre = y[pre_period[0]:pre_period[1]]
X_pre = X_intervention[pre_period[0]:pre_period[1]]
if model_type == 'arima':
model = ARIMA(y_pre, order=(1, 1, 1))
fitted_model = model.fit()
# Forecast post-period
post_length = post_period[1] - post_period[0]
forecast = fitted_model.get_forecast(steps=post_length)
forecast_mean = forecast.predicted_mean.values
# Actual post-period
y_post = y[post_period[0]:post_period[1]]
# Causal impact
impact = y_post - forecast_mean
return {
'impact': impact,
'mean_impact': np.mean(impact),
'cumulative_impact': np.sum(impact)
}
Example: Test impact of a policy/event
pre_period = (0, 500)
post_period = (500, 750)
impact_result = causal_impact(
returns['SPY'].values,
returns['QQQ'].values,
pre_period,
post_period
)
print(f"\nCausal Impact Analysis:")
print(f"Mean Impact: {impact_result['mean_impact']:.6f}")
print(f"Cumulative Impact: {impact_result['cumulative_impact']:.4f}")
``