src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density

EpisodeDetector — KDE-based episode boundary detection.

Detects contiguous periods in a historical event log where the estimated event rate exceeds a fraction of the query incident’s reference density.

KDE implementation uses an event-centric Gaussian kernel: each event contributes only to grid points within 4σ of its timestamp. This is O(N × support_window / grid_res) instead of O(N × G), which is essential for long historical windows with fine grid resolution.

Attributes

_log
_SQRT_2PI
_KDE_SUPPORT_SIGMAS

Classes

EpisodeDetector

Detects episode boundaries in a continuous historical event log using

Functions

_kde_evaluate(t_seconds, grid, bw, grid_res)	Evaluates a Gaussian KDE at each grid point.
_extract_contiguous_regions(mask, grid, t_epoch, grid_res)	Extracts (start, end) datetime pairs for every contiguous True run in mask.
_merge_overlapping(boundaries)	Merges overlapping or touching (start, end) intervals.

Module Contents

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density._log

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density._SQRT_2PI: float

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density._KDE_SUPPORT_SIGMAS: float = 4.0

class src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density.EpisodeDetector(config)

Detects episode boundaries in a continuous historical event log using kernel density estimation (KDE) over event timestamps.

The detection threshold is relative to the query incident density, making the method self-calibrating: a busier query requires historically busier windows to qualify as matching episodes.

Pipeline (per detect() call):

1. Convert event timestamps to float seconds since earliest event.

2. Evaluate Gaussian KDE on a fine time grid.

3. Threshold: mask = KDE(t) >= delta * rho_query.

4. Extract contiguous masked regions as raw episode boundaries.

5. Apply beta buffer expansion to each boundary.

6. Merge overlapping expanded boundaries.

7. Discard episodes shorter than query_duration / 10.

Parameters:

config (src.dackar.RCA.log_pattern_recognition.rca_pattern_search.config.SearchConfig)

config

compute_reference_density(query_events, window_start, window_end)

Computes the reference event density for the query incident.

rho_query = N_query / D_query

N_query: events with timestamp_start in [window_start, window_end]. D_query: (window_end - window_start).total_seconds()

All sources contribute equally. Returns 0.0 if duration <= 0.

Parameters:

• query_events (list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent])

• window_start (datetime.datetime)

• window_end (datetime.datetime)

Return type:

float

detect(historical_events, rho_query, query_duration)

Detects episode boundaries in the historical event log.

Parameters:

• historical_events (list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]) – Full flat event list, all sources merged. No episode_id expected at this stage.

• rho_query (float) – Reference density from compute_reference_density(). Units: events per second.

• query_duration (float) – D_query in seconds. Used for KDE bandwidth and minimum episode duration filter.

Returns:

List of (episode_start, episode_end) tuples, sorted ascending. These are expanded boundaries (beta already applied) ready for fingerprinting. Empty list if no qualifying episodes found.

Return type:

list[tuple[datetime.datetime, datetime.datetime]]

bandwidth_scan(historical_events, rho_query, query_duration, bandwidths=None)

Multi-scale diagnostic: counts detected episodes at different bandwidths.

Helps operators validate episode segmentation by showing how many episodes are detected when the smoothing scale varies. Useful when query timescale may not match historical episode timescales (e.g., fast transient query for slow degradation history, or vice versa).

Parameters:

• historical_events (list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]) – Full flat event list, all sources merged.

• rho_query (float) – Reference density from compute_reference_density(). Units: events per second.

• query_duration (float) – D_query in seconds.

• bandwidths (Optional[list[float]]) – Explicit bandwidth list in seconds. If None, defaults to [D/32, D/16, D/8, D/4, D/2, D, 2D, 4D] for broad coverage.

Returns:

dict mapping bandwidth (float, seconds) to episode count (int), sorted by bandwidth ascending.

Return type:

dict[float, int]

_run_detection(t_seconds, t_epoch, rho_query, query_duration, bw)

Core detection pipeline: KDE → threshold → extract → expand → merge → filter.

Called by detect() and bandwidth_scan() to avoid code duplication.

Parameters:

• t_seconds (numpy.ndarray) – Event timestamps as float seconds since t_epoch.

• t_epoch (datetime.datetime) – Reference time (datetime).

• rho_query (float) – Reference density (events/second).

• query_duration (float) – Duration of query window in seconds.

• bw (float) – Bandwidth in seconds (already resolved, > 0).

Returns:

Expanded, merged, filtered episode boundaries.

Return type:

list[tuple[datetime.datetime, datetime.datetime]]

assign_episode_ids(historical_events, episode_boundaries)

Assigns episode_id to each historical event based on detected boundaries.

An event is assigned to the episode whose expanded boundary contains its timestamp_start. Events outside all boundaries retain episode_id = None (background noise).

If an event falls within multiple boundaries (should not occur after merging, handled defensively), it is assigned to the first match and a WARNING is logged.

Episode ids: “EP_{asset_id}_{index:05d}” where asset_id is the dominant asset among events in that episode.

Parameters:

• historical_events (list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]) – Full flat event list.

• episode_boundaries (list[tuple[datetime.datetime, datetime.datetime]]) – Output of detect(), sorted (start, end) tuples.

Returns:

New list of UnifiedEvents with episode_id populated. Original list is not mutated.

Return type:

list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density._kde_evaluate(t_seconds, grid, bw, grid_res)

Evaluates a Gaussian KDE at each grid point.

Uses an event-centric approach: each event contributes only to grid points within _KDE_SUPPORT_SIGMAS * bw seconds, limiting work to O(N × support_window / grid_res) instead of O(N × G).

Returns rho_hist in events per second:

rho_hist(t) = Σ_i (1 / (bw √2π)) · exp(-½ · ((t − tᵢ) / bw)²)

Parameters:

• t_seconds (numpy.ndarray)

• grid (numpy.ndarray)

• bw (float)

• grid_res (float)

Return type:

numpy.ndarray

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density._extract_contiguous_regions(mask, grid, t_epoch, grid_res)

Extracts (start, end) datetime pairs for every contiguous True run in mask.

Pads the mask with False on both ends to detect edges at array boundaries. Uses the last grid point of each run as t_end. If a run is a single grid point, t_end is set to t_start + grid_res to give it a non-zero duration.

Parameters:

• mask (numpy.ndarray)

• grid (numpy.ndarray)

• t_epoch (datetime.datetime)

• grid_res (float)

Return type:

list[tuple[datetime.datetime, datetime.datetime]]

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.density._merge_overlapping(boundaries)

Merges overlapping or touching (start, end) intervals.

Input order does not matter. Returns intervals sorted by start ascending.

Parameters:

boundaries (list[tuple[datetime.datetime, datetime.datetime]])

Return type:

list[tuple[datetime.datetime, datetime.datetime]]