API Reference

Welcome to the comprehensive API reference for TranCIT: Transient Causal Interaction Toolbox. This section provides detailed documentation for all classes, functions, and modules in the package.

The TranCIT package is organized into several core components:

Core Classes: Base analyzers, results, and configuration objects
Causality Analysis: Dynamic Causal Strength (DCS), relative Dynamic Causal Strength (rDCS), Transfer Entropy (TE), Granger Causality (GC) calculators
Pipeline System: Event-based analysis orchestration and stages
Model Estimation: VAR models, BIC selection, and validation
Simulation: Synthetic data generation for testing and validation
Utilities: Data preprocessing, visualization, and helper functions

Core Module (trancit.core)

The core module provides the foundational classes and interfaces used throughout the TranCIT package.

Base Classes 

class trancit.core.base.BaseAnalyzer(**kwargs)[source]

Bases: ABC

Abstract base class for all analyzers.

This class defines the common interface and functionality for all analysis components in the TranCIT package.

Initialize analyzer with configuration.

The abstract base class for all analysis components. All calculators and analyzers inherit from this class, providing a consistent interface.

Key Methods:

analyze(data, **kwargs): Main analysis method (must be implemented by subclasses)
validate_config(config): Validates configuration parameters
_log_analysis_start(), _log_analysis_complete(): Logging support

Example:

from trancit.core.base import BaseAnalyzer, BaseResult
import numpy as np

# Define a custom result class
class MyResult(BaseResult):
    """Custom result class for demonstration."""
    pass

# Define a custom analyzer
class MyAnalyzer(BaseAnalyzer):
    def analyze(self, data):
        # Your analysis implementation here
        return MyResult(result=data.sum())

# Generate sample data
data = np.random.randn(2, 100, 10)  # 2 vars, 100 time points, 10 trials

# Use the analyzer
analyzer = MyAnalyzer()
result = analyzer.analyze(data)
print(f"Analysis result: {result.result}")

__init__(**kwargs)[source]: Initialize analyzer with configuration.

abstractmethod analyze(data, **kwargs)[source]

Perform analysis on the given data.

Parameters:

data (np.ndarray) – Input data for analysis
**kwargs – Additional parameters specific to the analyzer

Returns:

Analysis results

Return type:

BaseResult

class trancit.core.base.BaseResult(**kwargs)[source]

Bases: object

Base class for all result objects.

Initialize result with keyword arguments.

Base class for all result objects. Provides common functionality for storing and accessing analysis results.

Key Methods:

to_dict(): Convert result to dictionary format
__getitem__(key): Dictionary-like access to results

Example:

from trancit.core.base import BaseResult
import numpy as np

# Define a custom result class
class MyResult(BaseResult):
    """Custom result class for demonstration."""
    pass

# Generate sample data arrays
n_time = 100
cs_array = np.random.randn(n_time, 2)  # Causal strength array
te_array = np.random.randn(n_time, 2)  # Transfer entropy array

# Create result object
result = MyResult(causal_strength=cs_array, transfer_entropy=te_array)
print(result.causal_strength.shape)
result_dict = result.to_dict()

__init__(**kwargs)[source]: Initialize result with keyword arguments.

__repr__()[source]

String representation of the result.

Return type:: str

to_dict()[source]

Convert result to dictionary.

Return type:: Dict[str, Any]

class trancit.core.base.BaseConfig(**kwargs)[source]

Bases: object

Base configuration class for all DCS components.

Initialize configuration with keyword arguments.

Base class for configuration objects. Provides parameter validation and default value handling.

Example:

from trancit.core.base import BaseConfig

# Define a custom configuration class
class MyConfig(BaseConfig):
    """Custom configuration class for demonstration."""
    pass

# Create configuration object
config = MyConfig(param1=10, param2="value", param3=True)

# Access parameters
print(f"param1: {config.param1}")
print(f"param2: {config.param2}")

# Convert to dictionary
config_dict = config.to_dict()
print(f"Config as dict: {config_dict}")

__init__(**kwargs)[source]: Initialize configuration with keyword arguments.

__repr__()[source]

String representation of the configuration.

Return type:: str

to_dict()[source]

Convert configuration to dictionary.

Return type:: Dict[str, Any]

validate()[source]

Validate configuration parameters. Override in subclasses.

Return type:: None

Exception Classes 

class trancit.core.exceptions.DCSError(message, details=None)[source]

Bases: Exception

Base exception for all DCS-related errors.

Initialize DCS error.

Parameters:

message (str) – Error message
details (str, optional) – Additional error details

Base exception class for all DCS-related errors.

__init__(message, details=None)[source]

Initialize DCS error.

Parameters:

message (str) – Error message
details (str, optional) – Additional error details

class trancit.core.exceptions.ValidationError(message, field=None, value=None)[source]

Bases: DCSError

Exception raised when input validation fails.

Initialize validation error.

Parameters:

message (str) – Validation error message
field (str, optional) – Name of the field that failed validation
value (any, optional) – Value that failed validation

Raised when input data or parameters fail validation checks.

Common Causes:

Wrong data dimensions (should be 3D: variables × time × trials)
Non-bivariate data (DCS requires exactly 2 variables)
Model order too large for available data
Invalid parameter values

__init__(message, field=None, value=None)[source]

Initialize validation error.

Parameters:

message (str) – Validation error message
field (str, optional) – Name of the field that failed validation
value (any, optional) – Value that failed validation

class trancit.core.exceptions.ComputationError(message, step=None, data_shape=None)[source]

Bases: DCSError

Exception raised when computation fails.

Initialize computation error.

Parameters:

message (str) – Computation error message
step (str, optional) – Step in the computation that failed
data_shape (tuple, optional) – Shape of the data being processed

Raised when numerical computation fails.

Common Causes:

Singular or near-singular matrices
Numerical instability
Insufficient data for reliable estimation

__init__(message, step=None, data_shape=None)[source]

Initialize computation error.

Parameters:

message (str) – Computation error message
step (str, optional) – Step in the computation that failed
data_shape (tuple, optional) – Shape of the data being processed

class trancit.core.exceptions.ConfigurationError(message, config_key=None, expected_type=None)[source]

Bases: DCSError

Exception raised when configuration is invalid.

Initialize configuration error.

Parameters:

message (str) – Configuration error message
config_key (str, optional) – Configuration key that caused the error
expected_type (str, optional) – Expected type for the configuration value

Raised when configuration parameters are invalid or inconsistent.

__init__(message, config_key=None, expected_type=None)[source]

Initialize configuration error.

Parameters:

message (str) – Configuration error message
config_key (str, optional) – Configuration key that caused the error
expected_type (str, optional) – Expected type for the configuration value

Causality Analysis (trancit.causality)

The causality module provides implementations of various causality measures.

Dynamic Causal Strength (DCS)

class trancit.causality.dcs.DCSCalculator(model_order, time_mode='inhomo', use_diagonal_covariance=False, **kwargs)[source]

Bases: BaseAnalyzer

Dynamic Causal Strength (DCS) calculator.

This class implements the DCS algorithm for quantifying causal relationships in time series data.

Initialize DCS calculator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
**kwargs – Additional configuration parameters

Main class for Dynamic Causal Strength analysis. Computes time-varying causal relationships using VAR models.

Parameters:

model_order (int): Number of time lags to include (typically 2-8)
time_mode (str): “inhomo” (time-varying) or “homo” (time-invariant)
use_diagonal_covariance (bool): Use diagonal approximation for numerical stability

Key Methods:

analyze(data): Perform DCS analysis on time series data
_validate_input_data(data): Check data format and quality
_prepare_data_structures(data): Prepare data for VAR analysis

Example:

from trancit import DCSCalculator
import numpy as np

# Generate sample data
data = np.random.randn(2, 500, 20)  # 2 vars, 500 time points, 20 trials

# Create calculator
calculator = DCSCalculator(model_order=4, time_mode="inhomo")

