package io.pivotal.demo import java.io.BufferedReader import java.io.File import java.io.InputStream import java.io.InputStreamReader import scala.util.parsing.json.JSON import org.apache.spark.SparkConf import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS import org.apache.spark.mllib.evaluation.MulticlassMetrics import org.apache.spark.mllib.feature.StandardScaler import org.apache.spark.mllib.feature.StandardScalerModel import org.apache.spark.mllib.linalg.Vectors import org.apache.spark.mllib.linalg.Vectors import org.apache.spark.mllib.regression.LabeledPoint import org.apache.spark.mllib.regression.LabeledPoint import org.apache.spark.mllib.regression.LinearRegressionModel import org.apache.spark.mllib.regression.LinearRegressionWithSGD import org.apache.spark.mllib.regression.StreamingLinearRegressionWithSGD import org.apache.spark.rdd._ import org.apache.spark.sql.DataFrame import org.apache.spark.sql.Row import org.apache.spark.sql.SQLContext import io.pivotal.gemfire.spark.connector._ /** * @author fmelo */ object StockInferenceDemo { val MODEL_PATH = "SparkModel" val SCALER_FILE = MODEL_PATH+"/scaler.obj" val conf = new SparkConf().setMaster("local[*]").setAppName("StockInferenceMLDemo") conf.set("spark.gemfire.locators", "geode-server[10334]"); conf.set("spark.hadoop.validateOutputSpecs", "false") conf.set("spark.files.overwrite","true") val sc = new SparkContext(conf); val sqlContext = new SQLContext(sc); val numIterations = 2000 val stepSize = 0.2 val algorithm = new LinearRegressionWithSGD() algorithm.setIntercept(true) algorithm.optimizer .setNumIterations(numIterations) .setStepSize(stepSize) def train() = { val df = sqlContext.gemfireOQL("SELECT t.entryTimestamp, t.close, t.ema, t.future_ema, t.rsi, t.ema_diff, t.low_diff, t.high_diff FROM /TechIndicators t"); df.registerTempTable("tech_indicators"); val result = sqlContext.sql( "select entryTimestamp, close, ema, future_ema, rsi, ema_diff, low_diff, high_diff from tech_indicators t where t.rsi NOT LIKE 'NaN' AND t.rsi NOT LIKE '0' order by entryTimestamp desc limit 10000") val rdd = result.rdd.cache() val dataset = rdd.map { line => // tech indicators val entryTimestamp = line.getLong(0) val close = line.getString(1).toDouble val ema = line.getString(2).toDouble val future_ema = line.getString(3).toDouble val rsi = line.getString(4).toDouble val ema_diff = line.getString(5).toDouble val low_diff = line.getString(6).toDouble val high_diff = line.getString(7).toDouble LabeledPoint(future_ema, Vectors.dense(close, ema, rsi)) }.cache() val scaler = new StandardScaler(withMean = true, withStd = true) .fit(dataset.map(x => x.features)) val scaledData = dataset .map(x => LabeledPoint(x.label, scaler.transform(Vectors.dense(x.features.toArray)))).cache() val splits = scaledData.randomSplit(Array(0.8, 0.2), seed = 11L) val trainingData = splits(0).cache() val testingData = splits(1).cache() println("\nGot " + dataset.count() + " values from Gem. Using " + trainingData.count() + " for training and "+ testingData.count() + " for testing\n") val model = algorithm.run(trainingData) val modelFileDirectory = new File(MODEL_PATH) if (modelFileDirectory.exists()) Utils.deleteRecursive(modelFileDirectory) // save the trained model model.save(sc, MODEL_PATH) // save the scaler sc.parallelize(Seq(scaler), 1).saveAsObjectFile(SCALER_FILE) // Test model on training examples val valuesAndPreds = testingData.map { point => val prediction = model.predict(point.features) (point.label, point.features, prediction) } // Print out features, actual and predicted values... valuesAndPreds.foreach({case (v, f, p) => println(s"Features: ${f}, Predicted: ${p}, Actual: ${v}")}) val MSE = valuesAndPreds.map{case(v, f, p) => math.pow((v - p), 2)}.mean() println("training Mean Squared Error = " + MSE) } // Sample Input: {"entryTimestamp":250214653814611,"ema":"245.511394357033","future_ema":"245.511394357033","close":"245.53","rsi":"46.2483326678155","ema_diff":"0.0186056429666053","high_diff":"9.03","low_diff":"-1.03"} def evaluate() ={ // load the model val model = LinearRegressionModel.load(sc, MODEL_PATH) // load the scaler val scaler = sc.objectFile[StandardScalerModel](SCALER_FILE).first() val br = new BufferedReader(new InputStreamReader(System in)) while (true){ val line = br.readLine() val json = JSON.parseFull(line) val keyValueProps: Map[String,Any] = json.get.asInstanceOf[Map[String,Any]] val entryTimestamp = keyValueProps.get("entryTimestamp").get.asInstanceOf[Number].longValue() val close = keyValueProps.get("close").get.asInstanceOf[String].toDouble val ema = keyValueProps.get("ema").get.asInstanceOf[String].toDouble val rsi = keyValueProps.get("rsi").get.asInstanceOf[String].toDouble val ema_diff = keyValueProps.get("ema_diff").get.asInstanceOf[String].toDouble val low_diff = keyValueProps.get("low_diff").get.asInstanceOf[String].toDouble val high_diff = keyValueProps.get("high_diff").get.asInstanceOf[String].toDouble val input = scaler.transform(Vectors.dense(close, ema, rsi)) val prediction = model.predict(input) // save on Gem using connector. /* val pairRDD = sc.parallelize(Array( ("entryTimestamp", entryTimestamp), ("close", close), ("ema", ema), ("rsi", rsi), ("ema_diff", ema_diff), ("low_diff", low_diff), ("high_diff", high_diff), ("predicted", prediction) )) val outRdd = sc.parallelize(Array(jsonOutput)) outRdd.saveToGemfire("Predictions", item => (entryTimestamp, item)) * */ val jsonOutput = "{"+ "entryTimestamp : "+entryTimestamp + ", "+ "close : "+close + ", "+ "ema : "+ema + ", "+ "rsi : "+rsi + ", "+ "ema_diff : "+ema_diff + ", "+ "low_diff : "+low_diff + ", "+ "high_diff : "+high_diff + ", "+ "predicted : "+prediction+ "}" println(jsonOutput) println("\r\n") } } def printUsage()={ // spark-submit --class io.pivotal.demo.StockInferenceDemo --driver-memory 4G --executor-memory 2G --master local[*] StockInference-1.0.jar println("Usage: StockInferenceDemo ") } def main(args: Array[String]) { if (args.length!=1){ printUsage() } else if (args(0).equalsIgnoreCase("train")){ train() } else if (args(0).equalsIgnoreCase("evaluate")){ evaluate() } } }