Processing API Reference

Import: from ecgdatakit.processing import ...

Requires: pip install ecgdatakit[processing] (scipy ≥ 1.10)

All filter and transform functions accept a Lead and return a new Lead (immutable pattern via dataclasses.replace). The original lead is never modified.

Filters

All filters use SOS (second-order sections) + zero-phase sosfiltfilt to preserve ECG morphology.

Function	Returns	Description
`lowpass(lead, cutoff, order=4)`	`Lead`	Butterworth low-pass filter
`highpass(lead, cutoff, order=4)`	`Lead`	Butterworth high-pass filter
`bandpass(lead, low, high, order=4)`	`Lead`	Butterworth band-pass filter
`notch(lead, freq=50.0, quality=30.0)`	`Lead`	IIR notch (band-stop) filter
`remove_baseline(lead, cutoff=0.5, order=2)`	`Lead`	Remove baseline wander (highpass at 0.5 Hz)
`diagnostic_filter(lead, notch_freq=50.0)`	`Lead`	AHA diagnostic: 0.05–150 Hz bandpass + notch
`monitoring_filter(lead, notch_freq=50.0)`	`Lead`	Monitoring: 0.67–40 Hz bandpass + notch

from ecgdatakit.processing import diagnostic_filter, notch

# Diagnostic-grade filtering
filtered = diagnostic_filter(lead)

# Or build a custom pipeline
filtered = notch(lead, freq=60.0)  # 60 Hz for US mains

Resampling

Function	Returns	Description
`resample(lead, target_rate)`	`Lead`	Polyphase resampling to target sample rate

from ecgdatakit.processing import resample

lead_250 = resample(lead, target_rate=250)
print(lead_250.sample_rate)  # 250

Normalization

Pure numpy — no scipy required.

Function	Returns	Description
`normalize_minmax(lead)`	`Lead`	Scale to [−1, 1] range
`normalize_zscore(lead)`	`Lead`	Zero mean, unit variance
`normalize_amplitude(lead, target_mv=1.0)`	`Lead`	Scale peak amplitude to target

R-Peak Detection

Function	Returns	Description
`detect_r_peaks(lead, method="pan_tompkins")`	`NDArray[np.intp]`	R-peak sample indices. Methods: `"pan_tompkins"` (bandpass + derivative + adaptive threshold) or `"shannon_energy"` (Shannon energy envelope detector).
`heart_rate(lead, peaks=None)`	`float`	Average heart rate in bpm
`rr_intervals(lead, peaks=None)`	`NDArray[np.float64]`	RR intervals in milliseconds
`instantaneous_heart_rate(lead, peaks=None)`	`NDArray[np.float64]`	Beat-by-beat heart rate in bpm

All functions auto-detect peaks if peaks=None.

from ecgdatakit.processing import detect_r_peaks, heart_rate, rr_intervals

# Pan-Tompkins (default)
peaks = detect_r_peaks(filtered)

# Shannon energy envelope — good for noisy signals
peaks_se = detect_r_peaks(filtered, method="shannon_energy")

hr = heart_rate(filtered, peaks)      # e.g. 72.5
rr = rr_intervals(filtered, peaks)    # array of RR in ms

Heart Rate Variability

Function	Returns	Description
`time_domain(rr_ms)`	`dict`	Keys: `mean_rr`, `sdnn`, `rmssd`, `sdsd`, `nn50_count`, `pnn50`, `nn20_count`, `pnn20`, `hr_mean`, `hr_std`
`frequency_domain(rr_ms, method="welch", interp_fs=4.0)`	`dict`	Keys: `vlf_power` (0–0.04 Hz), `lf_power` (0.04–0.15 Hz), `hf_power` (0.15–0.4 Hz), `lf_hf_ratio`, `total_power`
`poincare(rr_ms)`	`dict`	Keys: `sd1`, `sd2`, `sd1_sd2_ratio`

from ecgdatakit.processing import time_domain, frequency_domain, poincare

hrv_t = time_domain(rr)
print(hrv_t["sdnn"], hrv_t["rmssd"], hrv_t["pnn50"])

hrv_f = frequency_domain(rr)
print(hrv_f["lf_hf_ratio"])

p = poincare(rr)
print(p["sd1"], p["sd2"])

Transforms & Segmentation

Function	Returns	Description
`power_spectrum(lead, method="welch", nperseg=None)`	`tuple[NDArray, NDArray]`	(frequencies, power) — Welch PSD
`fft(lead)`	`tuple[NDArray, NDArray]`	(frequencies, magnitudes) — single-sided FFT
`segment_beats(lead, peaks=None, before=0.2, after=0.4)`	`list[Lead]`	Extract individual heartbeats around R-peaks
`average_beat(lead, peaks=None, before=0.2, after=0.4)`	`Lead`	Ensemble-averaged beat template

Signal Quality

Function	Returns	Description
`signal_quality_index(lead)`	`float`	Composite SQI score (0.0–1.0)
`classify_quality(lead)`	`str`	`"excellent"` (>0.8), `"acceptable"` (0.5–0.8), `"unacceptable"` (<0.5)
`snr_estimate(lead)`	`float`	Signal-to-noise ratio in dB

Lead Derivation

Pure numpy — no scipy required. Validates sample rate and length match.

Function	Returns	Description
`derive_lead_iii(lead_i, lead_ii)`	`Lead`	III = II − I (Einthoven's law)
`derive_augmented(lead_i, lead_ii)`	`list[Lead]`	[aVR, aVL, aVF]
`derive_standard_12(lead_i, lead_ii, v1..v6)`	`list[Lead]`	Full 12-lead set in standard order
`find_lead(leads, label)`	`Lead \| None`	Case-insensitive lead lookup

from ecgdatakit.processing import derive_augmented, find_lead

lead_i = find_lead(record.leads, "I")
lead_ii = find_lead(record.leads, "II")

avr, avl, avf = derive_augmented(lead_i, lead_ii)

ECG Cleaning

Unified cleaning interface with multiple backends. All methods return a new Lead.

Function	Returns	Description
`clean_ecg(lead, method="default", **kwargs)`	`Lead`	Clean an ECG lead. Methods: `"default"`, `"biosppy"`, `"neurokit2"`, `"combined"`, `"deepfade"`.

Method	Extra dependency	Description
`"default"`	scipy	Bandpass 0.5–40 Hz + 50 Hz notch
`"biosppy"`	`pip install biosppy`	BioSPPy ECG filter
`"neurokit2"`	`pip install neurokit2`	NeuroKit2 adaptive pipeline
`"combined"`	biosppy + neurokit2	BioSPPy → NeuroKit2
`"deepfade"`	`pip install torch`	DeepFADE DenseNet encoder-decoder (weights bundled)

from ecgdatakit.processing import clean_ecg

cleaned = clean_ecg(lead)
cleaned = clean_ecg(lead, method="neurokit2")
cleaned = clean_ecg(lead, method="deepfade")
cleaned = clean_ecg(lead, method="deepfade", device="mps")