# Perform analysis
result = calculator.analyze(data)

print(f"Causal strength shape: {result.causal_strength.shape}")
print(f"Mean X→Y causality: {result.causal_strength[:, 1].mean():.4f}")

__init__(model_order, time_mode='inhomo', use_diagonal_covariance=False, **kwargs)[source]

Initialize DCS calculator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
**kwargs – Additional configuration parameters

analyze(data, **kwargs)[source]

Perform DCS analysis on the given data.

Parameters:

data (np.ndarray) – Time series data with shape (n_vars, n_observations, n_trials)
**kwargs – Additional parameters

Returns:

DCS analysis results

Return type:

DCSResult

class trancit.causality.dcs.DCSResult(causal_strength, transfer_entropy, granger_causality, coefficients, te_residual_cov)[source]

Bases: BaseResult

Result container for DCS analysis.

Initialize DCS result.

Parameters:

causal_strength (np.ndarray) – Dynamic Causal Strength values
transfer_entropy (np.ndarray) – Transfer Entropy values
granger_causality (np.ndarray) – Granger Causality values
coefficients (np.ndarray) – VAR coefficients
te_residual_cov (np.ndarray) – Transfer entropy residual covariance

Result object containing DCS analysis outputs.

Attributes:

causal_strength (ndarray): Dynamic causal strength values, shape (time, 2)
transfer_entropy (ndarray): Transfer entropy values, shape (time, 2)
granger_causality (ndarray): Granger causality values, shape (time, 2)
coefficients (ndarray): VAR model coefficients
te_residual_cov (ndarray): Residual covariance matrices

Column Convention:

Column 0: Y → X (variable 2 influences variable 1)
Column 1: X → Y (variable 1 influences variable 2)

__init__(causal_strength, transfer_entropy, granger_causality, coefficients, te_residual_cov)[source]

Initialize DCS result.

Parameters:

causal_strength (np.ndarray) – Dynamic Causal Strength values
transfer_entropy (np.ndarray) – Transfer Entropy values
granger_causality (np.ndarray) – Granger Causality values
coefficients (np.ndarray) – VAR coefficients
te_residual_cov (np.ndarray) – Transfer entropy residual covariance

Transfer Entropy 

class trancit.causality.transfer_entropy.TransferEntropyCalculator(model_order, time_mode='inhomo', use_diagonal_covariance=False, **kwargs)[source]

Bases: BaseAnalyzer

Transfer Entropy (TE) calculator.

This class implements the Transfer Entropy algorithm for quantifying information flow between time series variables.

Initialize Transfer Entropy calculator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
**kwargs – Additional configuration parameters

Calculator for Transfer Entropy, an information-theoretic measure of causality.

About Transfer Entropy:

Transfer Entropy (TE) quantifies the information flow between time series. It measures how much knowing the past of one variable reduces uncertainty about the future of another variable.

Example:

from trancit.causality import TransferEntropyCalculator
import numpy as np

# Generate sample data
data = np.random.randn(2, 500, 20)  # 2 vars, 500 time points, 20 trials

calculator = TransferEntropyCalculator(model_order=3)
result = calculator.analyze(data)

print(f"TE X→Y: {result.transfer_entropy[:, 1].mean():.4f}")
print(f"TE Y→X: {result.transfer_entropy[:, 0].mean():.4f}")

__init__(model_order, time_mode='inhomo', use_diagonal_covariance=False, **kwargs)[source]

Initialize Transfer Entropy calculator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
**kwargs – Additional configuration parameters

analyze(data, **kwargs)[source]

Perform Transfer Entropy analysis on the given data.

Parameters:

data (np.ndarray) – Time series data with shape (n_vars, n_observations, n_trials)
**kwargs – Additional parameters

Returns:

Transfer Entropy analysis results

Return type:

TransferEntropyResult

class trancit.causality.transfer_entropy.TransferEntropyResult(transfer_entropy, te_residual_cov, coefficients)[source]

Bases: BaseResult

Result container for Transfer Entropy analysis.

Initialize Transfer Entropy result.

Parameters:

transfer_entropy (np.ndarray) – Transfer Entropy values
te_residual_cov (np.ndarray) – Transfer entropy residual covariance
coefficients (np.ndarray) – VAR coefficients

__init__(transfer_entropy, te_residual_cov, coefficients)[source]

Initialize Transfer Entropy result.

Parameters:

transfer_entropy (np.ndarray) – Transfer Entropy values
te_residual_cov (np.ndarray) – Transfer entropy residual covariance
coefficients (np.ndarray) – VAR coefficients

Granger Causality 

class trancit.causality.granger.GrangerCausalityCalculator(model_order, time_mode='inhomo', use_diagonal_covariance=False, **kwargs)[source]

Bases: BaseAnalyzer

Granger Causality (GC) calculator.

This class implements the Granger Causality algorithm for detecting linear causal relationships in time series data.

Initialize Granger Causality calculator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
**kwargs – Additional configuration parameters

Calculator for Granger Causality, a linear measure of predictive causality.

About Granger Causality:

Granger Causality tests whether past values of one time series help predict future values of another, beyond what can be predicted from the target series alone.

Example:

from trancit.causality import GrangerCausalityCalculator
import numpy as np

# Generate sample data
data = np.random.randn(2, 500, 20)  # 2 vars, 500 time points, 20 trials

calculator = GrangerCausalityCalculator(model_order=4)
result = calculator.analyze(data)

# Check significance
significant_xy = result.pvalues[:, 1] < 0.05
print(f"Significant X→Y causality: {significant_xy.sum()} time points")

__init__(model_order, time_mode='inhomo', use_diagonal_covariance=False, **kwargs)[source]

Initialize Granger Causality calculator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
**kwargs – Additional configuration parameters

analyze(data, **kwargs)[source]

Perform Granger Causality analysis on the given data.

Parameters:

data (np.ndarray) – Time series data with shape (n_vars, n_observations, n_trials)
**kwargs – Additional parameters

Returns:

Granger Causality analysis results

Return type:

GrangerCausalityResult

class trancit.causality.granger.GrangerCausalityResult(granger_causality, coefficients, residual_variances)[source]

Bases: BaseResult

Result container for Granger Causality analysis.

Initialize Granger Causality result.

Parameters:

granger_causality (np.ndarray) – Granger Causality values
coefficients (np.ndarray) – VAR coefficients
residual_variances (np.ndarray) – Residual variances for each variable

Additional Attributes:

pvalues (ndarray): Statistical significance p-values
f_statistics (ndarray): F-test statistics

__init__(granger_causality, coefficients, residual_variances)[source]

Initialize Granger Causality result.

Parameters:

granger_causality (np.ndarray) – Granger Causality values
coefficients (np.ndarray) – VAR coefficients
residual_variances (np.ndarray) – Residual variances for each variable

Relative Dynamic Causal Strength (rDCS)

class trancit.causality.rdcs.RelativeDCSCalculator(model_order, reference_time, estimation_mode='OLS', use_diagonal_covariance=False, use_old_version=False, **kwargs)[source]

Bases: BaseAnalyzer

Relative Dynamic Causal Strength (rDCS) calculator.

This class implements the Relative Dynamic Causal Strength algorithm for quantifying causal relationships relative to a baseline period.

Initialize Relative DCS calculator.

Parameters:

model_order (int) – Model order for VAR analysis
reference_time (int) – Reference time point for baseline calculation
estimation_mode (str) – Estimation mode: ‘OLS’ or ‘RLS’
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
use_old_version (bool) – Whether to use old version of rDCS calculation
**kwargs – Additional configuration parameters

Calculator for Relative Dynamic Causal Strength, used for event-based analysis.

About rDCS:

Relative DCS measures causal strength relative to a baseline reference time, making it particularly useful for analyzing event-related changes in causality.

Example:

from trancit.causality import RelativeDCSCalculator
import numpy as np

