TDA modules

Mapper clustering

Mapper-based clustering and portfolio-weight construction.

This module contains the main utilities used to build Mapper graphs from financial price windows and to transform the resulting clusters into portfolio weights.

class tda_finance.tda.mapper_clustering.HACAClusterer(distance_threshold: float = 0.7, linkage: Literal['complete', 'average', 'single'] = 'average')

Bases: BaseEstimator, ClusterMixin

Agglomerative clustering wrapper compatible with KeplerMapper.

The wrapper computes a correlation distance matrix and then applies agglomerative clustering with a precomputed distance matrix. It is useful when the local clustering step should not discard observations as noise.

Parameters:

• distance_threshold (float, default=0.7) – Distance threshold above which clusters are not merged.

• linkage ({"complete", "average", "single"}, default="average") – Linkage criterion used by AgglomerativeClustering.

fit(X: Any, y: Any = None) → HACAClusterer

Fit the estimator.

This method is provided for compatibility with the scikit-learn API.

Parameters:

• X (array-like) – Input data.

• y (object, optional) – Ignored. Present for API compatibility.

Returns:

Fitted estimator.

Return type:

HACAClusterer

fit_predict(X: Any, y: Any = None) → ndarray

Cluster observations using agglomerative clustering.

Parameters:

• X (array-like) – Input data with observations in rows.

• y (object, optional) – Ignored. Present for API compatibility.

Returns:

Cluster labels for each observation.

Return type:

numpy.ndarray

class tda_finance.tda.mapper_clustering.MapperParams(pca_var: float = 0.8, umap_dim: int = 1, random_state: int = 1, n_cubes: int = 10, perc_overlap: float = 0.2, dbscan_eps: float = 0.3, dbscan_min_samples: int = 2, clusterer: str = 'dbscan', haca_distance_threshold: float = 0.7, haca_linkage: Literal['complete', 'average', 'single'] = 'average', min_assets: int = 5, verbose: bool = False)

Bases: object

Hyperparameters of the Mapper pipeline.

Parameters:

• pca_var (float, default=0.80) – Fraction of variance retained by the PCA step before applying UMAP.

• umap_dim (int, default=1) – Dimension of the final lens used by Mapper.

• random_state (int, default=1) – Random seed used in PCA and UMAP for reproducibility.

• n_cubes (int, default=10) – Number of intervals or hypercubes used in the Mapper cover.

• perc_overlap (float, default=0.2) – Percentage of overlap between adjacent cover elements.

• dbscan_eps (float, default=0.3) – Maximum distance between two samples for DBSCAN.

• dbscan_min_samples (int, default=2) – Minimum number of samples required by DBSCAN to form a core point.

• clusterer ({"dbscan", "haca", "hac", "agglomerative"}, default="dbscan") – Local clustering algorithm used inside each cover element.

• haca_distance_threshold (float, default=0.7) – Distance threshold used by the agglomerative clustering wrapper.

• haca_linkage ({"complete", "average", "single"}, default="average") – Linkage criterion used by the agglomerative clustering wrapper.

• min_assets (int, default=5) – Minimum number of assets required to attempt Mapper construction.

• verbose (bool, default=False) – Whether to print diagnostic information during Mapper construction.

clusterer: str = 'dbscan'

dbscan_eps: float = 0.3

dbscan_min_samples: int = 2

haca_distance_threshold: float = 0.7

haca_linkage: Literal['complete', 'average', 'single'] = 'average'

min_assets: int = 5

n_cubes: int = 10

pca_var: float = 0.8

perc_overlap: float = 0.2

random_state: int = 1

umap_dim: int = 1

verbose: bool = False

tda_finance.tda.mapper_clustering.build_clusters_from_prices(prices_window: DataFrame, params: MapperParams) → List[List[List[str]]] | None

Build Mapper clusters from a price window.

The function represents each asset as a vector of recent log-returns, standardizes each asset individually, builds a Mapper graph and returns the connected components of that graph as nested clusters of asset identifiers.

Parameters:

• prices_window (pandas.DataFrame) – Price matrix indexed by date, with one column per asset.

• params (MapperParams) – Hyperparameters used by the Mapper pipeline.

Returns:

Nested cluster structure. The outer list contains connected components of the Mapper graph. Each component contains Mapper nodes, and each node contains the asset identifiers assigned to that node. Returns None when there are not enough assets or observations to build Mapper.

