Bitstamp v2 Trades feed

xchange-bitstamp/src/main/java/info/bitrich/xchangestream/bitstamp/v2/BitstampStreamingMarketDataService.java

@@ -1,10 +1,16 @@
 package info.bitrich.xchangestream.bitstamp.v2;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+
+// old version?
 import info.bitrich.xchangestream.bitstamp.dto.BitstampOrderBook;
-import info.bitrich.xchangestream.bitstamp.dto.BitstampWebSocketTransaction;
+
+
+import info.bitrich.xchangestream.bitstamp.v2.dto.BitstampWebSocketTrade;
+
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
+
 import io.reactivex.Observable;
 import org.knowm.xchange.bitstamp.BitstampAdapters;
 import org.knowm.xchange.currency.CurrencyPair;
@@ -13,7 +19,6 @@
 import org.knowm.xchange.dto.marketdata.Trade;
 import org.knowm.xchange.exceptions.NotAvailableFromExchangeException;
 
-import java.util.Date;
 
 /**
  * Bitstamp WebSocket V2 Streaming Market Data Service implementation
@@ -64,10 +69,18 @@ public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
         return service.subscribeChannel(channelName, "trade")
                 .map(s -> {
                     ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
-                    BitstampWebSocketTransaction transactions = mapper.treeToValue(s.get("data"), BitstampWebSocketTransaction.class);
-                    transactions = new BitstampWebSocketTransaction(new Date().getTime() / 1000L, transactions.getTid(),
-                            transactions.getPrice(), transactions.getAmount(), transactions.getType());
-                    return BitstampAdapters.adaptTrade(transactions, currencyPair, 1000);
+                    // v2 websocket trade data
+                    BitstampWebSocketTrade transactions = mapper.treeToValue(s.get("data"), BitstampWebSocketTrade.class);
+                    // convert
+                    org.knowm.xchange.bitstamp.dto.marketdata.BitstampTransaction bitstampTransaction = 
+                    		new org.knowm.xchange.bitstamp.dto.marketdata.BitstampTransaction(
+                    				transactions.getDate(), 
+                    				transactions.getTid(), 
+                    				transactions.getPrice(), 
+                    				transactions.getAmount(), 
+                    				transactions.getType()); 
+
+                    return BitstampAdapters.adaptTrade(bitstampTransaction, currencyPair, 1000);
                 });
     }
 

xchange-bitstamp/src/main/java/info/bitrich/xchangestream/bitstamp/v2/BitstampStreamingService.java

@@ -9,6 +9,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.ArrayList;
 
 /**
  * Bitstamp WebSocket V2 streaming service implementation
@@ -19,10 +20,17 @@ public class BitstampStreamingService extends JsonNettyStreamingService {
 
     private static final String JSON_CHANNEL = "channel";
     private static final String JSON_EVENT = "event";
-    private static final String JSON_DATA = "data";
-
+
+    // can be: trade, data...
+    private static final ArrayList<String> JSON_DATA = new ArrayList<String>();
+
     public BitstampStreamingService(String apiUrl) {
         super(apiUrl, Integer.MAX_VALUE);
+
+        // add events
+        JSON_DATA.add("trade");
+        // not yet tested
+        //JSON_DATA.add("data");
     }
 
     @Override
@@ -56,8 +64,10 @@ protected void handleMessage(JsonNode message) {
             LOG.warn("The message has been received from disconnected channel '{}'. Skipped.", channel);
             return;
         }
-        if (event.equals(JSON_DATA)) {
-            super.handleMessage(message);
+        // event: trade
+        //if (event.equals(JSON_DATA)) {
+        if (JSON_DATA.contains(event)) {
+        	super.handleMessage(message);
         }
     }
 

xchange-bitstamp/src/main/java/info/bitrich/xchangestream/bitstamp/v2/dto/BitstampWebSocketTrade.java

@@ -0,0 +1,93 @@
+// Bitstamp v2 - websocket api
+// JSON mapping class
+
+/*  actual message
+{
+	"data":{
+		"microtimestamp":"1574158935282494",
+		"amount":0.01000309,
+		"buy_order_id":4352888198,
+		"sell_order_id":4352886848,
+		"amount_str":"0.01000309",
+		"price_str":"7360.72",
+		"timestamp":"1574158935",
+		"price":7360.72,
+		"type":0,
+		"id":100568768
+		},
+	"event":"trade",
+	"channel":"live_trades_btceur"
+}
+*/
+
+package info.bitrich.xchangestream.bitstamp.v2.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.math.BigDecimal;
+
+
+public class BitstampWebSocketTrade {
+
+	private long date;
+	private long tid;
+	private BigDecimal price;
+	private BigDecimal amount;
+	private int type;
+
+
+	public BitstampWebSocketTrade(
+			@JsonProperty("timestamp") long date, 
+			@JsonProperty("id") long tid,
+			@JsonProperty("price_str") String price_str, 
+			@JsonProperty("amount_str") String amount_str,
+			@JsonProperty("order_type") int type) {
+
+		this.date = date;
+		this.tid = tid;
+		this.price = new BigDecimal(price_str);
+		this.amount = new BigDecimal(amount_str);
+		this.type = type;
+	}
+
+	// GETTERS, SETTERS
+	public long getDate() {
+		return date;
+	}
+
+	public void setDate(long date) {
+		this.date = date;
+	}
+
+	public long getTid() {
+		return tid;
+	}
+
+	public void setTid(long tid) {
+		this.tid = tid;
+	}
+
+	public BigDecimal getPrice() {
+		return price;
+	}
+
+	public void setPrice(BigDecimal price) {
+		this.price = price;
+	}
+
+	public BigDecimal getAmount() {
+		return amount;
+	}
+
+	public void setAmount(BigDecimal amount) {
+		this.amount = amount;
+	}
+
+	public int getType() {
+		return type;
+	}
+
+	public void setType(int type) {
+		this.type = type;
+	}
+
+}