Online mode

jitterbug.streaming.online_analyzer

Online (streaming) Jitterbug pipeline.

The sequential pipeline is retrospective twice over: the offline Bayesian detector sees the whole series, and the verdict for period i needs change point i + 1 to close the period. This module runs the same decision rule one RTT sample at a time:

1. Minimum-RTT bins close when a sample from a later bin arrives (causal binning).
2. Each closed bin feeds a Bayesian online change point detector (Adams & MacKay 2007, bayesian_changepoint_detection.streaming.OnlineChangepointDetector). A change point is declared when the MAP run length drops (default), or when the posterior that a segment started lag bins ago (lag) or within the last lag bins (window) exceeds a threshold. On the paper dataset the MAP rule recovers 14 of the 15 reference congestion periods with no spurious one, like the offline pipeline, and reports congestion onsets one bin (15 min) after they start. The fixed-delay rules recover 5 to 12 (lag) or 5 to 15 (window) of the 15 depending on the setting but always with 3 to 6 spurious congested periods, because the posterior mass of a change spreads over several run lengths (see docs/ONLINE_MODE.md).
3. The stream start opens the baseline period. When a change point opens a new period, a provisional verdict is emitted as soon as the open period holds min_period_samples jitter samples. When the next change point closes the period, the final verdict is emitted with the full period, which is exactly what the sequential pipeline computes.

Both jitter methods work online. The Kolmogorov-Smirnov test uses consecutive RTT differences and is causal as it stands. Jitter dispersion uses the trailing version of the offline filters (JitterAnalyzer.compute_causal_jitter_dispersion): same window lengths, each ending at the bin it describes, so its series is the offline one delayed by JitterAnalyzer.causal_dispersion_delay bins (6 bins, 1.5 h, by default). A dispersion value reflects the new period alone only after that delay, so a provisional verdict with dispersion waits until the open period holds delay + moving_average_order dispersion values (12 bins, 3 h, by default); with fewer, the provisional verdict is still dominated by the previous period's jitter (11 of 31 flipped on the paper dataset with 6, none with 12).

Settings live in JitterbugConfig.streaming (models.config.StreamingConfig); the bin width, latency jump threshold, significance level and device come from the shared sections of JitterbugConfig.

StreamingEvent dataclass

Something the online pipeline decided while consuming the stream.

emitted_at instance-attribute

Epoch of the sample whose arrival triggered the event.

start_epoch instance-attribute

Change point epoch, or start of the period a verdict is about.

jitter_method = None class-attribute instance-attribute

Jitter test behind has_jitter; the KS fields are set for ks_test only.

jitter_metric = None class-attribute instance-attribute

The jitter test's metric: the KS statistic, or the dispersion change.

delay property

Seconds between the event's reference time and its emission.

to_dict()

Return the event as a plain dictionary (JSON-serializable).

OnlineJitterbug

Streaming congestion inference: feed RTT samples, get events.

Samples must arrive in non-decreasing epoch order; a sample earlier than the last one accepted is dropped (logged at DEBUG), whatever bin it falls in. Bins are indexed by the samples that arrive, so a gap in the stream does not produce empty bins: the detector sees the bins before and after a gap as consecutive, as the offline pipeline does after dropna in RTTDataset.compute_minimum_intervals.

Parameters:

Name	Type	Description	Default
config	JitterbugConfig	Full configuration; config.streaming holds the online-specific settings.	None

push(epoch, rtt)

Consume one RTT sample and return the events it triggered.

Parameters:

Name	Type	Description	Default
epoch	float	Unix timestamp of the sample, seconds.	required
rtt	float	Round-trip time in milliseconds.	required

flush()

Close the open bin at the end of a finite stream.

verdict_event(verdict, stage, emitted_at, start_epoch, end_epoch)

Wrap a PeriodVerdict as a verdict event.

jitterbug.streaming.window_backend

Sliding-window back end: rerun the offline pipeline as the stream grows.

The baseline the incremental detector is measured against. Every rerun_every_bins closed minimum-RTT bins, the sequential pipeline (JitterbugAnalyzer, with the configured detector and jitter method) runs on the trailing window_hours of raw samples. A change point is emitted once the detector has placed one within TOLERANCE_BINS of the same bin in stable_runs consecutive reruns; change points are emitted in time order and never retracted. Final verdicts form a contiguous chain: the next one is emitted once the offline pipeline has judged the period at the last emitted boundary identically (same verdict, boundaries within the tolerance) in stable_runs consecutive reruns and the change point that closes it has been emitted; its boundaries are those emitted change points, so change points and verdicts never disagree. If the window moves past a period before it stabilizes, that period is skipped with a warning and the chain resumes at the next emitted change point. A change point that only stabilizes after a later one was emitted is dropped (time order is kept), which can leave the chain waiting for the next emitted boundary for up to a window length. The open period after the last emitted change point gets a provisional verdict from the shared two-period rule, with the configured jitter method (trailing filters for dispersion), once it holds min_period_samples jitter samples, as in the incremental back end.

