From 33cc194bff991ff70ecfe051b3631a3a17f88916 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Thu, 27 Aug 2026 00:14:23 +0000 Subject: [PATCH] Make EfficientEvaluationLoops hookable for Python callbacks Add HookableEvaluationLoop, a test-then-train loop that notifies registered callbacks (OnLabel per instance, OnWindowClose per window) so CapyMOA can observe predictions and scores as they happen, unblocking score-based metrics like AUC without paying the cost of the slow pure-Python path. Hooks receive normalized predicted probabilities; evaluators still get raw votes so PrequentialResult is unchanged. The trailing partial window is always closed and handed to OnWindowClose hooks. Buffering only occurs when such hooks are registered. EfficientEvaluationLoops.PrequentialEvaluation now delegates to the new loop, preserving its signature and behavior for backwards compatibility. Test parity verified against pre-change baseline: 70.8% accuracy on 1000 RandomTreeGenerator instances with NaiveBayes (seed 1), 10 windows of 100. All evaluator-related tests still pass. Addresses adaptive-machine-learning/backlog#145. Co-Authored-By: Claude --- .../evaluation/EfficientEvaluationLoops.java | 69 +---- .../evaluation/HookableEvaluationLoop.java | 242 +++++++++++++++++ .../HookableEvaluationLoopTest.java | 252 ++++++++++++++++++ 3 files changed, 499 insertions(+), 64 deletions(-) create mode 100644 moa/src/main/java/moa/evaluation/HookableEvaluationLoop.java create mode 100644 moa/src/test/java/moa/evaluation/HookableEvaluationLoopTest.java diff --git a/moa/src/main/java/moa/evaluation/EfficientEvaluationLoops.java b/moa/src/main/java/moa/evaluation/EfficientEvaluationLoops.java index 73a1b405c..1816fc403 100644 --- a/moa/src/main/java/moa/evaluation/EfficientEvaluationLoops.java +++ b/moa/src/main/java/moa/evaluation/EfficientEvaluationLoops.java @@ -68,6 +68,7 @@ public PrequentialResult( * It can also be used just to calculate the test-then-train metrics (set windowed_evaluator to null) * Finally, it can also be used to calculate test-then-train metrics and sample them over time, * just set basic_evaluator to null and specify a BasicClassificationPerformanceEvaluator as the windowed_evaluator. + * Use {@link HookableEvaluationLoop} directly if you also need to observe the loop as it runs. * @param stream * @param learner * @param basicEvaluator @@ -83,70 +84,10 @@ public static PrequentialResult PrequentialEvaluation(ExampleStream stream, Lear LearningPerformanceEvaluator> windowedEvaluator, long maxInstances, long windowSize, boolean storeY, boolean storePredictions) { - int instancesProcessed = 0; - - if (!stream.hasMoreInstances()) - stream.restart(); - - ArrayList windowed_results = new ArrayList<>(); - ArrayList targetValues = new ArrayList<>(); - ArrayList predictions = new ArrayList<>(); - - - while (stream.hasMoreInstances() && - (maxInstances == -1 || instancesProcessed < maxInstances)) { - - Example instance = stream.nextInstance(); - - double[] prediction = learner.getVotesForInstance(instance); - - // Update evaluators and store predictions if requested - if (basicEvaluator != null) - basicEvaluator.addResult(instance, prediction); - if (windowedEvaluator != null) - windowedEvaluator.addResult(instance, prediction); - if (storePredictions) - predictions.add(Utils.maxIndex(prediction)); - if (storeY) - targetValues.add((int)Math.round(instance.getData().classValue())); - - learner.trainOnInstance(instance); - instancesProcessed++; - - // Store windowed results if requested - if (windowedEvaluator != null) - if (instancesProcessed % windowSize == 0) { - Measurement[] measurements = windowedEvaluator.getPerformanceMeasurements(); - double[] values = new double[measurements.length]; - for (int i = 0; i < values.length; ++i) - values[i] = measurements[i].getValue(); - windowed_results.add(values); - } - } - if (windowedEvaluator != null) - if (instancesProcessed % windowSize != 0) { - Measurement[] measurements = windowedEvaluator.getPerformanceMeasurements(); - double[] values = new double[measurements.length]; - for (int i = 0; i < values.length; ++i) - values[i] = measurements[i].getValue(); - windowed_results.add(values); - } - - double[] cumulative_results = null; - - if (basicEvaluator != null) { - Measurement[] measurements = basicEvaluator.getPerformanceMeasurements(); - cumulative_results = new double[measurements.length]; - for (int i = 0; i < cumulative_results.length; ++i) - cumulative_results[i] = measurements[i].getValue(); - } - - return new PrequentialResult( - windowed_results, - cumulative_results, - targetValues, - predictions - ); + return new HookableEvaluationLoop() + .registerBasic(basicEvaluator) + .registerWindowed(windowedEvaluator) + .run(stream, learner, maxInstances, windowSize, storeY, storePredictions); } public static PrequentialResult PrequentialSSLEvaluation( diff --git a/moa/src/main/java/moa/evaluation/HookableEvaluationLoop.java b/moa/src/main/java/moa/evaluation/HookableEvaluationLoop.java new file mode 100644 index 000000000..61a3894d8 --- /dev/null +++ b/moa/src/main/java/moa/evaluation/HookableEvaluationLoop.java @@ -0,0 +1,242 @@ +package moa.evaluation; + +import com.yahoo.labs.samoa.instances.Instance; +import moa.core.Example; +import moa.core.Measurement; +import moa.core.Utils; +import moa.evaluation.EfficientEvaluationLoops.PrequentialResult; +import moa.learners.Learner; +import moa.streams.ExampleStream; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +/** + * A prequential evaluation loop that callers can observe by registering hooks. + * + *

