ccm

ccm

Functions:

Name	Description
make_sample_func	Create a sample function with its own independent RNG.
bootstrap	Perform Convergent Cross Mapping and return per-sample correlations.
ccm	Perform Convergent Cross Mapping using a custom prediction function.
pearson_correlation	Compute vectorized Pearson correlation between X and Y.
with_simplex_projection	Perform Convergent Cross Mapping using simplex projection.
with_smap	Perform Convergent Cross Mapping using S-Map (local linear regression).

Attributes:

Name	Type	Description
SampleFunc	TypeAlias	SampleFunc is a function that takes (pool, size) and returns a sampled array.
AggregateFunc	TypeAlias	AggregateFunc is a function that takes an array of values and returns a single value.

SampleFunc

SampleFunc: TypeAlias = Callable[[np.ndarray, int], np.ndarray]

SampleFunc is a function that takes (pool, size) and returns a sampled array.

AggregateFunc

AggregateFunc: TypeAlias = Callable[[np.ndarray], float]

AggregateFunc is a function that takes an array of values and returns a single value.

make_sample_func

make_sample_func(seed: int | None = 42) -> SampleFunc

Create a sample function with its own independent RNG.

bootstrap

bootstrap(X: np.ndarray, Y: np.ndarray, lib_sizes: np.ndarray, predict_func: PredictFunc, n_samples: int = 20, *, library_pool: np.ndarray, prediction_pool: np.ndarray, sample_func: SampleFunc | None = None, batch_size: int | None = 10) -> np.ndarray

Perform Convergent Cross Mapping and return per-sample correlations.

Same as :func:ccm but returns the raw per-sample scores instead of aggregating them.

Parameters:

Name	Type	Description	Default
X	ndarray	Library time series (potential response)	required
Y	ndarray	Target time series (potential driver)	required
lib_sizes	ndarray	Array of library sizes to test convergence.	required
predict_func	PredictFunc	Prediction function with signature (X, Y, Q) -> predictions.	required
n_samples	int	Number of random samples per library size for bootstrapping.	20
library_pool	ndarray	1-D array of integer indices from which library members are sampled.	required
prediction_pool	ndarray	1-D array of integer indices that are predicted.	required
sample_func	SampleFunc or None	Function responsible for drawing a library sample of a given size. When None, a fresh RNG-backed sampler is created per call.	None
batch_size	int or None	If specified, predictions are made in batches to limit memory usage.	10

Returns:

Name	Type	Description
samples	ndarray	Per-sample correlation coefficients of shape (n_samples, len(lib_sizes)).

ccm

ccm(X: np.ndarray, Y: np.ndarray, lib_sizes: np.ndarray, predict_func: PredictFunc, n_samples: int = 20, *, library_pool: np.ndarray, prediction_pool: np.ndarray, sample_func: SampleFunc | None = None, aggregate_func: AggregateFunc = np.mean, batch_size: int | None = 10) -> np.ndarray

Perform Convergent Cross Mapping using a custom prediction function.

CCM tests for causality from X to Y by using the attractor reconstructed from Y to predict values of X. If X causes Y, then Y’s attractor contains information about X, allowing cross-mapping from Y to X.

Parameters:

Name	Type	Description	Default
X	ndarray	Library time series (potential response)	required
Y	ndarray	Target time series (potential driver)	required
lib_sizes	ndarray	Array of library sizes to test convergence.	required
predict_func	PredictFunc	Prediction function with signature (X, Y, Q) -> predictions. Can be simplex_projection, smap with partial application, or a custom function.	required
n_samples	int	Number of random samples per library size for bootstrapping.	100
library_pool	ndarray	1-D array of integer indices from which library members are sampled.	required
prediction_pool	ndarray	1-D array of integer indices that are predicted.	required
sample_func	SampleFunc or None	Function responsible for drawing a library sample of a given size. It receives (pool, size) and returns an array of indices. When None, a fresh RNG-backed sampler is created per call.	None
aggregate_func	AggregateFunc	Reducer applied to the correlation samples for each library size.	np.mean
batch_size	int or None	If not specified, batch_size == n_samples. If specified, predictions are made in batches to limit memory usage.	None

Returns:

Name	Type	Description
correlations	ndarray	Mean correlation coefficient for each library size.

