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.

Keep a time-ordered copy of the wide frame.

Parameters:

Name	Type	Description	Default
`dataframe`	`DataFrame`	Wide frame, one numeric column per signal. NumPy, nullable (`Float64`, `Int64`) and pyarrow numeric columns are signals; they are read as `float64`.	required
`time_column`	`str`	Time column; when present, rows are sorted by it and it becomes the index.	`'systime'`

Raises:

Type	Description
`ValueError`	If fewer than 2 numeric signal columns are found.

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. Only rows where both signals have a value are used. With too few of them to fit a lag, `f_statistic` and `p_value` are NaN and `is_causal` is False.

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; NaN when there are no more aligned samples (rows where both signals have a value) than `lag`.

pairwise_transfer_entropy ¤

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

Compute transfer entropy for all directed signal pairs.

Parameters:

Name	Type	Description	Default
`lag`	`int`	Number of samples between the past values and the future value of the target (see `transfer_entropy`). Default 1.	`1`
`bins`	`int`	Number of equal-width bins used to discretize each signal. Default 10.	`10`

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); NaN when no row has a value for both signals.

pairwise_synchronization ¤

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

Compute synchronization index for all signal pairs.

Parameters:

Name	Type	Description	Default
`method`	`str`	`'phase'` for the phase-locking value or `'amplitude'` for envelope correlation (see `synchronization_index`). Default `'phase'`.	`'phase'`

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. optimal_lag is the number of samples by which signal_a leads signal_b: positive when signal_b repeats signal_a later (leader is signal_a), negative when signal_b leads. Only rows where both signals have a value are used; with fewer than 2 such rows the numbers are NaN and leader / follower are None.

lead_lag_matrix ¤

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

Compute lead-lag for all signal pairs.

Parameters:

Name	Type	Description	Default
`max_lag`	`int`	Largest lag, in samples, tested in either direction. Default 50.	`50`

Returns:

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