{@link EfficientEvaluationLoops#PrequentialEvaluation} runs entirely inside Java and only + * returns aggregated measurements, so a caller on the other side of a language boundary cannot see + * anything the loop does while it runs. This class runs the same loop, but lets a caller register + * callbacks that are invoked per instance ({@link OnLabel}) or per window ({@link OnWindowClose}). + * + *

Hooks receive normalized predicted probabilities rather than MOA's raw votes, which is what + * score-based metrics such as AUC need. The registered evaluators still see the raw votes, so the + * returned {@link PrequentialResult} is unaffected by whether hooks are present. + * + * 

+ * PrequentialResult result = new HookableEvaluationLoop()
+ *         .registerBasic(basicEvaluator)
+ *         .registerWindowed(windowedEvaluator)
+ *         .register((OnWindowClose) (predProbs, labels) -> ...)
+ *         .run(stream, learner, 1000, 100, true, true);
+ *
+ */ +public class HookableEvaluationLoop { + + /** + * Marker for anything that can be passed to {@link #register(Hook)}. Implement one or both of + * {@link OnLabel} and {@link OnWindowClose}; a hook that implements neither is rejected. + */ + public interface Hook { + } + + /** Called once per instance, after the learner has predicted but before it has trained. */ + @FunctionalInterface + public interface OnLabel extends Hook { + /** + * @param instance the instance just tested on. It belongs to the stream and may be reused + * or mutated once this call returns, so a hook must not retain it. + * @param predProbs predicted probabilities for {@code instance}, summing to one. All zeros + * if the learner produced no votes at all, which happens while it is + * still untrained. + */ + void onLabel(Example instance, double[] predProbs); + } + + /** + * Called once per window, including the trailing partial window when the stream does not divide + * evenly into windows. + */ + @FunctionalInterface + public interface OnWindowClose extends Hook { + /** + * @param predProbs one row per instance in the window, each summing to one. Rows may differ + * in length: MOA grows the vote array as it encounters new classes, so + * early rows can be shorter than later ones. + * @param labels the true class index of each instance in the window, parallel to + * {@code predProbs}. + */ + void onWindowClose(double[][] predProbs, int[] labels); + } + + protected final List onLabelHooks = new ArrayList<>(); + protected final List onWindowCloseHooks = new ArrayList<>(); + protected LearningPerformanceEvaluator> basicEvaluator; + protected LearningPerformanceEvaluator> windowedEvaluator; + + /** + * Registers a callback. A hook implementing both sub-interfaces is registered for both. + * + * @throws IllegalArgumentException if the hook implements neither sub-interface, which would + * otherwise silently do nothing. + */ + public HookableEvaluationLoop register(Hook hook) { + boolean registered = false; + if (hook instanceof OnLabel) { + this.onLabelHooks.add((OnLabel) hook); + registered = true; + } + if (hook instanceof OnWindowClose) { + this.onWindowCloseHooks.add((OnWindowClose) hook); + registered = true; + } + if (!registered) + throw new IllegalArgumentException( + "Hook must implement OnLabel and/or OnWindowClose, got: " + hook.getClass().getName()); + return this; + } + + /** + * Sets the evaluator producing {@link PrequentialResult#cumulativeResults}. May be null, in + * which case no cumulative results are produced. + */ + public HookableEvaluationLoop registerBasic(LearningPerformanceEvaluator> evaluator) { + this.basicEvaluator = evaluator; + return this; + } + + /** + * Sets the evaluator sampled at each window boundary to produce + * {@link PrequentialResult#windowedResults}. May be null, in which case no windowed results are + * produced. {@link OnWindowClose} hooks fire regardless of whether this is set. + */ + public HookableEvaluationLoop registerWindowed(LearningPerformanceEvaluator> evaluator) { + this.windowedEvaluator = evaluator; + return this; + } + + /** + * Runs test-then-train over the stream, notifying any registered hooks as it goes. + * + * @param maxInstances stop after this many instances, or -1 for the whole stream. + * @param windowSize how many instances make up a window. + * @param storeY collect the true class of every instance into the result. + * @param storePredictions collect the predicted class of every instance into the result. + */ + public PrequentialResult run(ExampleStream stream, Learner learner, + long maxInstances, long windowSize, + boolean storeY, boolean storePredictions) { + int instancesProcessed = 0; + + if (!stream.hasMoreInstances()) + stream.restart(); + + ArrayList windowed_results = new ArrayList<>(); + ArrayList targetValues = new ArrayList<>(); + ArrayList predictions = new ArrayList<>(); + + // Only buffer a window's observations when somebody is going to be handed them. + boolean buffering = !this.onWindowCloseHooks.isEmpty(); + ArrayList windowPredProbs = buffering ? new ArrayList<>() : null; + ArrayList windowLabels = buffering ? new ArrayList<>() : null; + + while (stream.hasMoreInstances() && + (maxInstances == -1 || instancesProcessed < maxInstances)) { + + Example instance = stream.nextInstance(); + + double[] prediction = learner.getVotesForInstance(instance); + + // Update evaluators and store predictions if requested + if (this.basicEvaluator != null) + this.basicEvaluator.addResult(instance, prediction); + if (this.windowedEvaluator != null) + this.windowedEvaluator.addResult(instance, prediction); + if (storePredictions) + predictions.add(Utils.maxIndex(prediction)); + if (storeY) + targetValues.add((int) Math.round(instance.getData().classValue())); + + // Notify hooks. The evaluators above deliberately saw the raw votes. + if (!this.onLabelHooks.isEmpty()) { + double[] predProbs = toProbs(prediction); + for (OnLabel hook : this.onLabelHooks) + hook.onLabel(instance, predProbs); + } + if (buffering) { + windowPredProbs.add(toProbs(prediction)); + windowLabels.add((int) Math.round(instance.getData().classValue())); + } + + learner.trainOnInstance(instance); + instancesProcessed++; + + // Store windowed results if requested + if (instancesProcessed % windowSize == 0) { + if (this.windowedEvaluator != null) + windowed_results.add(flatten(this.windowedEvaluator)); + if (buffering) + closeWindow(windowPredProbs, windowLabels); + } + } + // The stream ran out mid-window; close it out anyway. + if (instancesProcessed % windowSize != 0) { + if (this.windowedEvaluator != null) + windowed_results.add(flatten(this.windowedEvaluator)); + if (buffering) + closeWindow(windowPredProbs, windowLabels); + } + + double[] cumulative_results = null; + + if (this.basicEvaluator != null) + cumulative_results = flatten(this.basicEvaluator); + + return new PrequentialResult( + windowed_results, + cumulative_results, + targetValues, + predictions + ); + } + + /** Hands the buffered window to every {@link OnWindowClose} hook, then empties the buffer. */ + protected void closeWindow(ArrayList windowPredProbs, ArrayList windowLabels) { + double[][] predProbs = windowPredProbs.toArray(new double[0][]); + int[] labels = new int[windowLabels.size()]; + for (int i = 0; i < labels.length; ++i) + labels[i] = windowLabels.get(i); + + for (OnWindowClose hook : this.onWindowCloseHooks) + hook.onWindowClose(predProbs, labels); + + windowPredProbs.clear(); + windowLabels.clear(); + } + + /** The current value of every measurement the evaluator reports, in its own order. */ + protected static double[] flatten(LearningPerformanceEvaluator> evaluator) { + Measurement[] measurements = evaluator.getPerformanceMeasurements(); + double[] values = new double[measurements.length]; + for (int i = 0; i < values.length; ++i) + values[i] = measurements[i].getValue(); + return values; + } + + /** + * Copies votes and scales them to sum to one. Unlike {@link Utils#normalize(double[])} this + * tolerates a zero sum, which an untrained learner produces routinely, by returning the zeros + * unchanged rather than throwing. + */ + protected static double[] toProbs(double[] votes) { + double[] probs = Arrays.copyOf(votes, votes.length); + double sum = 0; + for (double vote : probs) + sum += vote; + if (sum > 0 && !Double.isNaN(sum)) + for (int i = 0; i < probs.length; ++i) + probs[i] /= sum; + return probs; + } +} diff --git a/moa/src/test/java/moa/evaluation/HookableEvaluationLoopTest.java b/moa/src/test/java/moa/evaluation/HookableEvaluationLoopTest.java new file mode 100644 index 000000000..63847df81 --- /dev/null +++ b/moa/src/test/java/moa/evaluation/HookableEvaluationLoopTest.java @@ -0,0 +1,252 @@ +package moa.evaluation; + +import com.yahoo.labs.samoa.instances.Instance; +import moa.classifiers.bayes.NaiveBayes; +import moa.core.Example; +import moa.core.Utils; +import moa.evaluation.EfficientEvaluationLoops.PrequentialResult; +import moa.evaluation.HookableEvaluationLoop.Hook; +import moa.evaluation.HookableEvaluationLoop.OnLabel; +import moa.evaluation.HookableEvaluationLoop.OnWindowClose; +import moa.streams.generators.RandomTreeGenerator; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class HookableEvaluationLoopTest { + + private static final int N = 1000; + private static final int WINDOW = 100; + + /** + * Accuracy of the configuration below, captured by running the loop before it was made + * hookable. Pinning it here is what makes the parity assertions meaningful, since + * EfficientEvaluationLoops now delegates to the class under test. + */ + private static final double BASELINE_ACCURACY = 70.8; + + private static final int ACCURACY_INDEX = 1; // "classifications correct (percent)" + + private RandomTreeGenerator stream() { + RandomTreeGenerator stream = new RandomTreeGenerator(); + stream.treeRandomSeedOption.setValue(1); + stream.instanceRandomSeedOption.setValue(1); + stream.prepareForUse(); + return stream; + } + + private NaiveBayes learner(RandomTreeGenerator stream) { + NaiveBayes learner = new NaiveBayes(); + learner.setModelContext(stream.getHeader()); + learner.prepareForUse(); + return learner; + } + + private BasicClassificationPerformanceEvaluator basicEvaluator() { + BasicClassificationPerformanceEvaluator evaluator = new BasicClassificationPerformanceEvaluator(); + evaluator.prepareForUse(); + return evaluator; + } + + private WindowClassificationPerformanceEvaluator windowedEvaluator() { + WindowClassificationPerformanceEvaluator evaluator = new WindowClassificationPerformanceEvaluator(); + evaluator.widthOption.setValue(WINDOW); + evaluator.prepareForUse(); + return evaluator; + } + + /** Runs the loop with the standard configuration and whatever hooks are given. */ + private PrequentialResult run(Hook... hooks) { + RandomTreeGenerator stream = stream(); + HookableEvaluationLoop loop = new HookableEvaluationLoop() + .registerBasic(basicEvaluator()) + .registerWindowed(windowedEvaluator()); + for (Hook hook : hooks) + loop.register(hook); + return loop.run(stream, learner(stream), N, WINDOW, true, true); + } + + @Test + public void unhookedRunMatchesBaseline() { + PrequentialResult result = run(); + + assertEquals(BASELINE_ACCURACY, result.cumulativeResults[ACCURACY_INDEX], 1e-9); + assertEquals(N / WINDOW, result.windowedResults.size()); + assertEquals(N, result.targets.size()); + assertEquals(N, result.predictions.size()); + } + + @Test + public void staticEntryPointStillMatchesBaseline() { + RandomTreeGenerator stream = stream(); + PrequentialResult result = EfficientEvaluationLoops.PrequentialEvaluation( + stream, learner(stream), basicEvaluator(), windowedEvaluator(), N, WINDOW, true, true); + + assertEquals(BASELINE_ACCURACY, result.cumulativeResults[ACCURACY_INDEX], 1e-9); + assertEquals(N / WINDOW, result.windowedResults.size()); + } + + @Test + public void hooksDoNotChangeResults() { + PrequentialResult unhooked = run(); + PrequentialResult hooked = run((OnLabel) (instance, predProbs) -> { + }, (OnWindowClose) (predProbs, labels) -> { + }); + + assertEquals(unhooked.cumulativeResults[ACCURACY_INDEX], hooked.cumulativeResults[ACCURACY_INDEX], 1e-9); + assertEquals(unhooked.targets, hooked.targets); + assertEquals(unhooked.predictions, hooked.predictions); + assertEquals(unhooked.windowedResults.size(), hooked.windowedResults.size()); + for (int i = 0; i < unhooked.windowedResults.size(); ++i) + assertArrayEquals(unhooked.windowedResults.get(i), hooked.windowedResults.get(i), 1e-9); + } + + @Test + public void onLabelFiresOncePerInstanceWithProbabilities() { + List seenPredictions = new ArrayList<>(); + List seenLabels = new ArrayList<>(); + + PrequentialResult result = run((OnLabel) (instance, predProbs) -> { + assertProbabilities(predProbs); + seenPredictions.add(Utils.maxIndex(predProbs)); + seenLabels.add((int) Math.round(instance.getData().classValue())); + }); + + assertEquals(N, seenPredictions.size()); + assertEquals(result.predictions, seenPredictions); + assertEquals(result.targets, seenLabels); + } + + @Test + public void onWindowCloseFiresPerWindowWithTheWindowsInstances() { + List windows = new ArrayList<>(); + List windowLabels = new ArrayList<>(); + + PrequentialResult result = run((OnWindowClose) (predProbs, labels) -> { + assertEquals(predProbs.length, labels.length); + windows.add(predProbs); + windowLabels.add(labels); + }); + + assertEquals(N / WINDOW, windows.size()); + + List flatPredictions = new ArrayList<>(); + List flatLabels = new ArrayList<>(); + for (int w = 0; w < windows.size(); ++w) { + assertEquals(WINDOW, windows.get(w).length); + for (double[] predProbs : windows.get(w)) { + assertProbabilities(predProbs); + flatPredictions.add(Utils.maxIndex(predProbs)); + } + for (int label : windowLabels.get(w)) + flatLabels.add(label); + } + + assertEquals(result.predictions, flatPredictions); + assertEquals(result.targets, flatLabels); + } + + @Test + public void trailingPartialWindowIsClosed() { + RandomTreeGenerator stream = stream(); + List windowSizes = new ArrayList<>(); + + PrequentialResult result = new HookableEvaluationLoop() + .registerWindowed(windowedEvaluator()) + .register((OnWindowClose) (predProbs, labels) -> windowSizes.add(labels.length)) + .run(stream, learner(stream), 250, WINDOW, false, false); + + assertEquals(3, windowSizes.size()); + assertEquals(Integer.valueOf(WINDOW), windowSizes.get(0)); + assertEquals(Integer.valueOf(WINDOW), windowSizes.get(1)); + assertEquals(Integer.valueOf(50), windowSizes.get(2)); + assertEquals(3, result.windowedResults.size()); + } + + /** A hook implementing both interfaces should be wired up by a single register() call. */ + @Test + public void combinedHookReceivesBothCallbacks() { + class Both implements OnLabel, OnWindowClose { + int labels = 0; + int windows = 0; + + @Override + public void onLabel(Example instance, double[] predProbs) { + this.labels++; + } + + @Override + public void onWindowClose(double[][] predProbs, int[] labels) { + this.windows++; + } + } + + Both hook = new Both(); + run(hook); + + assertEquals(N, hook.labels); + assertEquals(N / WINDOW, hook.windows); + } + + @Test + public void hookImplementingNeitherInterfaceIsRejected() { + try { + new HookableEvaluationLoop().register(new Hook() { + }); + fail("expected IllegalArgumentException"); + } catch (IllegalArgumentException expected) { + assertTrue(expected.getMessage().contains("OnLabel")); + } + } + + /** + * An untrained learner votes all zeros, which cannot be normalized. Those rows should come + * through as zeros rather than throwing. + */ + @Test + public void zeroVotesSurviveAsZeros() { + RandomTreeGenerator stream = stream(); + List firstRows = new ArrayList<>(); + + new HookableEvaluationLoop() + .register((OnLabel) (instance, predProbs) -> { + if (firstRows.size() < 5) + firstRows.add(predProbs); + }) + .run(stream, learner(stream), 5, WINDOW, false, false); + + assertEquals(5, firstRows.size()); + double sum = 0; + for (double p : firstRows.get(0)) + sum += p; + assertEquals(0.0, sum, 1e-9); + } + + @Test + public void nullEvaluatorsAreAccepted() { + RandomTreeGenerator stream = stream(); + PrequentialResult result = new HookableEvaluationLoop() + .registerBasic(null) + .registerWindowed(null) + .run(stream, learner(stream), N, WINDOW, false, false); + + assertEquals(null, result.cumulativeResults); + assertTrue(result.windowedResults.isEmpty()); + } + + /** Rows sum to one, or to zero while the learner has nothing to say. */ + private static void assertProbabilities(double[] predProbs) { + double sum = 0; + for (double p : predProbs) { + assertTrue("probability out of range: " + p, p >= 0 && p <= 1); + sum += p; + } + assertTrue("row sums to " + sum, Math.abs(sum - 1) < 1e-9 || sum == 0); + } +}