fix(spc): Cp/Cpk usa sigma within (MRbar/1.128), Pp/Ppk la stdev campionaria (#19)

Prima Cp/Cpk usava stdev campionaria e Pp/Ppk la population stdev: stesso
indice due volte, proporzionale. Ora Cp/Cpk (capability a breve termine)
usa sigma_within dalla moving range dei valori consecutivi in ordine
cronologico; Pp/Ppk (performance complessiva) resta sulla stdev campionaria
(n-1). MRbar == 0 -> Cp/Cpk None. Stessa sigma_within per UCL/LCL della
carta di controllo a valori individuali (mean +/- 2.66*MRbar), non piu la
stdev globale che allargava i limiti e nascondeva la deriva.

Aggiunto campo opzionale std_dev_within a CapabilityData.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

src/backend/models/api/statistics.py

@@ -50,7 +50,8 @@ class CapabilityData(BaseModel):
     pp: Optional[float] = None
     ppk: Optional[float] = None
     mean: float
-    std_dev: float
+    std_dev: float  # overall sample std dev (n-1), used for Pp/Ppk
+    std_dev_within: Optional[float] = None  # MRbar/1.128, used for Cp/Cpk
     n: int
     utl: Optional[float] = None
     ltl: Optional[float] = None

src/backend/services/spc_service.py

@@ -55,6 +55,27 @@ def compute_summary(
     )
 
 
+def _sigma_within(values: list[float]) -> float | None:
+    """Estimate within-subgroup sigma from the moving range of individuals.
+
+    sigma_within = MRbar / d2, with d2 = 1.128 (moving range of consecutive
+    pairs, i.e. subgroup size 2). `values` must be in chronological order
+    (the order in which measurements arrived) — the moving range is not
+    meaningful on a re-sorted series.
+
+    Returns None when fewer than 2 values are available, or when MRbar == 0
+    (no variation between consecutive points: the index would be undefined).
+    """
+    n = len(values)
+    if n < 2:
+        return None
+    moving_ranges = [abs(values[i] - values[i - 1]) for i in range(1, n)]
+    mr_bar = stats.mean(moving_ranges)
+    if mr_bar == 0:
+        return None
+    return mr_bar / 1.128
+
+
 def compute_capability(
     values: list[float],
     utl: float | None,
@@ -63,12 +84,18 @@ def compute_capability(
 ) -> CapabilityData:
     """Compute capability indices Cp, Cpk, Pp, Ppk.
 
+    `values` must be in chronological order (arrival order), since Cp/Cpk are
+    derived from the moving range between consecutive points:
+
+    - sigma_within = MRbar / 1.128 (MRbar = mean of |x_i - x_(i-1)|)
     - Cp  = (UTL - LTL) / (6 * sigma_within)
-    - Cpk = min((UTL - mean) / (3 * sigma), (mean - LTL) / (3 * sigma))
+    - Cpk = min((UTL - mean) / (3 * sigma_within), (mean - LTL) / (3 * sigma_within))
-    - Pp/Ppk: same formulas using population std dev (same data, no subgrouping)
+    - Pp/Ppk: same formulas using the overall sample std dev (n-1, `stats.stdev`)
+      of the whole series — this is the "process performance" pair and is
+      allowed to be wider than Cp/Cpk when there is drift between subgroups.
 
     Args:
-        values: Measured values.
+        values: Measured values, in chronological order.
         utl: Upper Tolerance Limit (None if not defined).
         ltl: Lower Tolerance Limit (None if not defined).
         nominal: Nominal value (None if not defined).
@@ -81,39 +108,39 @@ def compute_capability(
         return CapabilityData(
             cp=None, cpk=None, pp=None, ppk=None,
             mean=values[0] if n == 1 else 0.0,
-            std_dev=0.0, n=n,
+            std_dev=0.0, std_dev_within=None, n=n,
             utl=utl, ltl=ltl, nominal=nominal,
         )
 
     mean = stats.mean(values)
-    # Population std dev for Pp/Ppk
+    # Overall sample std dev (n-1) for Pp/Ppk
-    std_dev_pop = stats.pstdev(values)
-    # Sample std dev for Cp/Cpk
     std_dev_sample = stats.stdev(values)
+    # Within-subgroup sigma (moving range) for Cp/Cpk
+    sigma_within = _sigma_within(values)
 
     cp = cpk = pp = ppk = None
 
-    if utl is not None and ltl is not None and std_dev_sample > 0:
+    if sigma_within is not None:
-        cp = round((utl - ltl) / (6 * std_dev_sample), 4)
+        if utl is not None and ltl is not None:
+            cp = round((utl - ltl) / (6 * sigma_within), 4)
 
-    if std_dev_sample > 0:
         cpk_values = []
         if utl is not None:
-            cpk_values.append((utl - mean) / (3 * std_dev_sample))
+            cpk_values.append((utl - mean) / (3 * sigma_within))
         if ltl is not None:
-            cpk_values.append((mean - ltl) / (3 * std_dev_sample))
+            cpk_values.append((mean - ltl) / (3 * sigma_within))
         if cpk_values:
             cpk = round(min(cpk_values), 4)
 
-    if utl is not None and ltl is not None and std_dev_pop > 0:
+    if std_dev_sample > 0:
-        pp = round((utl - ltl) / (6 * std_dev_pop), 4)
+        if utl is not None and ltl is not None:
+            pp = round((utl - ltl) / (6 * std_dev_sample), 4)
 
-    if std_dev_pop > 0:
         ppk_values = []
         if utl is not None:
-            ppk_values.append((utl - mean) / (3 * std_dev_pop))
+            ppk_values.append((utl - mean) / (3 * std_dev_sample))
         if ltl is not None:
-            ppk_values.append((mean - ltl) / (3 * std_dev_pop))
+            ppk_values.append((mean - ltl) / (3 * std_dev_sample))
         if ppk_values:
             ppk = round(min(ppk_values), 4)
 
@@ -121,6 +148,7 @@ def compute_capability(
         cp=cp, cpk=cpk, pp=pp, ppk=ppk,
         mean=round(mean, 6),
         std_dev=round(std_dev_sample, 6),
+        std_dev_within=round(sigma_within, 6) if sigma_within is not None else None,
         n=n,
         utl=utl, ltl=ltl, nominal=nominal,
     )
@@ -135,10 +163,12 @@ def compute_control_chart(
     ltl: float | None,
     nominal: float | None,
 ) -> ControlChartData:
-    """Compute control chart data with UCL/LCL and out-of-control detection.
+    """Compute control chart data (individuals chart) with UCL/LCL and
+    out-of-control detection.
 
-    UCL = mean + 3*sigma
+    UCL = mean + 3*sigma_within = mean + 2.66*MRbar
-    LCL = mean - 3*sigma
+    LCL = mean - 3*sigma_within = mean - 2.66*MRbar
+    with sigma_within = MRbar/1.128 (see `_sigma_within`).
     Out-of-control: points outside UCL/LCL.
 
     Args:
@@ -160,10 +190,8 @@ def compute_control_chart(
 
     mean = stats.mean(values)
 
-    if n >= 2:
+    sigma_within = _sigma_within(values)
-        sigma = stats.stdev(values)
+    sigma = sigma_within if sigma_within is not None else 0.0
-    else:
-        sigma = 0.0
 
     ucl = mean + 3 * sigma
     lcl = mean - 3 * sigma

src/backend/tests/test_spc_service.py

@@ -0,0 +1,123 @@
+"""Unit tests for spc_service.py: Cp/Cpk (within, moving-range) vs Pp/Ppk (overall).
+
+Expected values are computed in each test from the formulas themselves
+(not copied from the implementation's output), so a regression in the
+formula would also break the test's own expectation derivation... except
+we compute them independently here with plain stdlib code, mirroring the
+spec in gitea:Adriano/TieMeasureFlow#19.
+"""
+import statistics as stats
+from datetime import datetime, timedelta
+
+from src.backend.services.spc_service import compute_capability, compute_control_chart
+
+
+def _expected_sigma_within(values: list[float]) -> float | None:
+    """Independent re-implementation of MRbar/1.128 for test expectations."""
+    if len(values) < 2:
+        return None
+    mr = [abs(values[i] - values[i - 1]) for i in range(1, len(values))]
+    mr_bar = stats.mean(mr)
+    if mr_bar == 0:
+        return None
+    return mr_bar / 1.128
+
+
+# 13 values around 10.00, then 12 values around 10.08: a level shift that
+# a global (overall) std dev smears, but that stays invisible to the
+# consecutive-point moving range within each stable block.
+_SHIFTED_SERIES = (
+    [10.00, 10.01, 9.99, 10.02, 9.98, 10.01, 9.99, 10.00, 10.02, 9.98, 10.01, 9.99, 10.00]
+    + [10.08, 10.09, 10.07, 10.08, 10.09, 10.07, 10.08, 10.09, 10.07, 10.08, 10.09, 10.07]
+)
+
+
+class TestComputeCapability:
+    def test_cp_uses_sigma_within_pp_uses_overall_sample_stdev(self):
+        utl, ltl = 10.2, 9.8
+        result = compute_capability(_SHIFTED_SERIES, utl, ltl, nominal=10.0)
+
+        sample_std = stats.stdev(_SHIFTED_SERIES)
+        sigma_within = _expected_sigma_within(_SHIFTED_SERIES)
+        assert sigma_within is not None
+
+        expected_cp = round((utl - ltl) / (6 * sigma_within), 4)
+        expected_pp = round((utl - ltl) / (6 * sample_std), 4)
+
+        assert result.cp == expected_cp
+        assert result.pp == expected_pp
+        # Cp (within, tight) must differ from Pp (overall, widened by the
+        # level shift) — this is exactly the drift #19 says must surface.
+        assert result.cp != result.pp
+        assert result.cp > result.pp
+        assert result.std_dev == round(sample_std, 6)
+        assert result.std_dev_within == round(sigma_within, 6)
+
+    def test_constant_series_returns_none_indices(self):
+        values = [10.0] * 10
+        result = compute_capability(values, 10.2, 9.8, 10.0)
+        assert result.cp is None
+        assert result.cpk is None
+        assert result.pp is None
+        assert result.ppk is None
+        assert result.std_dev_within is None
+
+    def test_n_less_than_2_returns_none(self):
+        result = compute_capability([10.0], 10.2, 9.8, 10.0)
+        assert result.cp is None
+        assert result.cpk is None
+        assert result.pp is None
+        assert result.ppk is None
+        assert result.n == 1
+
+    def test_one_sided_tolerance_only_cpk(self):
+        values = [10.0, 10.01, 9.99, 10.02, 9.98, 10.01, 9.99, 10.0]
+        result = compute_capability(values, utl=10.2, ltl=None, nominal=10.0)
+        assert result.cp is None  # both UTL and LTL required
+        assert result.pp is None
+        assert result.cpk is not None
+        assert result.ppk is not None
+
+
+class TestComputeControlChart:
+    def test_ucl_lcl_use_sigma_within_not_overall_stdev(self):
+        base = datetime(2026, 1, 1)
+        timestamps = [base + timedelta(minutes=i) for i in range(len(_SHIFTED_SERIES))]
+
+        result = compute_control_chart(
+            _SHIFTED_SERIES, timestamps, 10.2, None, None, 9.8, 10.0
+        )
+
+        mean = stats.mean(_SHIFTED_SERIES)
+        sigma_within = _expected_sigma_within(_SHIFTED_SERIES)
+        expected_ucl = round(mean + 3 * sigma_within, 6)
+        expected_lcl = round(mean - 3 * sigma_within, 6)
+
+        assert result.ucl == expected_ucl
+        assert result.lcl == expected_lcl
+
+        # Points beyond the within-sigma limits must be flagged: the level
+        # shift is detectable this way but was invisible with the overall
+        # sample std dev (limits too wide to ever trip).
+        assert result.out_of_control != []
+        overall_std = stats.stdev(_SHIFTED_SERIES)
+        overall_ucl = mean + 3 * overall_std
+        overall_lcl = mean - 3 * overall_std
+        overall_ooc = [
+            i for i, v in enumerate(_SHIFTED_SERIES) if v > overall_ucl or v < overall_lcl
+        ]
+        assert overall_ooc == []
+
+    def test_constant_series_zero_limits(self):
+        values = [10.0] * 6
+        timestamps = [datetime(2026, 1, 1) + timedelta(minutes=i) for i in range(6)]
+        result = compute_control_chart(values, timestamps, 10.2, None, None, 9.8, 10.0)
+        assert result.ucl == 10.0
+        assert result.lcl == 10.0
+        assert result.out_of_control == []
+
+    def test_single_value_zero_limits(self):
+        timestamps = [datetime(2026, 1, 1)]
+        result = compute_control_chart([10.0], timestamps, 10.2, None, None, 9.8, 10.0)
+        assert result.ucl == 10.0
+        assert result.lcl == 10.0