fix(kpi): per-metric bucket reduction — Rate series sum per bucket instead of keeping one delta (plan R2-04 T5)

src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiSeriesBucketer.cs

@@ -3,8 +3,14 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
 /// <summary>
 /// Pure, deterministic downsampling helper for KPI series charting ("KPI History &amp; Trends").
 /// Reduces a raw <see cref="KpiSeriesPoint"/> series to at most <c>maxPoints</c> points using
-/// last-value-per-bucket / gauge semantics — suitable for step/area charts where the most
+/// <em>per-metric</em> bucket reduction driven by <see cref="KpiRollupAggregation"/>:
-/// recent value in a window best represents that window.
+/// <see cref="KpiRollupAggregation.Gauge"/> (the default) keeps the <strong>last value</strong>
+/// in each bucket — suitable for step/area charts where the most recent reading best represents
+/// the window; <see cref="KpiRollupAggregation.Rate"/> <strong>sums per bucket</strong> — each
+/// raw point of a Rate series is a per-interval delta, so the bucket's true total is the sum of
+/// its deltas. Summing on both the raw-minute and hourly-rollup paths charts the same unit
+/// ("events per chart bucket") on both sides of the raw/rollup routing boundary, eliminating the
+/// ~60× magnitude jump at that boundary (arch-review 04 round 2, R3).
 /// </summary>
 public static class KpiSeriesBucketer
 {
@@ -27,12 +33,20 @@ public static class KpiSeriesBucketer
     /// <param name="fromUtc">UTC start of the query window (inclusive).</param>
     /// <param name="toUtc">UTC end of the query window (inclusive on the right edge).</param>
     /// <param name="maxPoints">Maximum number of output points. Must be ≥ 2.</param>
+    /// <param name="aggregation">
+    ///   How each bucket is reduced. <see cref="KpiRollupAggregation.Gauge"/> (default) keeps the
+    ///   last value in the bucket — the historical behavior. <see cref="KpiRollupAggregation.Rate"/>
+    ///   sums the values in the bucket — each raw point of a Rate series is a per-interval delta, so
+    ///   the bucket total is the sum of its deltas (arch-review 04 round 2, R3).
+    /// </param>
     /// <returns>
     ///   An <see cref="IReadOnlyList{T}"/> of at most <paramref name="maxPoints"/> bucketed points,
     ///   ordered by <see cref="KpiSeriesPoint.BucketStartUtc"/> ascending.
-    ///   Returns <paramref name="raw"/> unchanged (same reference) when
+    ///   For <see cref="KpiRollupAggregation.Gauge"/> returns <paramref name="raw"/> unchanged
-    ///   <c>raw.Count &lt;= maxPoints</c>; callers must not mutate the underlying
+    ///   (same reference) when <c>raw.Count &lt;= maxPoints</c>; callers must not mutate the
-    ///   collection in that case, as it is the same object passed in.
+    ///   underlying collection in that case. <see cref="KpiRollupAggregation.Rate"/> always runs
+    ///   the bucketing pass — two deltas landing in one bucket must sum even in a short series —
+    ///   so it never returns the input reference.
     /// </returns>
     /// <exception cref="ArgumentOutOfRangeException">
     ///   Thrown when <paramref name="maxPoints"/> &lt; 2 or
@@ -44,7 +58,8 @@ public static class KpiSeriesBucketer
         IReadOnlyList<KpiSeriesPoint> raw,
         DateTime fromUtc,
         DateTime toUtc,
-        int maxPoints)
+        int maxPoints,
+        KpiRollupAggregation aggregation = KpiRollupAggregation.Gauge)
     {
         if (maxPoints < 2)
             throw new ArgumentOutOfRangeException(nameof(maxPoints),
@@ -54,11 +69,15 @@ public static class KpiSeriesBucketer
             throw new ArgumentOutOfRangeException(nameof(toUtc),
                 toUtc, "toUtc must be strictly greater than fromUtc.");
 
-        // Normal runtime case — empty or short series: return as-is.
+        // Normal runtime case — empty series: return as-is.
         if (raw is null || raw.Count == 0)
             return Array.Empty<KpiSeriesPoint>();
 
-        if (raw.Count <= maxPoints)
+        // Short-series early return is Gauge-only: for Rate, two deltas landing in one bucket
+        // must still sum, so a short Rate series cannot be returned verbatim (arch-review 04
+        // round 2, R3). For well-spaced short Rate series the per-point buckets make the sum an
+        // identity, so the output still equals the input anyway.
+        if (aggregation == KpiRollupAggregation.Gauge && raw.Count <= maxPoints)
             return raw;
 
         // Divide the window into maxPoints equal-width buckets.
@@ -67,12 +86,11 @@ public static class KpiSeriesBucketer
         double windowTicks = (double)(toUtc.Ticks - fromUtc.Ticks);
         double bucketWidthTicks = windowTicks / maxPoints;
 
-        // For each bucket, track the candidate point: the one with the
+        // For each bucket, track the candidate point: for Gauge the one with the maximum
-        // maximum BucketStartUtc (last value within the bucket).
+        // BucketStartUtc (last value within the bucket); for Rate a running sum of the bucket's
-        // We use a fixed-size array indexed by bucket number.
+        // deltas. Fixed-size arrays indexed by bucket number.
-        // Nullable KpiSeriesPoint[] with 'hasValue' flags is fine since the
-        // struct is small.
         var best = new KpiSeriesPoint[maxPoints];
+        var sums = new double[maxPoints];
         var occupied = new bool[maxPoints];
 
         foreach (var point in raw)
@@ -89,13 +107,19 @@ public static class KpiSeriesBucketer
             if (bucketIndex >= maxPoints)
                 bucketIndex = maxPoints - 1;
 
+            if (aggregation == KpiRollupAggregation.Rate)
+            {
+                // Rate: accumulate the bucket's per-interval deltas.
+                sums[bucketIndex] += point.Value;
+                occupied[bucketIndex] = true;
+            }
             // Keep the last point in iteration order within this bucket. Because the stored
             // candidate's BucketStartUtc is the bucket-START timestamp (not the raw point's
             // capture time), the comparison below is true for essentially any in-bucket point,
             // so each later-in-iteration point overwrites the previous one. For the ascending-
             // sorted input this method requires, last-in-iteration IS the largest-timestamp
             // point — i.e. last-value-per-bucket semantics.
-            if (!occupied[bucketIndex] ||
+            else if (!occupied[bucketIndex] ||
                 point.BucketStartUtc > best[bucketIndex].BucketStartUtc)
             {
                 best[bucketIndex] = new KpiSeriesPoint(
@@ -109,8 +133,13 @@ public static class KpiSeriesBucketer
         var result = new List<KpiSeriesPoint>(maxPoints);
         for (int i = 0; i < maxPoints; i++)
         {
-            if (occupied[i])
+            if (!occupied[i])
-                result.Add(best[i]);
+                continue;
+
+            result.Add(aggregation == KpiRollupAggregation.Rate
+                ? new KpiSeriesPoint(
+                    fromUtc + TimeSpan.FromTicks((long)(i * bucketWidthTicks)), sums[i])
+                : best[i]);
         }
 
         return result;

tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Kpi/KpiSeriesBucketerTests.cs

@@ -369,4 +369,97 @@ public class KpiSeriesBucketerTests
         Assert.Equal(4.0, result[0].Value);
         Assert.Equal(9.0, result[1].Value);
     }
+
+    // -----------------------------------------------------------------------
+    // Per-metric reduction: Rate series sum per bucket (arch-review 04 round 2, R3)
+    // -----------------------------------------------------------------------
+
+    [Fact]
+    public void Bucket_RateSeries_SumsPerBucket_InsteadOfLastValue()
+    {
+        // 6 points of value 10 across a 60-min / 2-bucket window → each bucket = 30 (sum),
+        // not 10 (last value). Three deltas per bucket; last-value would keep 1 of 3.
+        var raw = new[]
+        {
+            new KpiSeriesPoint(T(5),  10.0), // bucket 0: [0, 30)
+            new KpiSeriesPoint(T(15), 10.0), // bucket 0
+            new KpiSeriesPoint(T(25), 10.0), // bucket 0
+            new KpiSeriesPoint(T(35), 10.0), // bucket 1: [30, 60]
+            new KpiSeriesPoint(T(45), 10.0), // bucket 1
+            new KpiSeriesPoint(T(55), 10.0), // bucket 1
+        };
+
+        var result = KpiSeriesBucketer.Bucket(
+            raw, T(0), T(60), maxPoints: 2, aggregation: KpiRollupAggregation.Rate);
+
+        Assert.Equal(2, result.Count);
+        Assert.Equal(30.0, result[0].Value);
+        Assert.Equal(30.0, result[1].Value);
+    }
+
+    [Fact]
+    public void Bucket_RateSeries_ShortSeries_StillSumsClusteredPoints()
+    {
+        // 3 points (< maxPoints 5) with two sharing a bucket → the shared bucket is their SUM.
+        // Rate series skip the raw.Count <= maxPoints early return so totals stay truthful.
+        // 60-min / 5-bucket window → 12 min each. T(2),T(5) → bucket 0 [0,12); T(30) → bucket 2 [24,36).
+        var raw = new[]
+        {
+            new KpiSeriesPoint(T(2),  10.0),
+            new KpiSeriesPoint(T(5),  10.0),
+            new KpiSeriesPoint(T(30), 10.0),
+        };
+
+        var result = KpiSeriesBucketer.Bucket(
+            raw, T(0), T(60), maxPoints: 5, aggregation: KpiRollupAggregation.Rate);
+
+        Assert.Equal(2, result.Count);
+        Assert.Equal(20.0, result[0].Value); // T(2)+T(5) summed
+        Assert.Equal(10.0, result[1].Value); // T(30) alone
+    }
+
+    [Fact]
+    public void Bucket_RateSeries_PreservesSeriesTotal()
+    {
+        // Strong invariant: sum(output values) == sum(in-window input values) for Rate.
+        var raw = Enumerable
+            .Range(0, 20)
+            .Select(i => new KpiSeriesPoint(T(i * 3), (double)(i + 1)))
+            .ToArray();
+
+        var result = KpiSeriesBucketer.Bucket(
+            raw, T(0), T(60), maxPoints: 4, aggregation: KpiRollupAggregation.Rate);
+
+        var inWindowTotal = raw
+            .Where(p => p.BucketStartUtc >= T(0) && p.BucketStartUtc <= T(60))
+            .Sum(p => p.Value);
+        Assert.Equal(inWindowTotal, result.Sum(p => p.Value));
+    }
+
+    [Fact]
+    public void Bucket_DefaultAggregation_IsGauge_AndBehaviorUnchanged()
+    {
+        // Calling the existing 4-arg shape produces identical output to the explicit Gauge call.
+        var raw = new[]
+        {
+            new KpiSeriesPoint(T(5),  1.0),
+            new KpiSeriesPoint(T(15), 2.0),
+            new KpiSeriesPoint(T(25), 99.0),
+            new KpiSeriesPoint(T(35), 5.0),
+        };
+
+        var fourArg = KpiSeriesBucketer.Bucket(raw, T(0), T(60), maxPoints: 2);
+        var explicitGauge = KpiSeriesBucketer.Bucket(
+            raw, T(0), T(60), maxPoints: 2, aggregation: KpiRollupAggregation.Gauge);
+
+        Assert.Equal(fourArg.Count, explicitGauge.Count);
+        for (int i = 0; i < fourArg.Count; i++)
+        {
+            Assert.Equal(fourArg[i].BucketStartUtc, explicitGauge[i].BucketStartUtc);
+            Assert.Equal(fourArg[i].Value, explicitGauge[i].Value);
+        }
+        // And the last-value semantics are unchanged.
+        Assert.Equal(99.0, fourArg[0].Value);
+        Assert.Equal(5.0, fourArg[1].Value);
+    }
 }