# Generate synthetic event data and statistics
# In practice, these would come from PipelineOrchestrator
np.random.seed(42)
model_order = 3
nvar = 2
n_obs = 50
n_trials = 10

# Event data: shape (nvar * (model_order + 1), n_obs, n_trials)
event_data = np.random.randn(nvar * (model_order + 1), n_obs, n_trials)

# Statistics dictionary with required structure
stats = {
    "OLS": {
        "At": np.random.randn(n_obs, nvar, nvar * model_order + 1),
        "Sigma_Et": np.array([np.eye(nvar) * 0.1 for _ in range(n_obs)]),
    },
    "Sigma": np.random.randn(
        n_obs, nvar * (model_order + 1), nvar * (model_order + 1)
    ),
    "mean": np.random.randn(nvar * (model_order + 1), n_obs),
}

# Create rDCS calculator
calculator = RelativeDCSCalculator(
    model_order=model_order,
    reference_time=25,  # Reference time index
    estimation_mode="OLS"
)

# Analyze with event data and statistics
result = calculator.analyze(event_data, stats)

print(f"rDCS result shape: {result.relative_dynamic_causal_strength.shape}")
print(f"Mean rDCS X→Y: {result.relative_dynamic_causal_strength[:, 1].mean():.4f}")

__init__(model_order, reference_time, estimation_mode='OLS', use_diagonal_covariance=False, use_old_version=False, **kwargs)[source]

Initialize Relative DCS calculator.

Parameters:

model_order (int) – Model order for VAR analysis
reference_time (int) – Reference time point for baseline calculation
estimation_mode (str) – Estimation mode: ‘OLS’ or ‘RLS’
use_diagonal_covariance (bool) – Whether to use diagonal covariance approximation
use_old_version (bool) – Whether to use old version of rDCS calculation
**kwargs – Additional configuration parameters

analyze(event_data, stats, **kwargs)[source]

Perform Relative Dynamic Causal Strength analysis.

Parameters:

event_data (np.ndarray) – Event data array of shape (nvar * (model_order + 1), nobs, ntrials)
stats (Dict) –
Model statistics with keys: - ‘OLS’ or ‘RLS’: Sub-dict with ‘At’ (coefficients) and ‘Sigma_Et’

(residual covariance)
- ’Sigma’: Covariance matrices
- ’mean’: Mean values
**kwargs – Additional parameters

Returns:

Relative DCS analysis results

Return type:

RelativeDCSResult

class trancit.causality.rdcs.RelativeDCSResult(transfer_entropy, dynamic_causal_strength, relative_dynamic_causal_strength, coefficients)[source]

Bases: BaseResult

Result container for Relative Dynamic Causal Strength analysis.

Initialize Relative DCS result.

Parameters:

transfer_entropy (np.ndarray) – Transfer Entropy values
dynamic_causal_strength (np.ndarray) – Dynamic Causal Strength values
relative_dynamic_causal_strength (np.ndarray) – Relative Dynamic Causal Strength values
coefficients (np.ndarray) – VAR coefficients

__init__(transfer_entropy, dynamic_causal_strength, relative_dynamic_causal_strength, coefficients)[source]

Initialize Relative DCS result.

Parameters:

transfer_entropy (np.ndarray) – Transfer Entropy values
dynamic_causal_strength (np.ndarray) – Dynamic Causal Strength values
relative_dynamic_causal_strength (np.ndarray) – Relative Dynamic Causal Strength values
coefficients (np.ndarray) – VAR coefficients

Utility Functions 

trancit.causality.rdcs.time_varying_causality(event_data, stats, causal_params)[source]

Compute time-varying causality measures for bivariate signals.

Calculates Transfer Entropy (TE), Dynamic Causal Strength (DCS), and Relative Dynamic Causal Strength (rDCS) based on a VAR model. This function maintains exact mathematical alignment with the previous implementation while providing enhanced error handling and validation.

Parameters:

event_data (np.ndarray) – Event data array of shape (nvar * (model_order + 1), nobs, ntrials). Must be bivariate (nvar = 2).
stats (Dict) –
Model statistics with keys: - ‘OLS’ or ‘RLS’: Sub-dict with ‘At’ (coefficients) and ‘Sigma_Et’

(residual covariance).
- ’Sigma’: Covariance matrices of shape (nobs, nvar * (model_order + 1), nvar * (model_order + 1)).
- ’mean’: Mean values of shape (nvar * (model_order + 1), nobs).
causal_params (Dict) – Parameters with keys: - ‘ref_time’: Reference time index for rDCS calculation. - ‘estim_mode’: ‘OLS’ or ‘RLS’ estimation mode. - ‘morder’: Model order (number of lags). - ‘diag_flag’: Boolean for diagonal covariance approximation. - ‘old_version’: Boolean for rDCS calculation method.

Returns:

Causality measures: - ‘TE’: Transfer Entropy, shape (nobs, 2) where [:, 0] is Y->X,

[:, 1] is X->Y.

’DCS’: Dynamic Causal Strength, shape (nobs, 2) where [:, 0] is Y->X, [:, 1] is X->Y.
’rDCS’: Relative Dynamic Causal Strength, shape (nobs, 2) where [:, 0] is Y->X, [:, 1] is X->Y.

Return type:

Dict[str, np.ndarray]

Raises:

ValidationError – If input data or parameters are invalid.
ComputationError – If computation fails due to numerical issues.

Notes

The function assumes bivariate data (nvar = 2).
Transfer Entropy (TE) measures directed information flow.
Dynamic Causal Strength (DCS) measures direct causal influence.
Relative Dynamic Causal Strength (rDCS) measures causal influence relative to a baseline.
All measures are computed using the Structural Causal Model (SCM) framework.

Examples

>>> event_data = np.random.randn(6, 50, 10)  # (nvar * (morder + 1), nobs, ntrials)
>>> stats = {
...     'OLS': {
...         'At': np.random.randn(50, 2, 4),
...         'Sigma_Et': np.array([np.eye(2) for _ in range(50)])
...     },
...     'Sigma': np.random.randn(50, 6, 6),
...     'mean': np.random.randn(6, 50)
... }
>>> causal_params = {
...     'ref_time': 10, 'estim_mode': 'OLS', 'morder': 2,
...     'diag_flag': False, 'old_version': False
... }
>>> result = time_varying_causality(event_data, stats, causal_params)
>>> print(f"TE shape: {result['TE'].shape}")
>>> print(f"DCS shape: {result['DCS'].shape}")
>>> print(f"rDCS shape: {result['rDCS'].shape}")

Core function for computing time-varying causality measures. Used internally by the pipeline and can be called directly for advanced use cases.

Parameters:

event_data (ndarray): Event data array, shape (nvar × (model_order + 1), nobs, ntrials)
stats (dict): Model statistics containing coefficients and covariances
causal_params (dict): Analysis parameters

Returns:

Dictionary containing ‘TE’, ‘DCS’, and ‘rDCS’ arrays.

Example:

from trancit.causality import time_varying_causality
import numpy as np

# Generate synthetic event data and statistics
# In practice, these would come from PipelineOrchestrator
np.random.seed(42)
model_order = 3
nvar = 2
n_obs = 50
n_trials = 10

# Event data: shape (nvar * (model_order + 1), n_obs, n_trials)
event_data = np.random.randn(nvar * (model_order + 1), n_obs, n_trials)

# Statistics dictionary with required structure
stats = {
    "OLS": {
        "At": np.random.randn(n_obs, nvar, nvar * model_order + 1),
        "Sigma_Et": np.array([np.eye(nvar) * 0.1 for _ in range(n_obs)]),
    },
    "Sigma": np.random.randn(
        n_obs, nvar * (model_order + 1), nvar * (model_order + 1)
    ),
    "mean": np.random.randn(nvar * (model_order + 1), n_obs),
}