Raises:

Type	Description
ValueError	- If X and Y have different lengths. - If lib_sizes contains non-positive values. - If predict_func is not callable. - If n_samples is not positive. - If aggregate_func is not callable. - If library_pool or prediction_pool is invalid.

Notes

• Higher correlation at larger library sizes indicates convergence and suggests X influences Y (X -> Y causality)
• Convergence is the key signature of causality in CCM
• The method uses Y’s attractor to predict X (cross-mapping)

Examples:

from functools import partial

import numpy as np

from edmkit.ccm import ccm, simplex_projection, smap
from edmkit.embedding import lagged_embed

# Generate coupled logistic maps (X drives Y)
N = 1000
rx, ry, Bxy = 3.8, 3.5, 0.02
X = np.zeros(N)
Y = np.zeros(N)
X[0], Y[0] = 0.4, 0.2
for i in range(1, N):
    X[i] = X[i - 1] * (rx - rx * X[i - 1])
    Y[i] = Y[i - 1] * (ry - ry * Y[i - 1]) + Bxy * X[i - 1]

tau = 1
E = 2

# To test X -> Y causality, cross-map from Y's attractor to X
Y_embedding = lagged_embed(Y, tau=tau, e=E)
shift = tau * (E - 1)
X_aligned = X[shift:]

