Basic validation

cmd/watcher/validator.go

@@ -2,10 +2,18 @@ package main
 
 import (
 	"github.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus/gdax"
 
+	"sort"
 	"time"
 )
 
+type ByTradeId []tacitus.Trade
+
+func (t ByTradeId) Len() int           { return len(t) }
+func (t ByTradeId) Swap(i, j int)      { t[i], t[j] = t[j], t[i] }
+func (t ByTradeId) Less(i, j int) bool { return t[i].TradeId > t[j].TradeId }
+
 type validator struct {
 	db     tacitus.DatabaseClientService
 	logger tacitus.Logger
@@ -22,7 +30,7 @@ func (v *validator) stop() {
 }
 
 func (v *validator) validate(products ...string) {
-	v.ticker = time.NewTicker(10 * time.Minute)
+	v.ticker = time.NewTicker(24 * time.Hour)
 	v.done = make(chan bool)
 
 	go func() {
@@ -46,13 +54,41 @@ func (v *validator) validate(products ...string) {
 
 func (v *validator) emitProducts(products ...string) {
 	for _, p := range products {
-		go v.validateProduct(p)
+		v.validateProduct(p)
 	}
 }
 
 func (v *validator) validateProduct(product string) {
-	v.logger.Info("Staring validation of %v", product)
+	v.logger.Info("Validating %v", product)
 
+	groups := v.findMissingGroups(product)
+	v.getMissingTrades(product, groups)
+
+	v.logger.Info("DONE: %v", product)
+}
+
+func (v *validator) getMissingTrades(product string, groups [][]int) {
+	c := gdax.NewPublicClient()
+
+	for _, group := range groups {
+		v.logger.Info("%v", group)
+		for i := group[1]; i >= group[0]; i-- {
+			ts, _ := c.GetTradesBefore(product, i+1)
+			sort.Sort(ByTradeId(ts))
+
+			for _, t := range ts {
+				if t.TradeId < group[0] && t.TradeId > group[1] {
+					i = ts[len(ts)-1].TradeId
+				} else {
+					v.db.TradeService().CreateTrade(t)
+					i = t.TradeId
+				}
+			}
+		}
+	}
+}
+
+func (v *validator) findMissingGroups(product string) [][]int {
 	results, err := v.db.TradeService().TradesAfterAll(product, 0)
 	if err != nil {
 		v.logger.Info("Error getting all trades: %v", err)
@@ -62,7 +98,6 @@ func (v *validator) validateProduct(product string) {
 	current := 0
 	missing := [][]int{}
 	for results.Next() {
-
 		trade = results.Value().(tacitus.Trade)
 
 		if trade.TradeId != current+1 {
@@ -72,9 +107,5 @@ func (v *validator) validateProduct(product string) {
 		current = trade.TradeId
 	}
 
-	for _, i := range missing {
-		v.logger.Info("%v", i)
-	}
-
-	v.logger.Info("DONE: %v", product)
+	return missing
 }

gdax/public_client.go

@@ -0,0 +1,66 @@
+package gdax
+
+import (
+	"github.com/kcotugno/tacitus"
+
+	"encoding/json"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	endpoint = "https://api.gdax.com"
+)
+
+type PublicClient struct {
+	client http.Client
+}
+
+func NewPublicClient() *PublicClient {
+	c := PublicClient{}
+	c.client = http.Client{Timeout: 30 * time.Second}
+
+	return &c
+}
+
+func (c *PublicClient) GetTradesBefore(product string, id int) ([]tacitus.Trade, error) {
+	url := strings.Join([]string{endpoint, "products", product, "trades"}, "/")
+	url = strings.Join([]string{url, "?after=", strconv.Itoa(id)}, "")
+
+	resp, err := c.client.Get(url)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	results := make([]tradeResponse, 0)
+
+	decoder := json.NewDecoder(resp.Body)
+	if err = decoder.Decode(&results); err != nil {
+		return nil, err
+	}
+
+	trades := make([]tacitus.Trade, 0)
+	for _, t := range results {
+		var trade tacitus.Trade
+
+		trade.TradeId = t.TradeId
+		trade.Product = strings.ToUpper(product)
+		trade.Price = t.Price
+		trade.Size = t.Size
+		trade.Timestamp = t.Time
+
+		switch t.Side {
+		case "buy":
+			trade.Buy = true
+		case "sell":
+			trade.Sell = true
+		}
+
+		trades = append(trades, trade)
+	}
+
+	return trades, nil
+}

gdax/trade_response.go

@@ -0,0 +1,15 @@
+package gdax
+
+import (
+	"github.com/shopspring/decimal"
+
+	"time"
+)
+
+type tradeResponse struct {
+	Time    time.Time       `json:"time"`
+	TradeId int             `json:"trade_id"`
+	Price   decimal.Decimal `json:"price"`
+	Size    decimal.Decimal `json:"size"`
+	Side    string          `json:"side"`
+}