Add opt-in winsorization for revenue/price outliers

main.py

@@ -14,6 +14,7 @@
     load_and_validate_csv,
     refine_activity_labels,
     optimize_dataframe_memory,
+    winsorize_series,
     create_analysis_config,
     run_full_analysis,
     format_business_report,
@@ -551,12 +552,69 @@ def _run():
             st.warning(msg)
     st.stop()
 
+# ---------------------------------------------------------------------------
+# Winsorize revenue/price outliers - opt-in, applied right after the data is
+# cached (both raw_df and df_ready) and before anything downstream reads
+# 'price' (business insights' AOV/revenue trend/category breakdown, sampling
+# strata, etc.), so a handful of extreme values don't dilute those reports.
+# Caps values rather than dropping rows - see prox.winsorize_series.
+# ---------------------------------------------------------------------------
+st.divider()
+st.header("2. Handle Outliers")
+if "price" not in raw_df.columns:
+    st.caption("No revenue/price column detected - nothing to winsorize.")
+else:
+    winsorize_enabled = st.checkbox(
+        "Winsorize Revenue/Price Outliers", value=False,
+        help=(
+            "Caps extreme values in the revenue/price column instead of "
+            "removing those rows, so a handful of outlier orders don't "
+            "dilute Average Order Value, revenue trend, or category "
+            "revenue breakdown in Business Insights."
+        )
+    )
+    if winsorize_enabled:
+        w_col1, w_col2 = st.columns(2)
+        with w_col1:
+            winsorize_method_label = st.radio(
+                "Method", ["Standard Deviation", "Percentile"], horizontal=True,
+                help=(
+                    "Standard Deviation: caps at mean +/- N standard deviations. "
+                    "Percentile: caps at the Nth/100-Nth percentile band."
+                )
+            )
+        with w_col2:
+            if winsorize_method_label == "Standard Deviation":
+                winsorize_param = st.slider(
+                    "Std deviations", 1.0, 5.0, 3.0, 0.5,
+                    help="Values beyond mean +/- this many standard deviations are capped.",
+                )
+            else:
+                winsorize_param = st.slider(
+                    "Percentile cutoff", 0.5, 10.0, 1.0, 0.5,
+                    help="Caps at this percentile and its mirror (e.g. 1 = 1st/99th percentile).",
+                )
+
+        winsorize_method = "std" if winsorize_method_label == "Standard Deviation" else "percentile"
+        clipped, lower, upper = winsorize_series(raw_df["price"], method=winsorize_method, param=winsorize_param)
+        n_capped = int(((raw_df["price"] < lower) | (raw_df["price"] > upper)).sum())
+
+        raw_df = raw_df.copy()
+        df_ready = df_ready.copy()
+        raw_df["price"] = clipped
+        df_ready["price"] = df_ready["price"].clip(lower, upper)
+
+        if n_capped > 0:
+            st.info(f"Capped {n_capped:,} value(s) to the range [{lower:,.2f}, {upper:,.2f}].")
+        else:
+            st.caption("No values fell outside the winsorization bounds - nothing was capped.")
+
 # ---------------------------------------------------------------------------
 # Data quality check - surfaced before filtering/analysis, so messy data is
 # caught here instead of showing up as a confusing downstream result
 # ---------------------------------------------------------------------------
 st.divider()
-st.header("2. Data Quality Check")
+st.header("3. Data Quality Check")
 data_quality = check_data_quality(raw_df)
 if data_quality["issues"]:
     with st.expander(f"{len(data_quality['issues'])} data quality issue(s) found", expanded=True):
@@ -569,7 +627,7 @@ def _run():
 # Filter events before analysis
 # ---------------------------------------------------------------------------
 st.divider()
