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); + } +}