library_pool = np.arange(Y_embedding.shape[0] // 2)
prediction_pool = np.arange(Y_embedding.shape[0] // 2, Y_embedding.shape[0])

# logarithmic within range 10 to max library size
lib_sizes = np.logspace(np.log10(10), np.log10(library_pool[-1]), num=5, dtype=int)

# Using simplex projection
correlations = ccm(
    Y_embedding,
    X_aligned,
    lib_sizes=lib_sizes,
    predict_func=simplex_projection,
    library_pool=library_pool,
    prediction_pool=prediction_pool,
)

# Using S-Map with partial application
correlations = ccm(
    Y_embedding,
    X_aligned,
    lib_sizes=lib_sizes,
    predict_func=partial(smap, theta=2.0, alpha=1e-10),
    library_pool=library_pool,
    prediction_pool=prediction_pool,
)

pearson_correlation

pearson_correlation(X: np.ndarray, Y: np.ndarray) -> np.ndarray

Compute vectorized Pearson correlation between X and Y.

Parameters:

Name	Type	Description	Default
X	ndarray	1D or 2D array of shape (L,) or (B, L)	required
Y	ndarray	1D or 2D array of shape (L,) or (B, L)	required

Returns:

Name	Type	Description
correlation	ndarray	Pearson correlation coefficient(s) between X and Y. Shape (B,) if inputs are 2D, else scalar.

with_simplex_projection

with_simplex_projection(X: np.ndarray, Y: np.ndarray, lib_sizes: np.ndarray, n_samples: int = 100, use_tensor: bool = False, *, library_pool: np.ndarray, prediction_pool: np.ndarray, sample_func: SampleFunc | None = None, aggregate_func: AggregateFunc = np.mean) -> np.ndarray

Perform Convergent Cross Mapping using simplex projection.

This is a convenience wrapper around the general ccm function using simplex_projection as the prediction method.

Parameters:

Name	Type	Description	Default
X	ndarray	Library time series (potential response)	required
Y	ndarray	Target time series (potential driver)	required
lib_sizes	ndarray	Array of library sizes to test convergence	required
n_samples	int	Number of random samples per library size for bootstrapping	100
use_tensor	bool	Whether to use tinygrad tensors for computation	False
library_pool	ndarray	Indices that can be used to draw library samples. Defaults to the full range.	required
prediction_pool	ndarray	Indices that should be predicted (leave-one-out over this set). Defaults to the full range.	required
sample_func	callable	Function responsible for drawing a library sample of a given size. When omitted, a fresh RNG-backed sampler is created per call.	None
aggregate_func	callable	Reducer applied to the correlation samples for each library size. Falls back to np.mean when omitted.	mean

Returns:

Name	Type	Description
correlations	ndarray	Mean correlation coefficient for each library size

Raises:

Type	Description
ValueError	- If the underlying ccm call detects invalid arguments

Examples:

import numpy as np

from edmkit import ccm
from edmkit.embedding import lagged_embed

# Generate coupled logistic maps (X drives Y)
N = 1000
rx, ry, Bxy = 3.8, 3.5, 0.02
X = np.zeros(N)
Y = np.zeros(N)
X[0], Y[0] = 0.4, 0.2
for i in range(1, N):
    X[i] = X[i - 1] * (rx - rx * X[i - 1])
    Y[i] = Y[i - 1] * (ry - ry * Y[i - 1]) + Bxy * X[i - 1]

tau = 1
E = 2

# To test X -> Y causality, cross-map from Y's attractor to X
Y_embedding = lagged_embed(Y, tau=tau, e=E)
shift = tau * (E - 1)
X_aligned = X[shift:]

library_pool = np.arange(Y_embedding.shape[0] // 2)
prediction_pool = np.arange(Y_embedding.shape[0] // 2, Y_embedding.shape[0])

# logarithmic within range 10 to max library size
lib_sizes = np.logspace(np.log10(10), np.log10(library_pool[-1]), num=5, dtype=int)

correlations = ccm.with_simplex_projection(
    Y_embedding,
    X_aligned,
    lib_sizes=lib_sizes,
    library_pool=library_pool,
    prediction_pool=prediction_pool,
)

with_smap

with_smap(X: np.ndarray, Y: np.ndarray, lib_sizes: np.ndarray, theta: float, alpha: float = 1e-10, n_samples: int = 100, use_tensor: bool = False, *, library_pool: np.ndarray, prediction_pool: np.ndarray, sample_func: SampleFunc | None = None, aggregate_func: AggregateFunc = np.mean) -> np.ndarray

Perform Convergent Cross Mapping using S-Map (local linear regression).

This is a convenience wrapper around the general ccm function using smap as the prediction method.

Parameters:

Name	Type	Description	Default
X	ndarray	Library time series (potential response)	required
Y	ndarray	Target time series (potential driver)	required
lib_sizes	ndarray	Array of library sizes to test convergence	required
theta	float	Nonlinearity parameter for S-Map	required
alpha	float	Regularization parameter for S-Map	1e-10
n_samples	int	Number of random samples per library size for bootstrapping	100
use_tensor	bool	Whether to use tinygrad tensors for computation	False
library_pool	ndarray	Indices that can be used to draw library samples. Defaults to the full range.	required
prediction_pool	ndarray	Indices that should be predicted (leave-one-out over this set). Defaults to the full range.	required
sample_func	callable	Function responsible for drawing a library sample of a given size. When omitted, a fresh RNG-backed sampler is created per call.	None
aggregate_func	callable	Reducer applied to the correlation samples for each library size. Falls back to np.mean when omitted.	mean

Returns:

Name	Type	Description
correlations	ndarray	Mean correlation coefficient for each library size

Raises:

Type	Description
ValueError	- If the underlying ccm call detects invalid arguments

Examples:

import numpy as np

from edmkit import ccm
from edmkit.embedding import lagged_embed

# Generate coupled logistic maps (X drives Y)
N = 1000
rx, ry, Bxy = 3.8, 3.5, 0.02
X = np.zeros(N)
Y = np.zeros(N)
X[0], Y[0] = 0.4, 0.2
for i in range(1, N):
    X[i] = X[i - 1] * (rx - rx * X[i - 1])
    Y[i] = Y[i - 1] * (ry - ry * Y[i - 1]) + Bxy * X[i - 1]

tau = 1
E = 2

# To test X -> Y causality, cross-map from Y's attractor to X
Y_embedding = lagged_embed(Y, tau=tau, e=E)
shift = tau * (E - 1)
X_aligned = X[shift:]

library_pool = np.arange(Y_embedding.shape[0] // 2)
prediction_pool = np.arange(Y_embedding.shape[0] // 2, Y_embedding.shape[0])

# logarithmic within range 10 to max library size
lib_sizes = np.logspace(np.log10(10), np.log10(library_pool[-1]), num=5, dtype=int)

correlations = ccm.with_smap(
    Y_embedding,
    X_aligned,
    lib_sizes=lib_sizes,
    theta=2.0,
    library_pool=library_pool,
    prediction_pool=prediction_pool,
)