Return type:

list of list of list of str or None

Notes

Missing prices are forward-filled inside the window, but backward filling is not used. This avoids introducing future information into the construction.

tda_finance.tda.mapper_clustering.cap_weights_strict(w: Dict[str, float], cap: float) → Dict[str, float]

Apply a strict maximum weight cap with iterative redistribution.

Parameters:

• w (dict of str to float) – Raw asset weights.

• cap (float) – Maximum allowed weight per asset.

Returns:

Normalized asset weights satisfying the weight cap when feasible.

Return type:

dict of str to float

tda_finance.tda.mapper_clustering.weight_distribution(clustered_symbols: List[List[List[str]]], max_weight: float = 0.1, method: str = 'node_overlap') → Dict[str, float]

Convert Mapper clusters into portfolio weights.

Parameters:

• clustered_symbols (list of list of list of str) – Nested cluster structure returned by build_clusters_from_prices.

• max_weight (float, default=0.1) – Maximum allowed weight per asset.

• method ({"node_overlap", "macro_cluster"}, default="node_overlap") – Weighting method used to transform Mapper clusters into asset weights.

Returns:

Portfolio weights indexed by asset identifier.

Return type:

dict of str to float

Raises:

ValueError – If method is not supported.

tda_finance.tda.mapper_clustering.weight_distribution_macro_cluster(clustered_symbols: List[List[List[str]]], max_weight: float = 0.1) → Dict[str, float]

Convert Mapper connected components into portfolio weights.

Each connected component of the Mapper graph receives the same total weight. Inside each component, the weight is distributed equally among the unique assets contained in that component.

Parameters:

• clustered_symbols (list of list of list of str) – Nested cluster structure returned by build_clusters_from_prices.

• max_weight (float, default=0.1) – Maximum allowed weight per asset.

Returns:

Portfolio weights indexed by asset identifier.

Return type:

dict of str to float

tda_finance.tda.mapper_clustering.weight_distribution_node_overlap(clustered_symbols: List[List[List[str]]], max_weight: float = 0.1) → Dict[str, float]

Convert Mapper nodes into portfolio weights using node overlap.

Each Mapper node receives the same total weight. The weight of a node is distributed equally among the assets contained in that node. Assets that appear in several overlapping nodes accumulate weight.

Parameters:

• clustered_symbols (list of list of list of str) – Nested cluster structure returned by build_clusters_from_prices.

• max_weight (float, default=0.1) – Maximum allowed weight per asset.

Returns:

Portfolio weights indexed by asset identifier.

Return type:

dict of str to float

Persistence diagrams

Persistent-homology utilities for financial return windows.

This module contains functions to clean return windows, build correlation-based distance matrices and compute persistence diagrams using Vietoris–Rips persistent homology.

class tda_finance.tda.persistence_diagrams.PHParams(maxdim: int = 1, corr_method: Literal['pearson', 'kendall', 'spearman'] = 'pearson', dist_variant: Literal['sqrt', 'linear'] = 'sqrt', winsor_q: float | None = 0.01, min_non_nan_frac: float = 0.98, min_std: float = 1e-08)

Bases: object

Hyperparameters for persistent-homology computation.

Parameters:

• maxdim (int, default=1) – Maximum homology dimension computed by ripser. With maxdim=1, both H0 and H1 diagrams are returned.

• corr_method ({"pearson", "kendall", "spearman"}, default="pearson") – Correlation method used to build the asset dependence matrix.

• dist_variant ({"sqrt", "linear"}, default="sqrt") – Transformation from correlation to distance. "sqrt" uses sqrt(2 * (1 - rho)) and "linear" uses 1 - rho.

• winsor_q (float, optional, default=0.01) – Quantile used for winsorization by asset. If None, winsorization is disabled.

• min_non_nan_frac (float, default=0.98) – Minimum fraction of non-missing observations required to keep an asset.

• min_std (float, default=1e-8) – Minimum standard deviation required to keep an asset.

corr_method: Literal['pearson', 'kendall', 'spearman'] = 'pearson'

dist_variant: Literal['sqrt', 'linear'] = 'sqrt'

maxdim: int = 1

min_non_nan_frac: float = 0.98

min_std: float = 1e-08

winsor_q: float | None = 0.01

tda_finance.tda.persistence_diagrams.clean_returns_window(returns_window: DataFrame, params: PHParams) → DataFrame