# Define causal parameters
causal_params = {
    'ref_time': 25,
    'estim_mode': 'OLS',
    'morder': model_order,
    'diag_flag': False,
    'old_version': False
}

# Compute time-varying causality
causality_results = time_varying_causality(event_data, stats, causal_params)

# Extract results
dcs_values = causality_results['DCS']
te_values = causality_results['TE']
rdcs_values = causality_results['rDCS']

print(f"DCS shape: {dcs_values.shape}")
print(f"TE shape: {te_values.shape}")
print(f"rDCS shape: {rdcs_values.shape}")
print(f"Mean DCS X→Y: {dcs_values[:, 1].mean():.4f}")

Pipeline System (trancit.pipeline)

The pipeline module provides orchestrated analysis workflows for event-based DCS analysis.

Pipeline Orchestration 

class trancit.pipeline.orchestrator.PipelineOrchestrator(config)[source]

Bases: BaseAnalyzer

Pipeline orchestrator for DCS analysis.

This class coordinates all stages of the DCS analysis pipeline, including event detection, snapshot extraction, and causality analysis.

Initialize pipeline orchestrator.

Parameters:: config (PipelineConfig) – Pipeline configuration

Main orchestrator class that coordinates the complete analysis pipeline.

Analysis Stages:

Input validation
Event detection
Border removal
BIC model selection
Snapshot extraction
Artifact removal
Statistics computation
Causality analysis
Bootstrap analysis
Output preparation

Key Methods:

run(original_signal, detection_signal): Execute complete pipeline
analyze(data, **kwargs): BaseAnalyzer interface compatibility

Example:

from trancit import PipelineOrchestrator, generate_signals
from trancit.config import PipelineConfig, PipelineOptions, DetectionParams, CausalParams
import numpy as np

# Generate sample data
data, _, _ = generate_signals(T=800, Ntrial=15, h=0.1,
                              gamma1=0.4, gamma2=0.6,
                              Omega1=0.9, Omega2=1.3)

# Prepare signals for pipeline
original_signal = np.mean(data, axis=2)  # Average across trials
detection_signal = original_signal * 1.5  # Amplified for detection

# Configure pipeline
config = PipelineConfig(
    options=PipelineOptions(
        detection=True,
        causal_analysis=True,
        bootstrap=False
    ),
    detection=DetectionParams(
        thres_ratio=2.0,
        l_extract=100,
        l_start=50
    ),
    causal=CausalParams(
        ref_time=50,
        estim_mode="OLS"
    )
)

# Run pipeline
orchestrator = PipelineOrchestrator(config)
result = orchestrator.run(original_signal, detection_signal)

# Access results
if result.results.get("CausalOutput"):
    dcs_results = result.results["CausalOutput"]["OLS"]["DCS"]
    print(f"Detected {len(dcs_results)} events")

__init__(config)[source]

Initialize pipeline orchestrator.

Parameters:: config (PipelineConfig) – Pipeline configuration

analyze(data, **kwargs)[source]

Perform pipeline analysis on the given data.

This method implements the BaseAnalyzer interface and serves as a wrapper around the run method for compatibility.

Parameters:

data (np.ndarray) – Original time series signal (first argument)
**kwargs – Additional parameters including: - detection_signal: Signal used for event detection

Returns:

Pipeline analysis results

Return type:

PipelineResult

run(original_signal, detection_signal)[source]

Run the complete DCS analysis pipeline.

Parameters:

original_signal (np.ndarray) – Original time series signal
detection_signal (np.ndarray) – Signal used for event detection

Returns:

Pipeline analysis results

Return type:

PipelineResult

class trancit.pipeline.orchestrator.PipelineResult(results, config, event_snapshots)[source]

Bases: BaseResult

Result container for pipeline analysis.

Initialize pipeline result.

Parameters:

results (Dict[str, Any]) – Analysis results
config (PipelineConfig) – Pipeline configuration
event_snapshots (np.ndarray) – Extracted event snapshots

Result object containing complete pipeline outputs.

Attributes:

results (dict): Complete analysis results
config (PipelineConfig): Configuration used for analysis
event_snapshots (ndarray): Extracted event windows

__init__(results, config, event_snapshots)[source]

Initialize pipeline result.

Parameters:

results (Dict[str, Any]) – Analysis results
config (PipelineConfig) – Pipeline configuration
event_snapshots (np.ndarray) – Extracted event snapshots

Pipeline Stages 

Individual pipeline stages can be used independently for custom workflows.

class trancit.pipeline.stages.InputValidationStage(config)[source]

Bases: PipelineStage

Stage for validating input parameters and data.

Validates input signals and ensures proper format.

Parameters:: config (PipelineConfig)

execute(**kwargs)[source]

Validate input parameters and data.

Return type:: Dict[str, Any]

class trancit.pipeline.stages.EventDetectionStage(config)[source]

Bases: PipelineStage

Stage for detecting and aligning events.

Detects events in the detection signal based on threshold criteria.

Parameters:: config (PipelineConfig)

execute(**kwargs)[source]

Detect and align events based on the detection signal.

Return type:: Dict[str, Any]

class trancit.pipeline.stages.BICSelectionStage(config)[source]

Bases: PipelineStage

Stage for BIC model selection.

Performs BIC-based model order selection.

Parameters:: config (PipelineConfig)

execute(**kwargs)[source]

Perform BIC model selection if enabled.

Return type:: Dict[str, Any]

class trancit.pipeline.stages.SnapshotExtractionStage(config)[source]

Bases: PipelineStage

Stage for extracting event snapshots.

Extracts fixed-length windows around detected events.

Parameters:: config (PipelineConfig)

execute(**kwargs)[source]

Extract event snapshots using configuration parameters.

Return type:: Dict[str, Any]

class trancit.pipeline.stages.CausalityAnalysisStage(config)[source]

Bases: PipelineStage

Stage for performing causality analysis.

Computes causality measures for extracted event windows.

Parameters:: config (PipelineConfig)

execute(**kwargs)[source]

Perform causality analysis if enabled.

Return type:: Dict[str, Any]

Example of Custom Pipeline:

from trancit.pipeline.stages import InputValidationStage, EventDetectionStage, CausalityAnalysisStage
from trancit import generate_signals
from trancit.config import PipelineConfig, PipelineOptions, DetectionParams, CausalParams
import numpy as np

# Generate sample data
data, _, _ = generate_signals(T=800, Ntrial=15, h=0.1,
                              gamma1=0.4, gamma2=0.6,
                              Omega1=0.9, Omega2=1.3)

# Prepare signals
original_signal = np.mean(data, axis=2)
detection_signal = original_signal * 1.5

# Configure pipeline
config = PipelineConfig(
    options=PipelineOptions(
        detection=True,
        causal_analysis=True,
        bootstrap=False
    ),
    detection=DetectionParams(
        thres_ratio=2.0,
        l_extract=100,
        l_start=50
    ),
    causal=CausalParams(
        ref_time=50,
        estim_mode="OLS"
    )
)

# Create individual stages
validation_stage = InputValidationStage(config)
detection_stage = EventDetectionStage(config)
causality_stage = CausalityAnalysisStage(config)

# Run stages sequentially
stage_data = {'original_signal': original_signal, 'detection_signal': detection_signal}

validated_data = validation_stage.execute(**stage_data)
detected_data = detection_stage.execute(**validated_data)
# ... continue with other stages

Configuration (trancit.config)

The configuration module defines all parameter classes used throughout DCS.

Main Configuration 

class trancit.config.PipelineConfig(options, detection, bic, causal, output, monte_carlo=None, desnap=None, sampling_rate=1252, passband=<factory>)[source]

Bases: object

Main configuration object for the analysis pipeline.

This class combines all configuration parameters into a single object for easy management and validation of pipeline settings.

Parameters:

options (PipelineOptions)
detection (DetectionParams)
bic (BicParams)
causal (CausalParams)
output (OutputParams)
monte_carlo (MonteCParams | None)
desnap (DeSnapParams | None)
sampling_rate (int)
passband (List[int])

