Implement movingSum, movingSumBy, movingProduct, and movingProductBy

CHANGELOG.md

@@ -2,6 +2,8 @@
 + Use greedy integrators where possible (Fixes #57)
 + Add [JSpecify](https://jspecify.dev/) annotations for static analysis
 + Implement `orderByFrequencyAscending()` and `orderByFrequencyDescending()`
++ Implement `movingProduct()` and `movingProductBy()`
++ Implement `movingSum()` and `movingSumBy()`
 
 ### 0.6.0
 + Implement `dropLast(n)`

README.md

@@ -64,6 +64,10 @@ implementation("com.ginsberg:gatherers4j:0.7.0")
 
 | Function                                   | Purpose                                                                                                                            |
 |--------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------|
+| `movingProduct(window)`                    | Create a moving product of `BigDecimal` values over the previous `window` values.                                                  |
+| `movingProductBy(fn, window)`              | Create a moving product of `BigDecimal` values over the previous `window` values, as mapped via `fn`.                              |
+| `movingSum(window)`                        | Create a moving sum of `BigDecimal` values over the previous `window` values.                                                      |
+| `movingSumBy(fn, window)`                  | Create a moving sum of `BigDecimal` values over the previous `window` values, as mapped via `fn`.                                  |
 | `runningPopulationStandardDeviation()`     | Create a stream of `BigDecimal` objects representing the running population standard deviation.                                    |
 | `runningPopulationStandardDeviationBy(fn)` | Create a stream of `BigDecimal` objects as mapped from the input via `fn`, representing the running population standard deviation. |
 | `runningProduct()`                         | Create a stream of `BigDecimal` objects representing the running product.                                                          |                                                          |

src/main/java/com/ginsberg/gatherers4j/BigDecimalMovingProductGatherer.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2024 Todd Ginsberg
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ginsberg.gatherers4j;
+
+import org.jspecify.annotations.Nullable;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import java.util.Arrays;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+public class BigDecimalMovingProductGatherer<INPUT extends @Nullable Object>
+        extends BigDecimalGatherer<INPUT> {
+
+    private final int windowSize;
+    private boolean includePartialValues = false;
+
+    BigDecimalMovingProductGatherer(
+            final Function<INPUT, @Nullable BigDecimal> mappingFunction,
+            final int windowSize) {
+        super(mappingFunction);
+        if (windowSize <= 0) {
+            throw new IllegalArgumentException("Window size must be positive");
+        }
+        this.windowSize = windowSize;
+    }
+
+    @Override
+    public Supplier<BigDecimalGatherer.State> initializer() {
+        return () -> new BigDecimalMovingProductGatherer.State(windowSize, includePartialValues);
+    }
+
+    /// When creating a moving product and the full size of the window has not yet been reached, the
+    /// gatherer should emit the product for what it has.
+    ///
+    /// For example, if the trailing product is over 10 values, but the stream has only emitted two
+    /// values, the gatherer should calculate the two values and emit the answer. The default is to not
+    /// emit anything until the full size of the window has been seen.
+    public BigDecimalMovingProductGatherer<INPUT> includePartialValues() {
+        includePartialValues = true;
+        return this;
+    }
+
+    /// When encountering a `null` value in a stream, treat it as `BigDecimal.ONE` instead.
+    public BigDecimalGatherer<INPUT> treatNullAsOne() {
+        return treatNullAs(BigDecimal.ONE);
+    }
+
+    static class State implements BigDecimalGatherer.State {
+        final boolean includePartialValues;
+        final BigDecimal[] series;
+        BigDecimal product = BigDecimal.ONE;
+        int index = 0;
+
+        private State(final int lookBack, final boolean includePartialValues) {
+            this.includePartialValues = includePartialValues;
+            this.series = new BigDecimal[lookBack];
+            Arrays.fill(series, BigDecimal.ONE);
+        }
+
+        @Override
+        public boolean canCalculate() {
+            return includePartialValues || index >= series.length;
+        }
+
+        @Override
+        public void add(final BigDecimal element, final MathContext mathContext) {
+            product = product.divide(series[index % series.length], mathContext).multiply(element, mathContext);
+            series[index % series.length] = element;
+            index++;
+        }
+
+        @Override
+        public BigDecimal calculate() {
+            return product;
+        }
+    }
+}

src/main/java/com/ginsberg/gatherers4j/BigDecimalMovingSumGatherer.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2024 Todd Ginsberg
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ginsberg.gatherers4j;
+
+import org.jspecify.annotations.Nullable;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import java.util.Arrays;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+public class BigDecimalMovingSumGatherer<INPUT extends @Nullable Object>
+        extends BigDecimalGatherer<INPUT> {
+
+    private final int windowSize;
+    private boolean includePartialValues = false;
+
+    BigDecimalMovingSumGatherer(
+            final Function<INPUT, @Nullable BigDecimal> mappingFunction,
+            final int windowSize) {
+        super(mappingFunction);
+        if (windowSize <= 0) {
+            throw new IllegalArgumentException("Window size must be positive");
+        }
+        this.windowSize = windowSize;
+    }
+
+    @Override
+    public Supplier<BigDecimalGatherer.State> initializer() {
+        return () -> new BigDecimalMovingSumGatherer.State(windowSize, includePartialValues);
+    }
+
+    /// When creating a moving sum and the full size of the window has not yet been reached, the
+    /// gatherer should emit the sum of what it has.
+    ///
+    /// For example, if the trailing sum is over 10 values, but the stream has only emitted two
+    /// values, the gatherer should calculate the two values and emit the answer. The default is to not
+    /// emit anything until the full size of the window has been seen.
+    public BigDecimalMovingSumGatherer<INPUT> includePartialValues() {
+        includePartialValues = true;
+        return this;
+    }
+
+    static class State implements BigDecimalGatherer.State {
+        final boolean includePartialValues;
+        final BigDecimal[] series;
+        BigDecimal sum = BigDecimal.ZERO;
+        int index = 0;
+
+        private State(final int lookBack, final boolean includePartialValues) {
+            this.includePartialValues = includePartialValues;
+            this.series = new BigDecimal[lookBack];
+            Arrays.fill(series, BigDecimal.ZERO);
+        }
+
+        @Override
+        public boolean canCalculate() {
+            return includePartialValues || index >= series.length;
+        }
+
+        @Override
+        public void add(final BigDecimal element, final MathContext mathContext) {
+            sum = sum.subtract(series[index % series.length]).add(element, mathContext);
+            series[index % series.length] = element;
+            index++;
+        }
+
+        @Override
+        public BigDecimal calculate() {
+            return sum;
+        }
+    }
+}

src/main/java/com/ginsberg/gatherers4j/Gatherers4j.java

@@ -94,7 +94,7 @@ public static <INPUT> SizeGatherer<INPUT> exactSize(final long size) {
     /// and its index.
     ///
     /// @param predicate A non-null `BiPredicate<Long,INPUT>` where the `Long` is the zero-based index of the element
-    ///                                   being filtered, and the `INPUT` is the element itself.
+    ///                                                    being filtered, and the `INPUT` is the element itself.
     /// @param <INPUT>   Type of elements in the input stream
     /// @return A non-null `FilteringWithIndexGatherer`
     public static <INPUT> FilteringWithIndexGatherer<INPUT> filterWithIndex(
@@ -188,6 +188,23 @@ public static <INPUT, MAPPED extends Comparable<MAPPED>> MinMaxGatherer<INPUT, M
         return new MinMaxGatherer<>(false, mappingFunction);
     }
 
+    public static BigDecimalMovingProductGatherer<BigDecimal> movingProduct(int windowSize) {
+        return movingProductBy(Function.identity(), windowSize);
+    }
+
+    public static <INPUT> BigDecimalMovingProductGatherer<INPUT> movingProductBy(Function<INPUT, BigDecimal> mappingFunction, int windowSize) {
+        return new BigDecimalMovingProductGatherer<>(mappingFunction, windowSize);
+    }
+
+    public static BigDecimalMovingSumGatherer<BigDecimal> movingSum(int windowSize) {
+        return movingSumBy(Function.identity(), windowSize);
+    }
+
+    public static <INPUT> BigDecimalMovingSumGatherer<INPUT> movingSumBy(Function<INPUT, BigDecimal> mappingFunction, int windowSize) {
+        return new BigDecimalMovingSumGatherer<>(mappingFunction, windowSize);
+    }
+
+
 
     /// Emit elements in the input stream ordered by frequency from least frequently occurring
     /// to most frequently occurring. Elements are emitted wrapped in `WithCount<INPUT>` objects
@@ -200,8 +217,8 @@ public static <INPUT, MAPPED extends Comparable<MAPPED>> MinMaxGatherer<INPUT, M
     ///       .toList();
     ///
     /// // Produces:
-    /// [WithCount("C", 1), WithCount("B", 2), WithCount("A", 4)]
-    /// ```
+    ///[WithCount("C", 1), WithCount("B", 2), WithCount("A", 4)]
+    ///```
     ///
     /// Note: This consumes the entire stream and holds it in memory, so it will not work on infinite
     /// streams and may cause memory pressure on very large streams.
@@ -223,8 +240,8 @@ public static <INPUT> FrequencyGatherer<INPUT> orderByFrequencyAscending() {
     ///       .toList();
     ///
     /// // Produces:
-    /// [WithCount("A", 4), WithCount("B", 2), WithCount("C", 1)]
-    /// ```
+    ///[WithCount("A", 4), WithCount("B", 2), WithCount("C", 1)]
+    ///```
     ///
     /// Note: This consumes the entire stream and holds it in memory, so it will not work on infinite
     /// streams and may cause memory pressure on very large streams.
@@ -261,7 +278,7 @@ public static BigDecimalStandardDeviationGatherer<BigDecimal> runningPopulationS
     /// objects mapped from a `Stream<BigDecimal>` via a `mappingFunction`.
     ///
     /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
-    ///                                               in the standard deviation calculation
+    ///                                                                      in the standard deviation calculation
     /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
     /// @return A non-null `BigDecimalStandardDeviationGatherer`
     public static <INPUT> BigDecimalStandardDeviationGatherer<INPUT> runningPopulationStandardDeviationBy(
@@ -284,7 +301,7 @@ public static BigDecimalProductGatherer<BigDecimal> runningProduct() {
     /// from a `Stream<INPUT>` via a `mappingFunction`.
     ///
     /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
-    ///                                               in the product calculation
+    ///                                                                      in the product calculation
     /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
     /// @return A non-null `BigDecimalProductGatherer`
     public static <INPUT> BigDecimalProductGatherer<INPUT> runningProductBy(
@@ -307,7 +324,7 @@ public static BigDecimalStandardDeviationGatherer<BigDecimal> runningSampleStand
     /// from a `Stream<INPUT>` via a `mappingFunction`.
     ///
     /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
-    ///                                               in the standard deviation calculation
+    ///                                                                      in the standard deviation calculation
     /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
     /// @return A non-null `BigDecimalStandardDeviationGatherer`
     public static <INPUT> BigDecimalStandardDeviationGatherer<INPUT> runningSampleStandardDeviationBy(
@@ -330,7 +347,7 @@ public static BigDecimalSumGatherer<BigDecimal> runningSum() {
     /// from a `Stream<INPUT>` via a `mappingFunction`.
     ///
     /// @param mappingFunction A function to map `<INPUT>` objects to `BigDecimal`, the results of which will be used
-    ///                                               in the running sum calculation
+    ///                                                                      in the running sum calculation
     /// @param <INPUT>         Type of elements in the input stream, to be remapped to `BigDecimal` by the `mappingFunction`
     /// @return A non-null `BigDecimalSumGatherer`
     public static <INPUT> BigDecimalSumGatherer<INPUT> runningSumBy(