Initial implementation of streaming forecasting

egads

@@ -0,0 +1,29 @@
+#!/usr/bin/perl
+
+sub usage {
+    print "Usage: egads <command> args\n";
+    print "  Commands:\n";
+    print "    StreamForecast\n";
+    print "    TrainForecastingModel\n";
+    print "    MakeEmptyModel\n";
+    exit (0);
+}
+
+sub launch {
+    my $cmd = "java -cp lib/OpenForecast-0.5.0.jar:target/egads-jar-with-dependencies.jar -Dlog4j.configurationFile=log4j.xml com.yahoo.egads." . join (' ', @ARGV);
+    print STDERR $cmd, "\n";
+    system $cmd;
+}
+
+my $command = $ARGV[0];
+
+if ($command eq "MakeEmptyModel") {
+    launch();
+} elsif ($command eq "StreamForecast") {
+    launch();
+} elsif ($command eq "TrainForecastingModel") {
+    launch();
+} else {
+    usage();
+}
+

pom.xml

@@ -20,6 +20,11 @@
 
     <dependencies>
         <!-- needed by egads -->
+        <dependency>
+          <groupId>gnu.getopt</groupId>
+          <artifactId>java-getopt</artifactId>
+          <version>1.0.13</version>
+        </dependency>
         <dependency>
           <groupId>org.apache.logging.log4j</groupId>
           <artifactId>log4j-api</artifactId>

src/main/java/com/yahoo/egads/Egads.java

@@ -2,10 +2,13 @@
 
 package com.yahoo.egads;
 
-import java.util.Properties;
 import java.io.FileInputStream;
 import java.io.InputStream;