options

Pipeline execution options

Type:: PipelineOptions

detection

Event detection parameters

Type:: DetectionParams

bic

BIC model selection parameters

Type:: BicParams

causal

Causality analysis parameters

Type:: CausalParams

output

Output file parameters

Type:: OutputParams

monte_carlo

Monte Carlo bootstrap parameters (default: None)

Type:: Optional[MonteCParams]

desnap

DeSnap analysis parameters (default: None)

Type:: Optional[DeSnapParams]

sampling_rate

Sampling rate in Hz (default: 1252)

Type:: int

passband

Frequency passband [low, high] in Hz (default: [140, 230])

Type:: List[int]

Main configuration class that contains all pipeline parameters.

Components:

options: PipelineOptions - controls which analysis stages to run
detection: DetectionParams - event detection parameters
bic: BicParams - BIC model selection parameters
causal: CausalParams - causality analysis parameters
monte_carlo: MonteCParams - bootstrap analysis parameters
output: OutputParams - output formatting parameters

Example:

from trancit.config import *

config = PipelineConfig(
    options=PipelineOptions(
        detection=True,
        bic=True,
        causal_analysis=True,
        bootstrap=False
    ),
    detection=DetectionParams(
        thres_ratio=2.5,
        l_extract=150,
        align_type="peak"
    ),
    bic=BicParams(
        morder=4,
        momax=8,
        mode="OLS"
    ),
    causal=CausalParams(
        ref_time=75,
        estim_mode="OLS",
        diag_flag=False
    )
)

options: PipelineOptions

detection: DetectionParams

bic: BicParams

causal: CausalParams

output: OutputParams

monte_carlo: MonteCParams | None = None

desnap: DeSnapParams | None = None

sampling_rate: int = 1252

passband: List[int]

__post_init__()[source]

Validate configuration parameters after initialization.

Return type:: None

__init__(options, detection, bic, causal, output, monte_carlo=None, desnap=None, sampling_rate=1252, passband=<factory>)

Parameters:

options (PipelineOptions)
detection (DetectionParams)
bic (BicParams)
causal (CausalParams)
output (OutputParams)
monte_carlo (MonteCParams | None)
desnap (DeSnapParams | None)
sampling_rate (int)
passband (List[int])

Return type:

None

Parameter Classes 

class trancit.config.PipelineOptions(detection=True, bic=False, causal_analysis=True, bootstrap=False, save_flag=False, debiased_stats=True)[source]

Bases: object

Options to control which pipeline steps are executed.

Parameters:

detection (bool)
bic (bool)
causal_analysis (bool)
bootstrap (bool)
save_flag (bool)
debiased_stats (bool)

detection

Whether to perform event detection (default: True)

Type:: bool

bic

Whether to perform BIC model selection (default: False)

Type:: bool

causal_analysis

Whether to perform causality analysis (default: True)

Type:: bool

bootstrap

Whether to perform bootstrap analysis (default: False)

Type:: bool

save_flag

Whether to save results to files (default: False)

Type:: bool

debiased_stats

Whether to perform DeSnap analysis (default: True)

Type:: bool

Controls which analysis stages are enabled.

Key Parameters:

detection (bool): Enable event detection
bic (bool): Enable BIC model selection
causal_analysis (bool): Enable causality analysis
bootstrap (bool): Enable bootstrap significance testing
save_flag (bool): Save intermediate results
debiased_stats (bool): Use debiased statistics

detection: bool = True

bic: bool = False

causal_analysis: bool = True

bootstrap: bool = False

save_flag: bool = False

debiased_stats: bool = True

__init__(detection=True, bic=False, causal_analysis=True, bootstrap=False, save_flag=False, debiased_stats=True)

Parameters:

detection (bool)
bic (bool)
causal_analysis (bool)
bootstrap (bool)
save_flag (bool)
debiased_stats (bool)

Return type:

None

class trancit.config.DetectionParams(thres_ratio, align_type, l_extract, l_start, shrink_flag=False, locs=None, remove_artif=False, remove_artif_threshold=-15000)[source]

Bases: object

Parameters for the event detection step.

Parameters:

thres_ratio (float)
align_type (str)
l_extract (int)
l_start (int)
shrink_flag (bool)
locs (ndarray | None)
remove_artif (bool)
remove_artif_threshold (float)

thres_ratio

Threshold ratio for event detection

Type:: float

align_type

Alignment type: ‘peak’ or ‘pooled’

Type:: str

l_extract

Length of extracted snapshots

Type:: int

l_start

Start offset for snapshot extraction

Type:: int

shrink_flag

Whether to use shrinking for pooled alignment (default: False)

Type:: bool

locs

Pre-provided event locations (default: None)

Type:: Optional[np.ndarray]

remove_artif

Whether to remove artifact trials (default: False)

Type:: bool

Parameters for event detection.

Key Parameters:

thres_ratio (float): Detection threshold (1.5-3.0 typical range)
l_extract (int): Length of extracted event windows
l_start (int): Start offset within windows
align_type (str): “peak” or “onset” alignment
shrink_flag (bool): Apply shrinkage procedure
remove_artif (bool): Remove artifact-contaminated trials

thres_ratio: float

align_type: str

l_extract: int

l_start: int

shrink_flag: bool = False

locs: ndarray | None = None

remove_artif: bool = False

remove_artif_threshold: float = -15000

__init__(thres_ratio, align_type, l_extract, l_start, shrink_flag=False, locs=None, remove_artif=False, remove_artif_threshold=-15000)

Parameters:

thres_ratio (float)
align_type (str)
l_extract (int)
l_start (int)
shrink_flag (bool)
locs (ndarray | None)
remove_artif (bool)
remove_artif_threshold (float)

Return type:

None

class trancit.config.CausalParams(ref_time, estim_mode='OLS', diag_flag=False, old_version=False)[source]

Bases: object

Parameters for causality calculation.

Parameters:

ref_time (int)
estim_mode (str)
diag_flag (bool)
old_version (bool)

ref_time

Reference time point for causality analysis

Type:: int

estim_mode

Estimation mode: ‘OLS’ or ‘RLS’ (default: ‘OLS’)

Type:: str

diag_flag

Whether to use diagonal covariance (default: False)

Type:: bool

old_version

Whether to use old version of rDCS calculation (default: False)

Type:: bool

Parameters for causality analysis.

Key Parameters:

ref_time (int): Reference time for rDCS calculation
estim_mode (str): “OLS” or “RLS” estimation
diag_flag (bool): Use diagonal covariance approximation
old_version (bool): Use legacy rDCS calculation method

ref_time: int

estim_mode: str = 'OLS'

diag_flag: bool = False

old_version: bool = False

__init__(ref_time, estim_mode='OLS', diag_flag=False, old_version=False)

Parameters:

ref_time (int)
estim_mode (str)
diag_flag (bool)
old_version (bool)

Return type:

None

class trancit.config.BicParams(morder, momax=None, tau=None, mode=None, estim_mode=None)[source]

Bases: object

Parameters for BIC model order selection.

Parameters:

morder (int)
momax (int | None)
tau (int | None)
mode (str | None)
estim_mode (str | None)

morder

Model order to use if BIC is False, or default

Type:: int

momax

Max order to test if BIC is True (default: None)

Type:: Optional[int]

tau

Lag step if BIC is True (default: None)

Type:: Optional[int]

mode

BIC mode, e.g., ‘biased’ (default: None)

Type:: Optional[str]

Parameters for BIC model selection.

morder: int

momax: int | None = None

tau: int | None = None

mode: str | None = None

estim_mode: str | None = None

__init__(morder, momax=None, tau=None, mode=None, estim_mode=None)

Parameters:

morder (int)
momax (int | None)
tau (int | None)
mode (str | None)
estim_mode (str | None)

