cross_signal

cross_signal

CrossSignalAnalytics

CrossSignalAnalytics(
    dataframe: DataFrame, time_column: str = "systime"
)

Cross-Signal Analytics for multi-signal timeseries.

Computes analytical metrics across pairs or groups of signals. Expects wide-format DataFrames (one column per signal, datetime index).

Methods: - granger_causality: Test if one signal Granger-causes another. - transfer_entropy: Estimate information transfer between signals. - pairwise_transfer_entropy: Transfer entropy for all directed pairs. - synchronization_index: Phase or amplitude synchronization between signals. - pairwise_synchronization: Synchronization for all pairs. - lead_lag: Detect lead-lag relationships via cross-correlation. - lead_lag_matrix: Lead-lag for all pairs.

granger_causality

granger_causality(
    cause: str,
    effect: str,
    max_lag: int = 10,
    significance: float = 0.05,
) -> dict

Test whether cause Granger-causes effect using F-test.

Parameters:

Name	Type	Description	Default
cause	str	Column name of the potential cause signal.	required
effect	str	Column name of the potential effect signal.	required
max_lag	int	Maximum number of lags to test.	10
significance	float	P-value threshold for declaring causality.	0.05

Returns:

Type	Description
dict	Dict with optimal_lag, f_statistic, p_value, is_causal, results_by_lag.

transfer_entropy

transfer_entropy(
    source: str, target: str, lag: int = 1, bins: int = 10
) -> float

Estimate transfer entropy from source to target.

TE(X->Y) = H(Y_future | Y_past) - H(Y_future | Y_past, X_past)

Parameters:

Name	Type	Description	Default
source	str	Column name of source signal.	required
target	str	Column name of target signal.	required
lag	int	Time lag for past values.	1
bins	int	Number of bins for discretization.	10

Returns:

Type	Description
float	Transfer entropy in bits.

pairwise_transfer_entropy

pairwise_transfer_entropy(
    lag: int = 1, bins: int = 10
) -> pd.DataFrame

Compute transfer entropy for all directed signal pairs.

Returns:

Type	Description
DataFrame	Square DataFrame where result.loc[A, B] is TE(A -> B).

synchronization_index

synchronization_index(
    signal_a: str, signal_b: str, method: str = "phase"
) -> float

Measure synchronization between two signals.

Parameters:

Name	Type	Description	Default
signal_a	str	First signal column name.	required
signal_b	str	Second signal column name.	required
method	str	'phase' for phase-locking value, 'amplitude' for envelope correlation.	'phase'

Returns:

Type	Description
float	Float between 0 (no sync) and 1 (perfect sync).

pairwise_synchronization

pairwise_synchronization(
    method: str = "phase",
) -> pd.DataFrame

Compute synchronization index for all signal pairs.

Returns:

Type	Description
DataFrame	Symmetric DataFrame (like a correlation matrix).

lead_lag

lead_lag(
    signal_a: str,
    signal_b: str,
    max_lag: int = 50,
    significance: float = 0.05,
) -> dict

Detect lead-lag relationship via cross-correlation.

Parameters:

Name	Type	Description	Default
signal_a	str	First signal column name.	required
signal_b	str	Second signal column name.	required
max_lag	int	Maximum lag to test.	50
significance	float	P-value threshold.	0.05

Returns:

Type	Description
dict	Dict with optimal_lag, correlation_at_lag, leader, follower, p_value, is_significant.

lead_lag_matrix

lead_lag_matrix(max_lag: int = 50) -> pd.DataFrame

Compute lead-lag for all signal pairs.

Returns:

Type	Description
DataFrame	DataFrame where result.loc[A, B] is the optimal lag (positive = A leads B).