At1uri/topological-data-analysis-vault / 05_multivariate_correlation_clique_complexes.py
At1uri's picture
download
raw
2.88 kB
"""
Multivariate Asset Correlation Clique & Flag Complexes
======================================================
Author: Ramprasad Atluri (At1uri)
Repository: At1uri/topological-data-analysis-vault
Constructs filtered clique complexes on cross-asset financial correlation matrices:
w_{ij} = sqrt(2 * (1 - rho_{ij}))
A k-simplex exists if all (k+1) assets have pairwise distance <= epsilon.
"""
import math
import numpy as np
from typing import List, Tuple, Dict, Any
class CorrelationCliqueComplex:
"""Simplicial clique complex over cross-asset correlation networks."""
def __init__(self, tickers: List[str], correlation_matrix: np.ndarray):
self.tickers = tickers
self.corr = np.array(correlation_matrix, dtype=np.float64)
self.N = len(tickers)
assert self.corr.shape == (self.N, self.N)
# Angular metric / correlation distance: d_ij = sqrt(2 * (1 - rho_ij))
self.dist_matrix = np.sqrt(np.maximum(0.0, 2.0 * (1.0 - self.corr)))
def filter_at_threshold(self, epsilon: float) -> Dict[str, Any]:
"""Finds all cliques (simplices) with edge weights <= epsilon."""
adj = self.dist_matrix <= epsilon
np.fill_diagonal(adj, False)
edges = []
for i in range(self.N):
for j in range(i + 1, self.N):
if adj[i, j]:
edges.append((self.tickers[i], self.tickers[j]))
triangles = []
for i in range(self.N):
for j in range(i + 1, self.N):
if adj[i, j]:
for k in range(j + 1, self.N):
if adj[i, k] and adj[j, k]:
triangles.append((self.tickers[i], self.tickers[j], self.tickers[k]))
# Graph density
possible_edges = self.N * (self.N - 1) / 2.0
density = len(edges) / possible_edges if possible_edges > 0 else 0.0
return {
"epsilon": float(epsilon),
"num_edges": len(edges),
"num_triangles": len(triangles),
"graph_density": density,
"edges": edges,
"triangles": triangles
}
if __name__ == "__main__":
print("=== MULTIVARIATE CORRELATION CLIQUE COMPLEXES ===")
assets = ["SPY", "QQQ", "IWM", "TLT", "GLD", "USO"]
corr = np.array([
[1.00, 0.88, 0.82, -0.35, 0.12, 0.25],
[0.88, 1.00, 0.85, -0.40, 0.15, 0.22],
[0.82, 0.85, 1.00, -0.30, 0.10, 0.30],
[-0.35, -0.40, -0.30, 1.00, 0.28, -0.15],
[0.12, 0.15, 0.10, 0.28, 1.00, 0.18],
[0.25, 0.22, 0.30, -0.15, 0.18, 1.00]
])
clique = CorrelationCliqueComplex(assets, corr)
for eps in [0.6, 0.9, 1.2]:
res = clique.filter_at_threshold(eps)
print(f"Epsilon {eps:.1f} -> Edges: {res['num_edges']} | Triangles: {res['num_triangles']} | Density: {res['graph_density']*100:.1f}%")

Xet Storage Details

Size:
2.88 kB
·
Xet hash:
94a84195d8ca6f7404acfa86e9019b3e5fcf8e8b9fdf6e089b1b9b342c50ecef

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.