Return type:

None

class trancit.config.MonteCParams(n_btsp=100)[source]

Bases: object

Parameters for Monte Carlo bootstrapping.

Parameters:: n_btsp (int)

n_btsp

Number of bootstrap samples (default: 100)

Type:: int

Parameters for bootstrap analysis.

n_btsp: int = 100

__init__(n_btsp=100)

Parameters:: n_btsp (int)
Return type:: None

class trancit.config.OutputParams(file_keyword, save_path='')[source]

Bases: object

Parameters for output file naming.

Parameters:

file_keyword (str)
save_path (str)

file_keyword

Keyword for output file naming

Type:: str

save_path

Path for saving output files

Type:: str

Parameters for output formatting.

file_keyword: str

save_path: str = ''

__init__(file_keyword, save_path='')

Parameters:

file_keyword (str)
save_path (str)

Return type:

None

Model Estimation (trancit.models)

The models module provides VAR model estimation and selection tools.

VAR Model Estimation 

class trancit.models.var_estimation.VAREstimator(model_order, time_mode='inhomo')[source]

Bases: object

Vector Autoregressive (VAR) model estimator.

This class provides methods for estimating VAR model parameters including coefficients, residuals, and covariance matrices.

Initialize VAR estimator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)

Vector Autoregressive (VAR) model estimation using OLS or RLS methods.

Example:

from trancit.models import VAREstimator
import numpy as np

# Generate sample data
data = np.random.randn(2, 1000, 20)  # 2 vars, 1000 time points, 20 trials

# Create estimator
estimator = VAREstimator(model_order=4, time_mode="inhomo")

# Estimate VAR coefficients
coefficients, residual_covariance, log_likelihood, hessian_sum = (
    estimator.estimate_var_coefficients(
        data, model_order=4, max_model_order=6,
        time_mode="inhomo", lag_mode="infocrit"
    )
)

print(f"Coefficients shape: {coefficients.shape}")
print(f"Residual covariance shape: {residual_covariance.shape}")
print(f"Log-likelihood: {log_likelihood:.4f}")

__init__(model_order, time_mode='inhomo')[source]

Initialize VAR estimator.

Parameters:

model_order (int) – Model order for VAR analysis
time_mode (str) – Time mode: ‘inhomo’ (time-inhomogeneous) or ‘homo’ (homogeneous)

estimate_var_coefficients(time_series_data, model_order, max_model_order, time_mode, lag_mode, epsilon=1e-8)[source]

Estimate VAR coefficients and compute residual covariance with regularization.

Supports both inhomogeneous (‘inhomo’) and homogeneous (‘homo’) time modes.

Parameters:

time_series_data (np.ndarray) – Data array, shape (n_vars, n_observations, n_trials).
model_order (int) – Model order for the VAR process.
max_model_order (int) – Maximum model order for lag_mode ‘infocrit’.
time_mode (str) – ‘inhomo’ or ‘homo’.
lag_mode (str) – ‘infocrit’ or ‘var’ to determine lag structure.
epsilon (float, optional) – Regularization term for singular matrices. Defaults to 1e-8.

Returns:

coefficients: VAR coefficients, shape varies by time_mode.
residual_covariance: Residual covariance, shape varies by time_mode.
log_likelihood: Log-likelihood of the model.
sum_log_det_hessian: Sum of log determinants of the Hessian.

Return type:

Tuple[np.ndarray, np.ndarray, float, float]

Raises:

ValueError – If n_vars == 1, or time_mode/lag_mode is invalid.

BIC Model Selection 

class trancit.models.bic_selection.BICSelector(config)[source]

Bases: object

Bayesian Information Criterion (BIC) selector for VAR model order selection.

This class provides methods for computing BIC values and selecting optimal model orders for Vector Autoregression (VAR) models.

Initialize BIC selector.

Parameters:: config (PipelineConfig) – Pipeline configuration

Bayesian Information Criterion (BIC) for automatic model order selection.

Example:

from trancit.models import BICSelector, select_model_order
from trancit.config import PipelineConfig, PipelineOptions, BicParams
import numpy as np

# Generate sample data
data = np.random.randn(2, 500, 20)  # 2 vars, 500 time points, 20 trials

# Configure pipeline with BIC parameters
config = PipelineConfig(
    options=PipelineOptions(bic=True),
    bic=BicParams(
        morder=4,
        momax=10,
        mode="biased"
    )
)

# Option 1: Use BICSelector (requires event data from pipeline)
# selector = BICSelector(config)
# bic_results = selector._compute_multi_trial_bic(event_data_max_order)

# Option 2: Use high-level select_model_order function (simpler)
bic_scores, optimal_orders, log_likelihoods, penalty_terms = (
    select_model_order(data, max_model_order=10, time_mode="inhomo")
)

print(f"Optimal model orders: {optimal_orders}")
print(f"BIC scores shape: {bic_scores.shape}")

__init__(config)[source]

Initialize BIC selector.

Parameters:: config (PipelineConfig) – Pipeline configuration

Model Validation 

class trancit.models.model_validation.ModelValidator(max_lag=10, significance_level=0.05, stability_threshold=1.0, **kwargs)[source]

Bases: BaseAnalyzer

Model validator for VAR models.

This class provides comprehensive validation tools for assessing the quality and reliability of fitted VAR models.

Initialize model validator.

Parameters:

max_lag (int) – Maximum lag for autocorrelation testing
significance_level (float) – Significance level for hypothesis tests
stability_threshold (float) – Threshold for model stability (eigenvalues)
**kwargs – Additional configuration parameters

Tools for validating fitted VAR models.

__init__(max_lag=10, significance_level=0.05, stability_threshold=1.0, **kwargs)[source]

Initialize model validator.

Parameters:

max_lag (int) – Maximum lag for autocorrelation testing
significance_level (float) – Significance level for hypothesis tests
stability_threshold (float) – Threshold for model stability (eigenvalues)
**kwargs – Additional configuration parameters

validate(coefficients, residuals, design_matrix, n_observations, n_parameters, **kwargs)[source]

Perform comprehensive model validation.

Parameters:

coefficients (np.ndarray) – VAR coefficients
residuals (np.ndarray) – Model residuals
design_matrix (np.ndarray) – Design matrix used for estimation
n_observations (int) – Number of observations
n_parameters (int) – Number of parameters in the model
**kwargs – Additional parameters

Returns:

Model validation results

Return type:

ModelValidationResult

Simulation (trancit.simulation)

The simulation module provides tools for generating synthetic time series data for testing and validation.

Signal Generation 

trancit.simulation.generate_signals(T, Ntrial, h, gamma1, gamma2, Omega1, Omega2, apply_morlet=False)[source]

Generate bivariate coupled oscillator signals based on a specified model.

Simulates two coupled second-order linear differential equations discretized using a time step h, with added noise terms. Allows for optional non-stationarity in the noise applied to the first signal (x) via a Morlet wavelet shape.

Parameters:

T (int) – Total number of time points to simulate (including initial points).
Ntrial (int) – Number of trials (realizations) to generate.
h (float) – Time step for discretization.
gamma1 (float) – Damping coefficient for the first oscillator (x).
gamma2 (float) – Damping coefficient for the second oscillator (y).
Omega1 (float) – Natural frequency for the first oscillator (x).
Omega2 (float) – Natural frequency for the second oscillator (y).
apply_morlet (bool, optional) – If True, applies a Morlet wavelet shape to modulate the noise variance (ns_x) for the first signal, introducing non-stationarity. Defaults to False, using constant noise variance.

Returns:

X : Generated signals, shape (2, T - burnin, Ntrial). Contains x and y signals after discarding burn-in points.
ns_x : Noise variance profile for the x signal, shape (T + 1,).
ns_y : Noise variance profile for the y signal, shape (T + 1,).

Return type:

Tuple[np.ndarray, np.ndarray, np.ndarray]