Clean a return window before computing persistent homology.

The cleaning step removes assets with too many missing observations, optionally winsorizes extreme returns, fills remaining missing values with zero and removes almost-constant assets.

Parameters:

• returns_window (pandas.DataFrame) – Return matrix indexed by date, with one column per asset.

• params (PHParams) – Parameters controlling the cleaning process.

Returns:

Cleaned return matrix.

Return type:

pandas.DataFrame

tda_finance.tda.persistence_diagrams.compute_persistence_diagrams_from_returns(returns_window: DataFrame, params: PHParams) → Dict[str, Any]

Compute persistence diagrams from a return window.

The function first builds a correlation-based distance matrix between assets and then computes Vietoris–Rips persistence diagrams using ripser.

Parameters:

• returns_window (pandas.DataFrame) – Return matrix indexed by date, with one column per asset.

• params (PHParams) – Parameters used to clean the window and compute persistence diagrams.

Returns:

Dictionary with two entries:

"dgms"

List of persistence diagrams returned by ripser.

"symbols"

Asset identifiers used to build the distance matrix.

Return type:

dict

Raises:

ImportError – If ripser is not installed.

tda_finance.tda.persistence_diagrams.corr_distance_matrix(returns_window: DataFrame, params: PHParams) → Tuple[ndarray, List[str]]

Compute a correlation-based distance matrix.

Parameters:

• returns_window (pandas.DataFrame) – Return matrix indexed by date, with one column per asset.

• params (PHParams) – Parameters controlling cleaning, correlation and distance conversion.

Returns:

Distance matrix and list of asset identifiers associated with the rows and columns of the matrix.

Return type:

tuple of numpy.ndarray and list of str

Raises:

ValueError – If params.dist_variant is not supported.

Persistence features

Feature extraction from persistence diagrams.

This module provides simple summary features from persistence diagrams. These features are mainly used as diagnostics for the persistent-homology regime control.

tda_finance.tda.persistence_features.ph_summary_features(dgms: List[ndarray]) → Dict[str, float]

Compute summary features from persistence diagrams.

The returned features are intentionally simple and interpretable. They are used to describe the persistence diagrams obtained from each return window, especially the H1 diagram.

Parameters:

dgms (list of numpy.ndarray) – Persistence diagrams. The first element corresponds to H0 and the second element, when available, corresponds to H1.

Returns:

Dictionary with counts, total persistence, maximum persistence and selected quantiles of H1 lifetimes.

Return type:

dict of str to float

Regime detection

Regime-detection utilities based on persistence landscapes.

This module provides a simple topological anomaly detector. It summarizes each persistence diagram through the L2 norm of a persistence landscape and compares that value with a rolling history of previous norms.

class tda_finance.tda.regime_detection.TopologicalAnomalyDetector(history_len: int = 12, danger_quantile: float = 0.9, min_history: int = 10)

Bases: object

Rolling detector for topological anomalies.

The detector stores a rolling history of persistence-landscape norms. Once enough history is available, the current norm is compared with a historical quantile. Values above that threshold are treated as anomalous.

Parameters:

• history_len (int, default=12) – Maximum number of past norms stored by the detector.

• danger_quantile (float, default=0.90) – Quantile used as anomaly threshold.

• min_history (int, default=10) – Minimum number of past observations required before the detector starts flagging anomalies.

is_market_safe(current_norm: float) → bool

Return whether the current regime is considered safe.

Parameters:

current_norm (float) – Current persistence-landscape norm.

Returns:

True if the current regime is considered safe, and False if the current norm is anomalously high relative to its recent history.

Return type:

bool

tda_finance.tda.regime_detection.compute_landscape_norm(dgms: List[ndarray], dimension: int = 1, resolution: int = 100, num_landscapes: int = 5) → float

Compute the L2 norm of a persistence landscape.

Parameters:

• dgms (list of numpy.ndarray) – Persistence diagrams indexed by homology dimension.

• dimension (int, default=1) – Homology dimension used to compute the landscape. The default value corresponds to H1.

• resolution (int, default=100) – Number of grid points used to discretize the landscape.

• num_landscapes (int, default=5) – Number of landscape layers computed by GUDHI.

Returns:

L2 norm of the selected persistence landscape. If the requested diagram is missing or empty, the function returns 0.0.

Return type:

float