The window start opens the baseline period, as the stream start does in the incremental back end: when the detector reports no change point within one bin of the first bin, one is inserted there, so the first period after the window start can be judged (the sequential pipeline needs three change points per verdict). Once the window has moved past the stream start, change points and periods touching its left edge are ignored: they are cut by the window and were emitted long before.

TOLERANCE_BINS = 2 module-attribute

Two boundaries this many bins apart or closer are the same boundary (30 min at 15-min bins).

The offline detector moves a change point by a bin or two as the window slides; the sequential pipeline keeps change points at least change_point_detection.min_time_elapsed apart (two bins by default), so a larger tolerance would merge real boundaries.

SlidingWindowJitterbug

Streaming congestion inference by rerunning the offline pipeline on a window.

Same interface as OnlineJitterbug: push(epoch, rtt) returns the events the sample triggered, flush() closes the open bin and reruns once, events holds everything emitted. Samples must arrive in non-decreasing epoch order.

Parameters:

Name	Type	Description	Default
config	JitterbugConfig	streaming.window_hours, streaming.rerun_every_bins and streaming.stable_runs drive the back end; the detector and jitter method are those of the sequential mode (analysis_mode is forced to sequential).	None

push(epoch, rtt)

Consume one RTT sample (seconds, milliseconds) and return the events it triggered.

flush()

Close the open bin and rerun once, whatever the cadence.

jitterbug.streaming.verdict

The two-period decision rule shared by the online back ends.

Given the minimum-RTT bins and the jitter observations of a period and of the period before it, decide whether the period is congested the way the sequential pipeline does: a latency jump is a rise of the mean minimum RTT above latency_jump.threshold; a jitter change is, with jitter_analysis.method: ks_test, a significant Kolmogorov-Smirnov test on the raw jitter samples with a statistic of at least streaming.min_ks_statistic, and with jitter_dispersion a rise of the mean dispersion (trailing filters, see JitterAnalyzer.compute_causal_jitter_dispersion) above jitter_analysis.threshold; the congestion state carries over: both → congested, no jump → not congested, jump without jitter change → unchanged.

PeriodVerdict dataclass

Outcome of two_period_verdict.

two_period_verdict(prev_bins, cur_bins, prev_jitter, cur_jitter, previous_state, config)

Classify a period against the one before it.

Parameters:

Name	Type	Description	Default
prev_bins	ndarray	Minimum RTT per bin of the previous and the current period (ms).	required
cur_bins	ndarray	Minimum RTT per bin of the previous and the current period (ms).	required
prev_jitter	ndarray	Jitter observations of the previous and the current period: consecutive RTT differences (ms) for ks_test, causal dispersion values for jitter_dispersion.	required
cur_jitter	ndarray	Jitter observations of the previous and the current period: consecutive RTT differences (ms) for ks_test, causal dispersion values for jitter_dispersion.	required
previous_state	bool	Congestion state after the previous period's final verdict.	required
config	JitterbugConfig	latency_jump.threshold, jitter_analysis.method, jitter_analysis.threshold, jitter_analysis.significance_level and streaming.min_ks_statistic are used.	required

Returns:

Type	Description
PeriodVerdict	The decision and the statistics behind it. Tests that could not run (too few observations) leave their fields None and the state unchanged.

jitterbug.streaming.replay

Replay a finite RTT dataset through the online pipeline and score the result.

replay feeds every sample of an RTTDataset to the configured online back end in timestamp order and returns the events. score summarizes them: change points, periods, detection delays, provisional-to-final agreement and, when a reference file is given, agreement with it using the metric of tests/test_paper_regression.py (a reference congested period counts as recovered when one of ours overlaps more than half of it) and how many reference period boundaries have a change point within 30 min.

Delays are measured from the change point to the sample that reported it. Onset is the change point that opens a congested final period, return the one that closes it.

replay(dataset, config=None)

Feed dataset sample by sample to the online pipeline.

Parameters:

Name	Type	Description	Default
dataset	RTTDataset	Samples to replay; they are sorted by epoch (stable) before being fed.	required
config	JitterbugConfig	Configuration; config.streaming holds the online-specific settings, including which back end runs (backend).	None

Returns:

Type	Description
list[StreamingEvent]	Every change point and verdict the pipeline emitted, in order, after flush.

load_reference(path)

Congested periods (start, end) of a reference CSV with starts,ends,congestion.

score(events, reference=None)

Summarize replay events.

Parameters:

Name	Type	Description	Default
events	list[StreamingEvent]	Output of replay.	required
reference	Path	CSV with starts,ends,congestion columns (the paper's expected_results). Adds recovered, reference_congested and spurious.	None

Returns:

Type	Description
dict	Counts, delays in minutes and provisional-to-final agreement.

events_to_json(events, path)

Write events as a JSON array of plain dictionaries.