Generate synthetic bivariate coupled oscillator signals.

Parameters:

T (int): Number of time points
Ntrial (int): Number of trials
h (float): Time step
gamma1, gamma2 (float): Damping parameters
Omega1, Omega2 (float): Natural frequencies

Returns:

Tuple of (data, time_axis, parameters) where data has shape (2, T, Ntrial).

Example:

from trancit import generate_signals

data, time_axis, params = generate_signals(
    T=1000, Ntrial=20, h=0.1,
    gamma1=0.5, gamma2=0.5,
    Omega1=1.0, Omega2=1.2
)

print(f"Generated data shape: {data.shape}")

AR Event Simulation 

trancit.simulation.simulate_ar_event(simobj, Yt_stats)[source]

Simulates AR events with non-stationary innovations using the specified simulation object and statistics.

Parameters: - simobj (dict): Simulation settings with fields:

nvar (int): Number of variables.

morder (int): Model order.

L (int): Length of the event.

Ntrials (int): Number of trials.

Yt_stats (dict): Statistics with fields:
- OLS.At (numpy.ndarray): Autoregressive coefficients.
- OLS.Sigma_Et (numpy.ndarray): Covariance matrices for errors.
- OLS.bt (numpy.ndarray): Mean term for innovations.
- mean (numpy.ndarray): Mean state.
- Sigma (numpy.ndarray): Covariance matrix of the state.

Returns: - Yt_event (numpy.ndarray): Simulated AR events.

Generate autoregressive events with controlled causality.

Parameters:

simobj (Dict)
Yt_stats (Dict)

Return type:

ndarray

trancit.simulation.simulate_ar_event_bootstrap(simobj, Yt_event, Yt_stats, Et)[source]

Simulate autoregressive (AR) events with bootstrapping.

Parameters: simobj : dict

A dictionary containing simulation parameters: - nvar : int, number of variables - morder : int, model order - L : int, length of each trial - Ntrials : int, number of trials

Yt_eventnp.ndarray: Original event data of shape (nvar * (morder + 1), L, trials).
Yt_statsdict: Dictionary containing OLS coefficients under OLS.At of shape (L, nvar, nvar * morder).
Etnp.ndarray: Residuals of shape (nvar, L, trials).

Returns: Yt_event_btsp : np.ndarray

Bootstrapped event data of shape (nvar * (morder + 1), L, trials).

Generate multiple bootstrap samples of AR events.

Parameters:

simobj (Dict)
Yt_event (ndarray)
Yt_stats (Dict)
Et (ndarray)

Return type:

ndarray

VAR Simulation 

trancit.simulation.generate_var_nonstat(A, SIG, morder, nvar, L_event, amp, dim, L_perturb, center)[source]

Generate a non-stationary VAR process.

Args: A: Coefficient matrix of the VAR process. SIG: Covariance matrix of the innovations. morder: Number of lags in the VAR model. nvar: Number of variables. L_event: Length of the event (time series). amp: Amplitude of the perturbation. dim: Dimension of variables the input goes into. L_perturb: Length of perturbation. center: Center point for perturbation.

Returns: X: Generated VAR process (nvar x L_event). Imp: Impulse matrix (2 x L_event).

Generate non-stationary VAR processes.

Parameters:

A (ndarray)
SIG (ndarray)
morder (int)
nvar (int)
L_event (int)
amp (float)
dim (int)
L_perturb (int)
center (int)

Return type:

Tuple[ndarray, ndarray]

Utilities (trancit.utils)

The utilities module provides helper functions for data processing, analysis, and visualization.

Data Processing 

trancit.utils.preprocess.normalize_data(data, method='zscore', axis=None)[source]

Normalize data using various methods.

This function provides different normalization methods for data preprocessing.

Parameters:

data (np.ndarray) – Input data to normalize.
method (str, optional) – Normalization method: ‘zscore’, ‘minmax’, ‘robust’, or ‘none’. Default is ‘zscore’.
axis (Optional[int], optional) – Axis along which to normalize. If None, normalize over all axes. Default is None.

Returns:

Normalized data.

Return type:

np.ndarray

Raises:

ValidationError – If input parameters are invalid.
ComputationError – If normalization fails.

Examples

>>> import numpy as np
>>> data = np.random.randn(2, 100, 10)
>>> normalized = normalize_data(data, method='zscore', axis=1)
>>> print(f"Original mean: {np.mean(data):.3f}")
>>> print(f"Normalized mean: {np.mean(normalized):.3f}")

Normalize time series data using various methods.

Parameters:

data (ndarray): Input data
method (str): “zscore”, “minmax”, or “robust”
axis (int): Axis along which to normalize

Example:

from trancit.utils.preprocess import normalize_data
import numpy as np

# Generate sample data
data = np.random.randn(2, 500, 20)  # 2 vars, 500 time points, 20 trials

# Normalize data
normalized_data = normalize_data(data, method="zscore", axis=1)

print(f"Original data mean: {data.mean():.4f}, std: {data.std():.4f}")
print(f"Normalized data mean: {normalized_data.mean():.4f}, std: {normalized_data.std():.4f}")

Core Analysis Functions 

trancit.utils.core.compute_event_statistics(event_data, model_order, epsilon=1e-5)[source]

Compute conditional statistics for VAR time series events.

This function computes mean, covariance, and OLS coefficients for VAR time series events. It handles both homogeneous and inhomogeneous VAR models.

Parameters:

event_data (np.ndarray) – VAR time series events of shape (nvar * (model_order + 1), time_points, trials).
model_order (int) – The model order for the VAR process.
epsilon (float, optional) – Small value for regularization if the matrix is singular. Default is 1e-4.

Returns:

Dictionary containing the conditional statistics:

’mean’: Mean of the events (shape: (nvar * (model_order + 1), time_points))
’Sigma’: Covariance matrices (shape: (time_points, nvar * (model_order + 1), nvar * (model_order + 1)))
’OLS’: Dictionary with:
- ’At’: OLS coefficients (shape: (time_points, nvar, nvar * model_order))
- ’bt’: Residual biases
- ’Sigma_Et’: Residual covariance
- ’sigma_Et’: Residual standard deviation

Return type:

Dict[str, Union[np.ndarray, Dict]]

Raises:

ValidationError – If input parameters are invalid.
ComputationError – If computation fails due to numerical issues.

Examples

>>> import numpy as np
>>> event_data = np.random.randn(6, 50, 10)  # (nvar * (morder + 1), time, trials)
>>> stats = compute_conditional_event_statistics(event_data, model_order=2)
>>> print(f"Mean shape: {stats['mean'].shape}")
>>> print(f"Sigma shape: {stats['Sigma'].shape}")

Compute statistics for event snapshots.

trancit.utils.core.perform_desnap_analysis(inputs)[source]

Perform DeSnap analysis for event-based causality.

This function implements the DeSnap (De-Snapshot) analysis method for event-based causality analysis. It performs a “de-snapshotting” analysis to derive unconditional statistics from conditional statistics by accounting for a conditioning variable ‘D’.

Parameters:

inputs (DeSnapParams) – Configuration object containing all parameters for DeSnap analysis.

Returns:

Dictionary containing DeSnap analysis results:

’loc_size’: Size of locations per bin of D
’p_t’, ‘q_t’: Coefficients from the first linear regression
’d_bin_bar’: Mean D values for each bin
’mean_Yt_cond’: Mean Yt events per bin of D
’mu_D’: Estimated unconditional mean of D
’event_stats_uncond’: Unconditional statistics
’cov_pt’: Covariance related to p_t
’c’: Covariance adjustment factor

Return type:

Dict[str, Union[np.ndarray, Dict]]

Raises:

ValidationError – If input parameters are invalid.
ComputationError – If computation fails due to numerical issues.

Examples

>>> from trancit.config import DeSnapParams
>>> params = DeSnapParams(...)  # Configure parameters
>>> results = perform_desnap_analysis(params)
>>> print(f"Results keys: {list(results.keys())}")

