factor App.java into domain classes

build.gradle

@@ -20,6 +20,8 @@ dependencies {
 
     implementation 'com.espertech:esper:7.1.0'
     implementation 'com.oanda.v20:v20:3.0.21'
+    implementation 'com.fasterxml.jackson.core:jackson-databind:2.9.5'
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-csv:2.9.5'
 }
 
 repositories {

src/main/java/AccountInfo.java

@@ -0,0 +1,18 @@
+
+
+public class AccountInfo {
+
+	// TODO use properties file
+
+	public String accountID() {
+		return "101-001-7935538-001";
+	}
+	
+	public boolean isLiveAccount() {
+		return false;
+	}
+	
+	public String accessToken() {
+		return "9a480f0b83e987f4015cf0846790c7d9-695ced635526744abd61bdf0e2ae8b71";
+	}
+}

src/main/java/App.java

@@ -1,75 +1,30 @@
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
+import java.io.*;
 import java.net.MalformedURLException;
 import java.net.SocketTimeoutException;
 import java.net.URL;
 import java.net.URLConnection;
-import java.net.URLEncoder;
 
 import javax.net.ssl.HttpsURLConnection;
-//import com.espertech.esper.client.*;
 
+import com.espertech.esper.client.EPServiceProvider;
+import com.espertech.esper.client.EPServiceProviderManager;
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.MappingIterator;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.dataformat.csv.CsvMapper;
+import com.fasterxml.jackson.dataformat.csv.CsvSchema;
 public class App {
 
     public static void main(String[] args) {
+		//EPServiceProvider esper = EPServiceProviderManager.getDefaultProvider();
 
-		//EPServiceProvider engine = EPServiceProviderManager.getDefaultProvider();
+		File f = new File("/home/alx/Nextcloud/projects/ATS_Esper/EURUSD-2017-01-small.csv");
+		// new CSVReader(f).run(new DebugProcessor());
 
-		try {
+		AccountInfo acctInfo = new AccountInfo();
 		String[] instruments = new String[] { "USD_CAD", "EUR_USD", "USD_JPY" };
-
-			String query = String.format("instruments=%s", 
-										 URLEncoder.encode(String.join(",", instruments), "UTF-8"));
-			
-			openConnection("https://stream-fxpractice.oanda.com/v3/accounts/101-001-7935538-001/pricing/stream?" + query);
-		} catch (IOException e) {
-			e.printStackTrace();
-		}
-    }
-
-	private static void openConnection(String urlStr)
-		throws IOException
-	{
-		HttpsURLConnection httpConn = null;
-		String line = null;
-		try {
-			URL url = new URL(urlStr);
-			URLConnection urlConn = url.openConnection();
-			if (!(urlConn instanceof HttpsURLConnection)) {
-				throw new IOException ("URL is not an Https URL");
-			}
-			httpConn = (HttpsURLConnection)urlConn;
-			httpConn.setAllowUserInteraction(false);
-			httpConn.setInstanceFollowRedirects(true);
-			httpConn.setRequestMethod("GET");
-			httpConn.setReadTimeout(50 * 1000);
-			httpConn.setRequestProperty("Authorization", "Bearer 9a480f0b83e987f4015cf0846790c7d9-695ced635526744abd61bdf0e2ae8b71");
-			BufferedReader is =
-                new BufferedReader(new InputStreamReader(httpConn.getInputStream()));
-
-			while ((line = is.readLine( )) != null) {
-				System.out.println(line);
-                // Message msg = Message.obtain();
-                // msg.what=1;
-                // msg.obj=line;
-                // mhandler.sendMessage(msg);
-
-			}
-		} catch (MalformedURLException e) {
-			e.printStackTrace();
-		} catch(SocketTimeoutException e){
-			e.printStackTrace();
-		} catch (IOException e) {
-			e.printStackTrace();
-			//Message msg = Message.obtain();
-            //msg.what=2;
-            //BufferedInputStream in = new BufferedInputStream(httpConn.getErrorStream());
-            //line = new String(readStream(in));
-            //msg.obj = line;
-            //mhandler.sendMessage(msg);
-		} finally {
-			httpConn.disconnect();
-		}
+		new OANDAReader(acctInfo, instruments).run(new DebugProcessor());
 	}
 }

src/main/java/CSVReader.java

@@ -0,0 +1,32 @@
+
+import java.io.File;
+import java.io.IOException;
+
+import com.espertech.esper.client.EPServiceProvider;
+import com.fasterxml.jackson.databind.MappingIterator;
+import com.fasterxml.jackson.dataformat.csv.CsvMapper;
+import com.fasterxml.jackson.dataformat.csv.CsvSchema;
+
+public class CSVReader /*implements TickStreamReader*/ {
+	File file;
+
+
+	public CSVReader(File file) {
+		this.file = file;
+	}
+
+	public void run(TickProcessor processor) {
+		CsvMapper mapper = new CsvMapper();
+		CsvSchema schema = mapper.schemaFor(TrueFXTick.class);
+
+		try {
+			MappingIterator<TrueFXTick> it = mapper.readerFor(TrueFXTick.class).with(schema).readValues(file);
+			while (it.hasNextValue()) {
+				TrueFXTick value = it.nextValue();
+				processor.process(value);
+			}
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+	}
+}

src/main/java/DebugProcessor.java

@@ -0,0 +1,7 @@
+
+
+public class DebugProcessor implements TickProcessor {
+	public void process(Tick tick) {
+		System.out.println(tick);
+	}
+}

src/main/java/OANDAReader.java

@@ -0,0 +1,100 @@
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.UnsupportedEncodingException;
+import java.net.MalformedURLException;
+import java.net.SocketTimeoutException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.net.URLEncoder;
+
+import javax.crypto.spec.OAEPParameterSpec;
+import javax.net.ssl.HttpsURLConnection;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+
+public class OANDAReader {
+    AccountInfo accountInfo;
+	String[] instruments;
+
+
+	public OANDAReader(AccountInfo accountInfo, String[] instruments) {
+		this.accountInfo = accountInfo;
+		this.instruments = instruments;
+	}
+
+    public void run(TickProcessor processor) {
+		readConnection(streamURL(), processor);
+	}
+
+	private String streamURL() {
+		String type = accountInfo.isLiveAccount() ? "fxlive" : "fxpractice";
+		String instrList = String.join(",", instruments);
+		String query = "";
+		try {
+			query = String.format("instruments=%s", URLEncoder.encode(instrList, "UTF-8"));
+		} catch (UnsupportedEncodingException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+
+		return String.format("https://stream-%s.oanda.com/v3/accounts/%s/pricing/stream?%s",
+							 type, accountInfo.accountID(), query);
+	}
+		
+	private static void readConnection(String urlStr, TickProcessor processor) {
+		HttpsURLConnection httpConn = null;
+		String line = null;
+		try {
+			URL url = new URL(urlStr);
+			URLConnection urlConn = url.openConnection();
+			if (!(urlConn instanceof HttpsURLConnection)) {
+				throw new IOException ("URL is not an https URL");
+			}
+			httpConn = (HttpsURLConnection)urlConn;
+			httpConn.setAllowUserInteraction(false);
+			//httpConn.setInstanceFollowRedirects(true);
+			httpConn.setRequestMethod("GET");
+			//httpConn.setReadTimeout(50 * 1000);
+			httpConn.setRequestProperty("Authorization", "Bearer 9a480f0b83e987f4015cf0846790c7d9-695ced635526744abd61bdf0e2ae8b71");
+			BufferedReader is =
+                new BufferedReader(new InputStreamReader(httpConn.getInputStream()));
+
+			while ((line = is.readLine( )) != null) {
+				processLine(line, processor);
+			}
+		} catch (MalformedURLException e) {
+			e.printStackTrace();
+		} catch(SocketTimeoutException e){
+			e.printStackTrace();
+		} catch (IOException e) {
+			e.printStackTrace();
+		} finally {
+			httpConn.disconnect();
+		}
+	}
+
+	static ObjectMapper mapper = new ObjectMapper(); // create once, reuse
+
+	private static void processLine(String line, TickProcessor processor) {
+		if (line.indexOf ("PRICE") > -1) {
+			OANDATick tick;
+			try {
+				tick = mapper.readValue(line, OANDATick.class);
+				processor.process(tick);
+			} catch (JsonParseException | JsonMappingException e) {
+				e.printStackTrace();
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
+		} else if (line.indexOf ("HEARTBEAT") > -1) {
+			// ignore
+		} else {
+			System.out.println("Unknown type: " + line);
+		}
+	}
+}

src/main/java/OANDATick.java

@@ -0,0 +1,75 @@
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Date;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+
+/**
+ * The specification of an Account-specific Price.
+ *
+ * {"type":"PRICE","time":"2018-04-05T20:35:20.983907480Z","bids":[{"price":"1.22376","liquidity":10000000}],"asks":[{"price":"1.22386","liquidity":10000000}],"closeoutBid":"1.22361","closeoutAsk":"1.22401","status":"tradeable","tradeable":true,"instrument":"EUR_USD"}
+ */
+public class OANDATick implements Tick {
+	public String type;
+	@JsonFormat(pattern="yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
+	private Date time;
+	public PriceBucket[] bids;
+	public PriceBucket[] asks;
+	public BigDecimal closeoutBid;
+	public BigDecimal closeoutAsk;
+	public String status;
+	public boolean tradeable;
+	private String instrument;
+
+	
+	/**
+	 * @return the instrument
+	 */
+	@Override
+	public String getInstrument() {
+		return instrument;
+	}
+
+	/**
+	 * @param instrument the instrument to set
+	 */
+	public void setInstrument(String instrument) {
+		this.instrument = instrument;
+	}
+
+	/**
+	 * @return the time
+	 */
+	@Override
+	public Date getTime() {
+		return time;
+	}
+
+	/**
+	 * @param time the time to set
+	 */
+	public void setTime(Date time) {
+		this.time = time;
+	}
+
+	/**
+	 * @return the bid
+	 */
+	@Override
+	public BigDecimal getBid() {
+		return bids[0].price;
+	}
+
+	/**
+	 * @return the ask
+	 */
+	@Override
+	public BigDecimal getAsk() {
+		return asks[0].price;
+	}
+
+	public String toString() {
+		return String.format("OANDATick[%s,%s]", getInstrument(), getBid());
+	}
+}

src/main/java/PriceBucket.java

@@ -0,0 +1,16 @@
+
+import java.math.BigDecimal;
+
+/**
+ * PriceBucket represents a price available for an amount of liquidity.
+ *
+ * {"price":"1.22376","liquidity":10000000}
+ */
+public class PriceBucket {
+	public BigDecimal price;
+	public Integer liquidity;
+
+	public String toString() {
+		return String.format("PriceBucket[%f, %d]", price, liquidity);
+	}
+}

src/main/java/Tick.java

@@ -0,0 +1,26 @@
+
+import java.math.BigDecimal;
+import java.util.Date;
+
+public interface Tick {
+	
+	/**
+	 * @return the instrument
+	 */
+	public String getInstrument();
+
+	/**
+	 * @return the time
+	 */
+	public Date getTime();
+
+	/**
+	 * @return the bid
+	 */
+	public BigDecimal getBid();
+	
+	/**
+	 * @return the ask
+	 */
+	public BigDecimal getAsk();
+}

src/main/java/TickProcessor.java

@@ -0,0 +1,3 @@
+public interface TickProcessor {
+	public void process(Tick tick);
+}

src/main/java/TrueFXTick.java

@@ -0,0 +1,87 @@
+
+import java.math.BigDecimal;
+import java.util.Date;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+
+/**
+ * TrueFXTick holds a single tick read from TrueFX data archive csv
+ * file.
+ *
+ * A line looks like:
+ * 	 EUR/USD,20170102 00:00:00.803,1.0523,1.05307
+ */
+@JsonPropertyOrder({ "instrument", "time", "bid", "ask" })
+public class TrueFXTick implements Tick {
+	private String instrument;
+	@JsonFormat(pattern="yyyyMMdd HH:mm:ss.SSS")
+	private Date time;
+	private BigDecimal bid;
+	private BigDecimal ask;
+
+
+	public String toString() {
+		return String.format("TrueFXTick[%s,%s,%s,%s]", instrument, time, bid, ask);
+	}
+
+	/**
+	 * @return the instrument
+	 */
+	@Override
+	public String getInstrument() {
+		return instrument;
+	}
+
+	/**
+	 * @param instrument the instrument to set
+	 */
+	public void setInstrument(String instrument) {
+		this.instrument = instrument;
+	}
+
+	/**
+	 * @return the time
+	 */
+	@Override
+	public Date getTime() {
+		return time;
+	}
+
+	/**
+	 * @param time the time to set
+	 */
+	public void setTime(Date time) {
+		this.time = time;
+	}
+
+	/**
+	 * @return the bid
+	 */
+	@Override
+	public BigDecimal getBid() {
+		return bid;
+	}
+
+	/**
+	 * @param bid the bid to set
+	 */
+	public void setBid(BigDecimal bid) {
+		this.bid = bid;
+	}
+
+	/**
+	 * @return the ask
+	 */
+	@Override
+	public BigDecimal getAsk() {
+		return ask;
+	}
+
+	/**
+	 * @param ask the ask to set
+	 */
+	public void setAsk(BigDecimal ask) {
+		this.ask = ask;
+	}
+}