import pandas as pd
from tigramite import data_processing as pp
from tigramite.independence_tests.parcorr import ParCorr
from tigramite.pcmci import PCMCI
from elasticsearch import Elasticsearch
from elasticsearch.helpers import scan
import networkx as nx
import matplotlib.pyplot as plt
import plotly.graph_objects as go
def load_csv_data(file_path):
"""
Load data from a CSV file and ensure it has multiple rows.
"""
df = pd.read_csv(file_path)
print("Loaded data shape:", df.shape)
if 'date_time' not in df.columns or 'measType' not in df.columns or 'measValue' not in df.columns:
raise KeyError("Data must contain 'date_time', 'measType', and 'measValue' columns.")
df['date_time'] = pd.to_datetime(df['date_time'])
df = df[['date_time', 'measType', 'measValue']] # Select relevant columns
df = df.pivot(index='date_time', columns='measType', values='measValue') # Pivot for wide format
df.fillna(0, inplace=True) # Handle missing values
if df.shape[0] < 2: # Ensure at least two observations
raise ValueError("Not enough observations for PCMCI analysis.")
return df
def load_elasticsearch_data(es_host, index_name):
"""
Load data from Elasticsearch.
"""
es = Elasticsearch(es_host, timeout=60)
query = {"query": {"match_all": {}}}
results = scan(es, index=index_name, query=query)
data = []
for item in results:
source = item['_source']
if 'date_time' in source and 'measType' in source and 'measValue' in source:
data.append({
'date_time': source['date_time'],
'measType': source['measType'],
'measValue': source['measValue']
})
if not data:
raise ValueError("No data found in the specified Elasticsearch index.")
df = pd.DataFrame(data)
df['date_time'] = pd.to_datetime(df['date_time'])
df = df.pivot(index='date_time', columns='measType', values='measValue') # Pivot for wide format
df.fillna(0, inplace=True) # Handle missing values
return df
def run_pcmci(df):
"""
Run PCMCI on the processed DataFrame.
"""
data_array = df.to_numpy()
dataframe = pp.DataFrame(data_array, var_names=df.columns.tolist())
parcorr = ParCorr(significance='analytic')
pcmci = PCMCI(dataframe=dataframe, cond_ind_test=parcorr)
results = pcmci.run_pcmci(tau_max=3, pc_alpha=0.05)
print("Significant Links:")
pcmci.print_significant_links(
p_matrix=results['p_matrix'],
val_matrix=results['val_matrix'],
alpha_level=0.05
)
return results, dataframe
def visualize_pcmci_graph(results, dataframe, alpha_level=0.05):
"""
Visualize the PCMCI causal graph.
"""
val_matrix = results['val_matrix']
p_matrix = results['p_matrix']
G = nx.DiGraph()
# Add edges for significant causal links
for i, var_from in enumerate(dataframe.var_names):
for j, var_to in enumerate(dataframe.var_names):
if i != j:
for lag in range(1, 4):
if p_matrix[i, j, lag - 1] < alpha_level:
weight = val_matrix[i, j, lag - 1]
G.add_edge(f"{var_from} (t-{lag})", var_to, weight=weight)
if G.number_of_edges() == 0:
print("No significant links to display in the causal graph.")
return
pos = nx.spring_layout(G, seed=42)
plt.figure(figsize=(12, 10))
nx.draw(
G, pos, with_labels=True, node_size=3000, node_color='skyblue',
font_size=10, font_weight='bold', edge_color='gray'
)
edge_labels = {(u, v): f"{d['weight']:.2f}" for u, v, d in G.edges(data=True)}
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_color='red', font_size=8)
plt.title('PCMCI Causal Graph with Significant Links')
plt.show()
def visualize_pcmci_graph_interactive(results, dataframe, alpha_level=0.05):
"""
Visualize the PCMCI causal graph interactively using Plotly.
"""
val_matrix = results['val_matrix']
p_matrix = results['p_matrix']
G = nx.DiGraph()
# Add edges for significant causal links
for i, var_from in enumerate(dataframe.var_names):
for j, var_to in enumerate(dataframe.var_names):
if i != j:
for lag in range(1, 4): # Lag range: 1 to 3
if p_matrix[i, j, lag - 1] < alpha_level:
weight = val_matrix[i, j, lag - 1]
G.add_edge(f"{var_from} (t-{lag})", var_to, weight=weight)
if G.number_of_edges() == 0:
print("No significant links to display in the causal graph.")
return
# Get positions for nodes
pos = nx.spring_layout(G, seed=42)
# Create edge traces
edge_x = []
edge_y = []
edge_weights = []
for edge in G.edges(data=True):
x0, y0 = pos[edge[0]]
x1, y1 = pos[edge[1]]
edge_x.append(x0)
edge_x.append(x1)
edge_x.append(None)
edge_y.append(y0)
edge_y.append(y1)
edge_y.append(None)
edge_weights.append(edge[2]['weight'])
edge_trace = go.Scatter(
x=edge_x, y=edge_y,
line=dict(width=0.5, color='#888'),
hoverinfo='none',
mode='lines'
)
# Create node traces
node_x = []
node_y = []
node_text = []
for node in G.nodes():
x, y = pos[node]
node_x.append(x)
node_y.append(y)
node_text.append(node)
node_trace = go.Scatter(
x=node_x, y=node_y,
mode='markers+text',
text=node_text,
textposition="top center",
marker=dict(
size=20,
color='skyblue',
line_width=2
),
hoverinfo='text'
)
# Combine edge and node traces into a single figure
fig = go.Figure(data=[edge_trace, node_trace],
layout=go.Layout(
title='Interactive PCMCI Causal Graph',
titlefont_size=16,
showlegend=False,
hovermode='closest',
margin=dict(b=0, l=0, r=0, t=40),
xaxis=dict(showgrid=False, zeroline=False),
yaxis=dict(showgrid=False, zeroline=False)
))
fig.show()
def convert_results_to_json(results, dataframe, alpha_level=0.05):
"""
Convert PCMCI results into JSON format for frontend UI.
"""
val_matrix = results['val_matrix']
p_matrix = results['p_matrix']
# Initialize the JSON structure
json_result = {
"nodes": [],
"edges": []
}
# Add nodes
for var_name in dataframe.var_names:
json_result["nodes"].append({
"id": var_name,
"label": var_name
})
# Add edges for significant causal links
for i, var_from in enumerate(dataframe.var_names):
for j, var_to in enumerate(dataframe.var_names):
if i != j:
for lag in range(1, 4): # Lag range: 1 to 3
if p_matrix[i, j, lag - 1] < alpha_level:
weight = val_matrix[i, j, lag - 1]
json_result["edges"].append({
"source": var_from,
"target": var_to,
"lag": lag,
"weight": weight,
"p_value": p_matrix[i, j, lag - 1]
})
return json_result
# Choose input method: CSV or Elasticsearch
source = "csv" # Change to 'elasticsearch' for Elasticsearch input
if source == "csv":
file_path = 'merged_kpi.csv'
data = load_csv_data(file_path)
elif source == "elasticsearch":
es_host = "http://localhost:9200"
index_name = "index_name"
data = load_elasticsearch_data(es_host, index_name)
# Run PCMCI analysis
results, dataframe = run_pcmci(data)
#Visualize the causal graph
#visualize_pcmci_graph(results, dataframe)
visualize_pcmci_graph_interactive(results, dataframe)
json_output = convert_results_to_json(results, dataframe, alpha_level=0.05)
import json
output_file = "pcmci_results.json"
with open(output_file, "w") as f:
json.dump(json_output, f, indent=4)
print(f"PCMCI results saved to {output_file}")
Ashoka Bhat Shared book 