make outlier detection trend aware

Author: shrimbly

sharp_frames/selection_methods.py

@@ -1,7 +1,7 @@
 """Canonical frame-selection algorithms shared by the CLI and TUI."""
 
 from collections import OrderedDict
-from math import log1p
+from math import expm1, log1p
 from statistics import median
 from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
 
@@ -14,8 +14,8 @@
 OUTLIER_DEFAULT_WINDOW_SIZE = 15
 OUTLIER_DEFAULT_SENSITIVITY = 60
 OUTLIER_MIN_WINDOW_SIZE = 5
-OUTLIER_MIN_RELATIVE_DROP = 0.05
-OUTLIER_MAX_RELATIVE_DROP = 0.30
+OUTLIER_MIN_RELATIVE_DROP = 0.005
+OUTLIER_MAX_RELATIVE_DROP = 0.08
 OUTLIER_MIN_ROBUST_THRESHOLD = 1.5
 
 
@@ -204,6 +204,20 @@ def _outlier_positions(
         for _, frame in group
     ]
     scores = [log1p(score) for score in raw_scores]
+    dip_magnitudes = [0.0] * len(scores)
+    relative_drops = [0.0] * len(scores)
+    for position in range(1, len(scores) - 1):
+        # A local dip must sit below both adjacent frames. Using the lower
+        # neighbor as the baseline prevents one unusually sharp neighbor from
+        # making an otherwise ordinary frame look like an outlier.
+        expected_score = min(scores[position - 1], scores[position + 1])
+        dip_magnitudes[position] = max(0.0, expected_score - scores[position])
+        expected_raw_score = expm1(expected_score)
+        if expected_raw_score > 0:
+            relative_drops[position] = max(
+                0.0,
+                (expected_raw_score - raw_scores[position]) / expected_raw_score,
+            )
 
     actual_window_size = max(OUTLIER_MIN_WINDOW_SIZE, window_size)
     if actual_window_size % 2 == 0:
@@ -221,38 +235,33 @@ def _outlier_positions(
     )
     outliers: Set[int] = set()
 
-    for position, current_score in enumerate(scores):
+    for position, dip_magnitude in enumerate(dip_magnitudes):
+        if position == 0 or position == len(scores) - 1:
+            continue
         window_start = max(0, position - half_window)
         window_end = min(len(group), position + half_window + 1)
-        neighbor_scores = scores[window_start:position] + scores[position + 1 : window_end]
+        neighbor_dips = (
+            dip_magnitudes[window_start:position]
+            + dip_magnitudes[position + 1 : window_end]
+        )
         required_neighbors = max(2, min_neighbors)
-        if len(neighbor_scores) < required_neighbors:
+        if len(neighbor_dips) < required_neighbors:
             continue
-        neighbor_median = median(neighbor_scores)
-        raw_neighbor_scores = (
-            raw_scores[window_start:position]
-            + raw_scores[position + 1 : window_end]
-        )
-        raw_neighbor_median = median(raw_neighbor_scores)
+        neighbor_median = median(neighbor_dips)
         absolute_deviations = [
-            abs(score - neighbor_median) for score in neighbor_scores
+            abs(neighbor_dip - neighbor_median) for neighbor_dip in neighbor_dips
         ]
         median_absolute_deviation = median(absolute_deviations)
         robust_scale = median_absolute_deviation * 1.4826
         if robust_scale == 0:
             robust_scale = max(abs(neighbor_median) * 0.01, 1e-9)
 
-        deficit = neighbor_median - current_score
-        robust_deficit = deficit / robust_scale
-        relative_drop = (
-            (raw_neighbor_median - raw_scores[position]) / raw_neighbor_median
-            if raw_neighbor_median > 0
-            else 0.0
-        )
+        excess_dip = dip_magnitude - neighbor_median
+        robust_excess = excess_dip / robust_scale
         if (
-            deficit > 0
-            and relative_drop >= minimum_relative_drop
-            and robust_deficit > robust_threshold
+            excess_dip > 0
+            and relative_drops[position] >= minimum_relative_drop
+            and robust_excess > robust_threshold
         ):
             outliers.add(position)
     return outliers

tests/test_selection_quality.py

@@ -154,7 +154,7 @@ def test_labeled_blur_probe_tracks_rejection_precision_and_recall():
 
 def test_default_outlier_detection_requires_a_meaningful_relative_drop():
     scores = [100.0] * 15
-    scores[5] = 90.0
+    scores[5] = 98.0
     scores[9] = 75.0
     frames = [
         {"index": index, "sharpnessScore": score}
@@ -187,3 +187,48 @@ def rejected_dip(baseline, dip):
 
     assert rejected_dip(10.0, 7.0) is True
     assert rejected_dip(1_000.0, 700.0) is True
+
+
+def test_smooth_trough_is_not_mistaken_for_an_outlier_cluster():
+    scores = [
+        162.32, 162.21, 161.43, 159.67, 160.00, 159.41, 159.17,
+        157.90, 158.01, 160.00, 160.01, 160.47, 160.46, 160.53,
+        160.82, 160.60, 160.10, 159.10, 158.01, 157.06, 155.20,
+        152.87, 149.68, 148.97, 150.22, 153.09, 156.55, 158.64,
+        159.78, 160.12, 160.68, 161.82, 162.06, 162.84, 162.69,
+        163.60, 165.84, 165.63, 164.79, 163.05,
+    ]
+    frames = [
+        {"index": index, "sharpnessScore": score}
+        for index, score in enumerate(scores)
+    ]
+
+    assessed = select_outlier_removal_frames_core(
+        frames,
+        window_size=31,
+        sensitivity=100,
+    )
+
+    assert all(assessed[position]["selected"] for position in (22, 23, 24))
+
+
+def test_abrupt_dip_in_a_broad_trend_is_detected_at_maximum_sensitivity():
+    scores = [
+        191.30, 190.19, 191.34, 189.59, 188.12, 185.22, 180.47,
+        177.02, 173.26, 170.84, 168.32, 165.99, 163.30, 162.04,
+        165.47, 166.67, 169.98, 173.32, 172.83, 169.09, 169.62,
+        169.61, 168.71, 167.32, 164.46, 162.95, 162.07, 160.60,
+        158.71, 157.39, 155.58,
+    ]
+    frames = [
+        {"index": index, "sharpnessScore": score}
+        for index, score in enumerate(scores)
+    ]
+
+    assessed = select_outlier_removal_frames_core(
+        frames,
+        window_size=31,
+        sensitivity=100,
+    )
+
+    assert assessed[13]["selected"] is False