diff --git a/README.md b/README.md index 2d3ba19..66832f8 100644 --- a/README.md +++ b/README.md @@ -233,6 +233,11 @@ The latest execution is compared against that baseline using a robust Z-score. This makes the detector less sensitive to historical outliers than approaches based only on mean and standard deviation. +Propagation scores are normalized to the `0.0–1.0` range. Downstream task +severity is weighted by graph distance with a `0.8` decay per hop, so anomalies +closer to the origin contribute more strongly than anomalies farther along the +same path. + ## Project Status FlowSense is currently in early development. @@ -245,7 +250,6 @@ Planned areas include: - DAG-level analysis models - configurable historical baseline windows -- improved propagation scoring - change-point detection - trend detection - richer CLI reporting diff --git a/src/flowsense/engine/propagation.py b/src/flowsense/engine/propagation.py index 6ebcd34..de87931 100644 --- a/src/flowsense/engine/propagation.py +++ b/src/flowsense/engine/propagation.py @@ -3,6 +3,9 @@ from flowsense.domain import DriftResult, PropagationResult, Severity from flowsense.domain.enums import SEVERITY_SCORE +_HOP_DECAY = 0.8 +_MAX_SEVERITY_SCORE = max(SEVERITY_SCORE.values()) + def _build_reverse_dependencies( dependencies: dict[str, list[str]], @@ -97,6 +100,28 @@ def _find_propagation_paths( return paths +def _calculate_propagation_score( + affected_tasks: list[str], + drift_results: dict[str, DriftResult], +) -> float: + """Return a bounded, distance-weighted score for a propagation path.""" + weighted_severity = 0.0 + total_weight = 0.0 + + for hop, task_id in enumerate(affected_tasks): + weight = _HOP_DECAY**hop + normalized_severity = ( + SEVERITY_SCORE[drift_results[task_id].severity] / _MAX_SEVERITY_SCORE + ) + weighted_severity += normalized_severity * weight + total_weight += weight + + if total_weight == 0.0: + return 0.0 + + return weighted_severity / total_weight + + def analyze_propagation( drift_results: dict[str, DriftResult], dependencies: dict[str, list[str]], @@ -131,14 +156,9 @@ def analyze_propagation( if not affected_tasks: continue - origin_score = SEVERITY_SCORE[drift.severity] - - downstream_scores = [ - SEVERITY_SCORE[drift_results[task].severity] for task in affected_tasks - ] - - propagation_score = sum(downstream_scores) / ( - len(downstream_scores) * origin_score + propagation_score = _calculate_propagation_score( + affected_tasks=affected_tasks, + drift_results=drift_results, ) results.append( diff --git a/tests/test_propagation.py b/tests/test_propagation.py index c44ac98..7727912 100644 --- a/tests/test_propagation.py +++ b/tests/test_propagation.py @@ -2,6 +2,65 @@ from flowsense.engine.propagation import analyze_propagation +def _drift(task_id: str, severity: str) -> DriftResult: + return DriftResult( + task_id=task_id, + baseline=1.0, + current=2.0, + mad=0.1, + robust_z_score=3.0, + deviation_percent=100.0, + severity=severity, + ) + + +def test_propagation_score_is_bounded_when_downstream_is_more_severe() -> None: + drift_results = { + "origin": _drift("origin", "HIGH"), + "critical": _drift("critical", "CRITICAL"), + } + + results = analyze_propagation( + drift_results, + {"origin": ["critical"], "critical": []}, + ) + + assert results[0].propagation_score == 1.0 + + +def test_propagation_score_gives_closer_tasks_more_weight() -> None: + closer_critical_results = analyze_propagation( + { + "origin": _drift("origin", "CRITICAL"), + "critical": _drift("critical", "CRITICAL"), + "medium": _drift("medium", "MEDIUM"), + }, + { + "origin": ["critical"], + "critical": ["medium"], + "medium": [], + }, + ) + farther_critical_results = analyze_propagation( + { + "origin": _drift("origin", "CRITICAL"), + "medium": _drift("medium", "MEDIUM"), + "critical": _drift("critical", "CRITICAL"), + }, + { + "origin": ["medium"], + "medium": ["critical"], + "critical": [], + }, + ) + + closer_critical = closer_critical_results[0].propagation_score + farther_critical = farther_critical_results[0].propagation_score + + assert closer_critical > farther_critical + assert 0.0 <= farther_critical <= closer_critical <= 1.0 + + def test_detects_downstream_propagation() -> None: drift_results = { "extract": DriftResult(