-st.header("3. Filter Events")
+st.header("4. Filter Events")
 st.caption(
     "Remove noisy or irrelevant events before analysis, or narrow it down to "
     "just the events you care about. Optional - leave the list empty to "
@@ -646,7 +704,7 @@ def _run():
 # Sampling - opt-in, with a warning above a "large" case-count threshold
 # ---------------------------------------------------------------------------
 st.divider()
-st.header("4. Sampling")
+st.header("5. Sampling")
 enable_sampling = st.checkbox(
     "Enable Sampling", value=False,
     help=(
@@ -1482,5 +1540,5 @@ def _has_priority_value(series: pd.Series) -> bool:
 # Export: Build a Custom PDF Report
 # ---------------------------------------------------------------------------
 st.divider()
-st.header("5. Build a Custom PDF Report")
+st.header("6. Build a Custom PDF Report")
 render_pdf_builder(results, segment_result=st.session_state.get("segment_result"))

prox/__init__.py

@@ -18,6 +18,7 @@
     sample_log_stratified,
     check_trace_length,
     check_data_quality,
+    winsorize_series,
 )
 from .config import CONFIG, create_analysis_config, get_column_mappings
 from .discovery import perform_process_discovery, DISCOVERY_ALGORITHMS
@@ -56,6 +57,7 @@
     "sample_log_stratified",
     "check_trace_length",
     "check_data_quality",
+    "winsorize_series",
     "CONFIG",
     "create_analysis_config",
     "get_column_mappings",

prox/data_manager.py

@@ -617,3 +617,37 @@ def sample_log_stratified(
     sampled_ids = pd.Series(all_ids).sample(n, replace=False).tolist()
     messages.append(f"Random sample: {len(sampled_ids)} cases.")
     return event_log_df[event_log_df['case:concept:name'].isin(sampled_ids)].copy(), messages
+
+
+def winsorize_series(
+    series: pd.Series, method: str = "std", param: float = 3.0
+) -> Tuple[pd.Series, float, float]:
+    """
+    Caps outliers in a numeric Series without dropping rows, so a handful of
+    extreme values (e.g. one enormous order) don't dilute downstream stats
+    like Average Order Value or a revenue trend. Same technique as
+    first-order-engine's ContinuousMetricEngine.winsorize_series, adapted to
+    return a Series (PRoX's DataFrame-in/DataFrame-out convention) instead of
+    a JSON-serializable list.
+
+    method: "std" caps at mean +/- param standard deviations.
+            "percentile" caps at the [param, 100 - param] percentile band
+            (e.g. param=1 -> 1st/99th percentile).
+
+    Returns (clipped_series, lower_bound, upper_bound). An empty or all-NaN
+    series is returned unchanged with bounds of (0.0, 0.0) - nothing to
+    winsorize against.
+    """
+    series_clean = series.dropna()
+    if series_clean.empty:
+        return series.copy(), 0.0, 0.0
+
+    if method == "std":
+        mean_val = series_clean.mean()
+        std_val = series_clean.std()
+        lower = mean_val - (param * std_val)
+        upper = mean_val + (param * std_val)
+    else:  # percentile
+        lower, upper = np.percentile(series_clean, [param, 100.0 - param])
+
+    return series.clip(lower, upper), float(lower), float(upper)

tests/test_data_manager.py

@@ -10,6 +10,7 @@
     optimize_dataframe_memory,
     refine_activity_labels,
     check_data_quality,
+    winsorize_series,
 )
 
 from conftest import make_event_log
@@ -334,3 +335,55 @@ def test_check_data_quality_empty_df_returns_no_issues():
     result = check_data_quality(pd.DataFrame())
     assert result['issues'] == []
     assert result['duplicate_events'] == 0
+
+
+# --- winsorize_series ---
+
+def test_winsorize_series_percentile_caps_a_single_extreme_outlier():
+    series = pd.Series([10.0, 12.0, 11.0, 9.0, 13.0, 10.0, 11.0, 12.0, 9.0, 999999.0])
+    clipped, lower, upper = winsorize_series(series, method='percentile', param=10.0)
+
+    assert clipped.max() == pytest.approx(upper)
+    assert clipped.max() < 999999.0
+    # Every non-outlier value is well inside the band, so only the injected
+    # outlier should actually get capped.
+    assert (series[:-1] == clipped[:-1]).all()
+
+
+def test_winsorize_series_std_caps_at_mean_plus_n_std():
+    series = pd.Series([10.0, 20.0, 30.0, 40.0, 50.0])
+    clipped, lower, upper = winsorize_series(series, method='std', param=1.0)
+
+    mean, std = series.mean(), series.std()
+    assert lower == pytest.approx(mean - std)
+    assert upper == pytest.approx(mean + std)
+    assert clipped.min() >= lower
+    assert clipped.max() <= upper
+
+
+def test_winsorize_series_preserves_nan_positions():
+    series = pd.Series([10.0, None, 30.0, None, 9999.0])
+    clipped, lower, upper = winsorize_series(series, method='percentile', param=10.0)
+
+    assert clipped.isna().tolist() == [False, True, False, True, False]
+
+
+def test_winsorize_series_no_outliers_leaves_values_unchanged():
+    series = pd.Series([10.0, 11.0, 12.0, 13.0, 14.0])
+    clipped, lower, upper = winsorize_series(series, method='std', param=3.0)
+
+    assert (clipped == series).all()
+
+
+def test_winsorize_series_empty_series_returns_zero_bounds():
+    clipped, lower, upper = winsorize_series(pd.Series([], dtype=float))
+    assert clipped.empty
+    assert lower == 0.0
+    assert upper == 0.0
+
+
+def test_winsorize_series_all_nan_returns_zero_bounds():
+    clipped, lower, upper = winsorize_series(pd.Series([None, None], dtype=float))
+    assert clipped.isna().all()
+    assert lower == 0.0
+    assert upper == 0.0