-import com.yahoo.egads.utilities.*;
+import java.util.Properties;
+
+import com.yahoo.egads.utilities.FileInputProcessor;
+import com.yahoo.egads.utilities.InputProcessor;
+import com.yahoo.egads.utilities.StdinProcessor;
 
 /*
  * Call stack.

src/main/java/com/yahoo/egads/StreamForecast.java

@@ -0,0 +1,92 @@
+package com.yahoo.egads;
+
+import gnu.getopt.Getopt;
+
+import java.util.HashMap;
+import java.util.Scanner;
+
+import com.yahoo.egads.data.FileModelStore;
+import com.yahoo.egads.data.Model;
+import com.yahoo.egads.data.ModelStore;
+import com.yahoo.egads.data.TimeSeries;
+import com.yahoo.egads.models.tsmm.StreamingOlympicModel;
+import com.yahoo.egads.models.tsmm.TimeSeriesStreamingModel;
+
+public class StreamForecast {
+	public static void main(String[] args) {
+		HashMap<Integer,String> options = processOptions(args);
+		Scanner sc = new Scanner(System.in);
+		ModelStore ms = new FileModelStore ("models");
+		while (sc.hasNextLine()) {
+			String line = sc.nextLine();
+			String[] fields = line.split(",");
+			String series;
+			int timestamp;
+			double measured;
+			double forecast = Double.NaN;
+			try {
+				series = fields[0];
+				timestamp = Integer.parseInt(fields[1]);
+				measured = Double.parseDouble(fields[2]);
+				TimeSeriesStreamingModel model = (TimeSeriesStreamingModel) ms.getModel(series, Model.ModelType.FORECAST);
+				if (model == null) {
+					model = new StreamingOlympicModel();
+					ms.storeModel(series, model);
+					System.err.println ("No such model " + series);
+				}
+				TimeSeries.Entry e = new TimeSeries.Entry(timestamp, (float)measured);
+				forecast = model.predict(e);
+				model.update(e);
+				if (!options.containsKey(new Integer('q'))) {
+					System.out.println(String.join(",", series, String.format("%d", timestamp), String.format("%f", measured), String.format("%f", forecast)));
+				}
+			} catch (Exception e) {
+				System.err.println("Invalid input line " + line);
+				continue;
+			}
+		}
+		ms.writeCachedModels();
+		if (options.containsKey(new Integer('t'))) {
+			for (Model m : ms.getCachedModels()) {
+				System.out.println(m.errorSummaryString());
+			}
+		}
+		sc.close();
+	}
+
+	public static void usage() {
+		System.out.println ("Usage: StreamForecast [-m <default modeltype>] [-p <properties file>] [-t] [-q] [-h]");
+		System.out.println ("  Modeltypes:");
+		System.out.println ("    sos: Streaming Olympic Scoring");
+		System.out.println ("  Default preperties file is config.ini");
+		System.out.println ("  -t: Run in model test mode.  This outputs error stats at the end of the model run");
+		System.exit(0);
+	}
+
+	public static HashMap<Integer, String> processOptions (String[] args) {
+		HashMap<Integer, String> result = new HashMap<Integer, String>();
+		// defaults
+		result.put(new Integer('m'),  "sos");
+		result.put(new Integer('p'), "config.ini");
+
+		Getopt g = new Getopt("TrainForecastingModel", args, "m:p:htq");
+		int c;
+		while ((c = g.getopt()) != -1) {
+			switch (c) {
+			case 'm':
+			case 'n':
+			case 'p':
+				result.put(c, g.getOptarg());
+				break;
+			case 't':
+			case 'q':
+				result.put(c, "True");
+				break;
+			case 'h':
+				usage();
+			}
+		}
+		return result;
+	}
+
+}

src/main/java/com/yahoo/egads/TrainForecastingModel.java

@@ -0,0 +1,87 @@
+package com.yahoo.egads;
+
+import gnu.getopt.Getopt;
+
+import java.io.FileInputStream;
+import java.util.HashMap;
+import java.util.Properties;
+import java.util.Scanner;
+
+import com.yahoo.egads.data.FileModelStore;
+import com.yahoo.egads.data.ModelFactory;
+import com.yahoo.egads.data.ModelStore;
+import com.yahoo.egads.data.TimeSeries;
+import com.yahoo.egads.models.tsmm.TimeSeriesModel;
+
+public class TrainForecastingModel {
+	public static void main(String[] args) throws Exception {
+		HashMap<Integer,String> options = processOptions(args);
+		HashMap<String, TimeSeries.DataSequence> inputs = new HashMap<String, TimeSeries.DataSequence>();
+		Scanner sc = new Scanner(System.in);
+		ModelStore ms = new FileModelStore ("models");
+		Properties osProps = new Properties();
+		osProps.load (new FileInputStream(options.get(new Integer('p'))));
+		while (sc.hasNextLine()) {
+			String line = sc.nextLine();
+			String[] fields = line.split(",");
+			String series;
+			int timestamp;
+			float measured;
+			try {
+				series = fields[0];
+				timestamp = Integer.parseInt(fields[1]);
+				measured = Float.parseFloat(fields[2]);
+			} catch (Exception e) {
+				System.err.println("Invalid input line " + line);
+				continue;
+			}
+			if (!inputs.containsKey(series)) {
+				inputs.put(series, new TimeSeries.DataSequence());
+			}
+			TimeSeries.DataSequence seq = inputs.get(series);
+			seq.add(new TimeSeries.Entry(timestamp, measured));				
+		}
+		sc.close();
+		ModelFactory mf = new ModelFactory(osProps);
+		for (String series : inputs.keySet()) {
+			TimeSeries.DataSequence seq = inputs.get(series);
+			TimeSeriesModel m = mf.getTSModel(options.get(new Integer('m')));
+			if (m != null) {
+				m.train(seq);
+				ms.storeModel(series, m);
+			}
+		}
+	}
+
+	public static void usage() {
+		System.out.println ("Usage: TrainForecastingModel [-m <modeltype>] [-p <properties file>] [-h]");
+		System.out.println ("  Modeltypes:");
+		System.out.println ("    sos: Streaming Olympic Scoring");
+		System.out.println ("  Default preperties file is config.ini");
+		System.exit(0);
+	}
+
+	public static HashMap<Integer, String> processOptions (String[] args) {
+		HashMap<Integer, String> result = new HashMap<Integer, String>();
+		// defaults
+		result.put(new Integer('m'),  "sos");
+		result.put(new Integer('p'), "config.ini");
+
+		Getopt g = new Getopt("TrainForecastingModel", args, "m:p:h");
+		int c;
+		while ((c = g.getopt()) != -1) {
+			switch (c) {
+			case 'm':
+				result.put(c, g.getOptarg());
+				break;
+			case 'p':
+				result.put(c, g.getOptarg());
+				break;
+			case 'h':
+				usage();
+			}
+		}
+		return result;
+	}
+
+}

src/main/java/com/yahoo/egads/data/FileModelStore.java

@@ -5,20 +5,36 @@
 import java.io.FileOutputStream;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.util.Collection;
+import java.util.HashMap;
 
 public class FileModelStore implements ModelStore {
+	HashMap <String, Model> cache;
 	String path;
+    protected static org.apache.logging.log4j.Logger logger = org.apache.logging.log4j.LogManager.getLogger(FileModelStore.class.getName());
+
 	public FileModelStore (String path) {
-		File dir = new File (path);
-		dir.mkdirs();
 		this.path = path;
+		cache = new HashMap<String, Model>();
+		new File (path).mkdirs();
+	}
+
+	private String getFilename (String tag, Model.ModelType type) {
+		String filename = tag.replaceAll("[^\\w_-]", "_");
+		if (type == Model.ModelType.ANOMALY) {
+			filename = "anomaly." + filename;
+		} else if (type == Model.ModelType.FORECAST) {
+			filename =  "forecast." + filename;
+		}
+		return filename;
 	}
 
 	@Override
 	public void storeModel(String tag, Model m) {
-		String filename = tag.replaceAll("[^\\w_-]", "_");
+		String filename = getFilename(tag, m.getModelType());
 		String fqn = path + "/" + filename;
 		try {
+			m.clearModified();
 			ObjectOutputStream o = new ObjectOutputStream(new FileOutputStream (fqn));
 			o.writeObject(m);
 			o.close();
@@ -28,18 +44,36 @@ public void storeModel(String tag, Model m) {
 	}
 
 	@Override
-	public Model retrieveModel(String tag) {
-		String filename = tag.replaceAll("[^\\w_-]", "_");
+	public Model getModel(String tag, Model.ModelType type) {
+		String filename = getFilename(tag, type);
+		if (cache.containsKey(filename)) {
+			return cache.get(filename);
+		}
 		String fqn = path + "/" + filename;
 		Model m = null;
 		try {
 			ObjectInputStream o = new ObjectInputStream(new FileInputStream(fqn));
 			m =  (Model) o.readObject();
 			o.close();
+			cache.put(filename, m);
+			return m;
 		} catch (Exception e) {
-			e.printStackTrace();
+			logger.debug("Model not found: " + tag);
 		}
-		return m;
+		return null;
+	}
+	public void writeCachedModels() {
+		for (String key : cache.keySet()) {
+			Model model = cache.get(key);
+			if (model.isModified()) {
+//				The key always has the model type prepended - remove it before storing
+				key = key.replaceFirst("[a-zA-Z]*\\.", "");
+				storeModel(key, model);
+			}
+		}
+	}
+	public Collection<Model> getCachedModels() {
+		return cache.values();
 	}
 
 }