Perform debiased snapshot analysis.

Signal Processing 

trancit.utils.signal.find_peak_locations(signal, candidate_locations, window_size)[source]

Refine peak locations by grouping candidates and local maximization.

Groups candidate locations within window_size, finds the maximum in each group, and refines the peak within a local window around each maximum.

Parameters:

signal (np.ndarray) – 1D array of signal data.
candidate_locations (np.ndarray) – 1D array of candidate peak indices.
window_size (int) – Size of the window for grouping and refinement.

Returns:

1D array of refined peak locations.

Return type:

np.ndarray

Raises:

ValueError – If window_size is negative or inputs are empty/invalid.

Find peak locations in signals for event detection.

Residual Analysis 

trancit.utils.residuals.estimate_residuals(event_stats)[source]

Estimate residuals and related statistics for each time step in a VAR model.

Parameters:

event_stats (Dict) –

Dictionary containing: - ‘OLS’: Sub-dictionary with ‘At’ (coefficients), shape (L, nvar,

nvar * morder).

’Sigma’: Covariance matrices, shape (L, nvar * (morder + 1), nvar * (morder + 1)).
’mean’: Mean values, shape (nvar * (morder + 1), L).

Returns:

residual_biases : Shape (nvar, L), residual biases.
residual_covariance : Shape (L, nvar, nvar), residual covariance matrices.
residual_trace : Shape (L, 1), trace of residual covariance matrices.

Return type:

Tuple[np.ndarray, np.ndarray, np.ndarray]

Estimate VAR model residuals.

trancit.utils.residuals.get_residuals(event_data, stats)[source]

Calculate residuals for each time step in a VAR model.

Parameters:

event_data (np.ndarray) – Event matrix, shape (nvar * (morder + 1), L, ntrials).
stats (Dict) –
Dictionary containing: - ‘OLS’: Sub-dictionary with ‘At’ (coefficients), shape (L, nvar,

nvar * morder).

Returns:

Residuals, shape (nvar, L, ntrials).

Return type:

np.ndarray

Raises:

ValueError – If coefficient and lagged data shapes are incompatible.

Get residuals from fitted models.

# Complete workflow example
import numpy as np
from trancit import DCSCalculator, generate_signals
from trancit.utils.preprocess import normalize_data
import matplotlib.pyplot as plt

# 1. Generate or load data
data, _, _ = generate_signals(T=500, Ntrial=20, h=0.1,
                             gamma1=0.5, gamma2=0.5,
                             Omega1=1.0, Omega2=1.2)

# 2. Preprocess data
clean_data = normalize_data(data, method="zscore", axis=1)

# 3. Perform analysis
calculator = DCSCalculator(model_order=4, time_mode="inhomo")
result = calculator.analyze(clean_data)

# 4. Examine results
print(f"Mean X→Y causality: {result.causal_strength[:, 1].mean():.4f}")
print(f"Mean Y→X causality: {result.causal_strength[:, 0].mean():.4f}")

# 5. Visualize results
plt.figure(figsize=(12, 4))
plt.subplot(1, 2, 1)
plt.plot(result.causal_strength[:, 1], label='X→Y', linewidth=2)
plt.plot(result.causal_strength[:, 0], label='Y→X', linewidth=2)
plt.xlabel('Time')
plt.ylabel('Causal Strength')
plt.legend()
plt.title('Dynamic Causal Strength')

plt.subplot(1, 2, 2)
plt.plot(result.transfer_entropy[:, 1], label='X→Y', linewidth=2)
plt.plot(result.transfer_entropy[:, 0], label='Y→X', linewidth=2)
plt.xlabel('Time')
plt.ylabel('Transfer Entropy')
plt.legend()
plt.title('Transfer Entropy')

plt.tight_layout()
plt.show()

Event-Based Analysis Workflow 

# Complete event-based analysis
from trancit import PipelineOrchestrator, generate_signals
from trancit.config import *
import numpy as np
import matplotlib.pyplot as plt

# Generate data
data, _, _ = generate_signals(T=800, Ntrial=15, h=0.1,
                             gamma1=0.4, gamma2=0.6,
                             Omega1=0.9, Omega2=1.3)

original_signal = np.mean(data, axis=2)
detection_signal = original_signal * 1.5

# Configure pipeline
config = PipelineConfig(
    options=PipelineOptions(
        detection=True,
        bic=False,
        causal_analysis=True,
        bootstrap=False
    ),
    detection=DetectionParams(
        thres_ratio=2.0,
        l_extract=100,
        l_start=50,
        align_type="peak"
    ),
    causal=CausalParams(
        ref_time=50,
        estim_mode="OLS"
    )
)

# Run analysis
orchestrator = PipelineOrchestrator(config)
result = orchestrator.run(original_signal, detection_signal)

# Process results
if result.results.get("CausalOutput"):
    causal_output = result.results["CausalOutput"]["OLS"]

    if "DCS" in causal_output:
        dcs_results = causal_output["DCS"]
        print(f"Events detected: {len(dcs_results)}")
        print(f"Mean event-wise X→Y causality: {dcs_results[:, 1].mean():.4f}")

        # Plot per-event causality
        plt.figure(figsize=(10, 6))
        event_numbers = range(1, len(dcs_results) + 1)

        plt.bar(event_numbers, dcs_results[:, 1], alpha=0.7, label='X→Y')
        plt.bar(event_numbers, dcs_results[:, 0], alpha=0.7, label='Y→X')
        plt.xlabel('Event Number')
        plt.ylabel('Causal Strength')
        plt.legend()
        plt.title('Event-Based Causality Analysis')
        plt.show()

Performance Notes 

Memory Usage:

DCS analysis memory scales approximately as O(T × N × M²) where T=time points, N=trials, M=model order
For large datasets, consider processing in chunks or using use_diagonal_covariance=True

Computation Time:

Typical analysis time: 0.1-10 seconds depending on data size
Time-inhomogeneous mode is slower but more accurate for non-stationary data
BIC model selection adds significant computation time but improves model quality

Optimization Tips:

Use time_mode="homo" for stationary data (faster)
Enable use_diagonal_covariance=True for numerical stability with highly correlated data
Choose model order carefully: higher order = more accurate but slower
Normalize data to improve numerical stability
Use multiple trials to improve statistical robustness

Troubleshooting 

Common Issues:

“Input data must be bivariate” - Ensure data shape is (2, time_points, trials)
Singular matrix errors - Try use_diagonal_covariance=True - Check data for perfect correlations - Normalize data using normalize_data()
“Insufficient observations” - Reduce model order - Collect more data - Use time_mode="homo" for short series
Pipeline finds no events - Lower thres_ratio parameter - Check detection signal characteristics - Visualize signals to understand event structure

Getting Help:

Check the tutorials in Tutorials
Review examples in the examples/ directory
Open an issue on GitHub
Read the scientific paper for theoretical background

API Reference

Core Module (trancit.core)

Base Classes 

Exception Classes 

Causality Analysis (trancit.causality)

Dynamic Causal Strength (DCS)

Transfer Entropy 

Granger Causality 

Relative Dynamic Causal Strength (rDCS)

Utility Functions 

Pipeline System (trancit.pipeline)

Pipeline Orchestration 

Pipeline Stages 

Configuration (trancit.config)

Main Configuration 

Parameter Classes 

Model Estimation (trancit.models)

VAR Model Estimation 

BIC Model Selection 

Model Validation 

Simulation (trancit.simulation)

Signal Generation 

AR Event Simulation 

VAR Simulation 

Utilities (trancit.utils)

Data Processing 

Core Analysis Functions 

Signal Processing 

Residual Analysis 

Visualization 

Helper Functions 

Complete Usage Examples 

Basic Analysis Workflow 

Event-Based Analysis Workflow 

Performance Notes 

Troubleshooting 