Compare commits
3 Commits
master
...
refactor-r
Author | SHA1 | Date | |
---|---|---|---|
f324ba576c | |||
81ec3c1eb8 | |||
3513e845c6 |
14
Makefile
14
Makefile
@ -23,13 +23,13 @@ watcher: $(WATCHER)
|
||||
webapp: $(WEBAPP)
|
||||
|
||||
get:
|
||||
go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus
|
||||
go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/cmd/interval
|
||||
go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/cmd/watcher
|
||||
go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/cmd/webapp
|
||||
go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/gdax
|
||||
go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/gdax/websocket
|
||||
go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/postgres
|
||||
go get $(GETFLAGS) github.com/kcotugno/tacitus
|
||||
go get $(GETFLAGS) github.com/kcotugno/tacitus/cmd/interval
|
||||
go get $(GETFLAGS) github.com/kcotugno/tacitus/cmd/watcher
|
||||
go get $(GETFLAGS) github.com/kcotugno/tacitus/cmd/webapp
|
||||
go get $(GETFLAGS) github.com/kcotugno/tacitus/gdax
|
||||
go get $(GETFLAGS) github.com/kcotugno/tacitus/gdax/websocket
|
||||
go get $(GETFLAGS) github.com/kcotugno/tacitus/postgres
|
||||
|
||||
fmt: $(TACITUS_FILES) $(OSUTIL_FILES) $(POSTGRES_FILES) $(OPS_FILES) \
|
||||
$(GDAX_FILES) $(GDAX_WEBSOCKET_FILES) $(HTTP_FILES)
|
||||
|
@ -9,7 +9,7 @@ import (
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"git.kevincotugno.com/kcotugno/tacitus/postgres"
|
||||
"github.com/kcotugno/tacitus/postgres"
|
||||
)
|
||||
|
||||
func main() {
|
||||
|
@ -1,91 +1,27 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/gdax"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/gdax/websocket"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/ops"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/osutil"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/postgres"
|
||||
"github.com/kcotugno/tacitus/gdax"
|
||||
"github.com/kcotugno/tacitus/gdax/websocket"
|
||||
"github.com/kcotugno/tacitus/ops"
|
||||
"github.com/kcotugno/tacitus/osutil"
|
||||
"github.com/kcotugno/tacitus/postgres"
|
||||
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Parameters struct {
|
||||
Products []string
|
||||
|
||||
Host string
|
||||
Port int
|
||||
User string
|
||||
Name string
|
||||
Password string
|
||||
SSL string
|
||||
}
|
||||
|
||||
type StringList []string
|
||||
|
||||
func parseFlags() Parameters {
|
||||
var p Parameters
|
||||
p.Products = make([]string, 0)
|
||||
|
||||
host := flag.String("host", "localhost", "Database host")
|
||||
port := flag.Int("port", 5432, "Database port")
|
||||
user := flag.String("user", "gdax", "Database user")
|
||||
db := flag.String("name", "gdax", "Database name")
|
||||
password := flag.String("password", `""`, "Database password")
|
||||
ssl := flag.String("ssl", "disable", "Database ssl")
|
||||
|
||||
products := make(StringList, 0)
|
||||
flag.Var(&products, "product", "Product to watch. This option can "+
|
||||
"be specified multiple time")
|
||||
|
||||
flag.Parse()
|
||||
|
||||
for _, v := range products {
|
||||
if !tacitus.ValidProduct(v) {
|
||||
fmt.Printf("error: %v is not a valid product", v)
|
||||
os.Exit(128)
|
||||
}
|
||||
}
|
||||
|
||||
p.Products = []string(products)
|
||||
p.Host = *host
|
||||
p.Port = *port
|
||||
p.User = *user
|
||||
p.Name = *db
|
||||
p.Password = *password
|
||||
p.SSL = *ssl
|
||||
|
||||
return p
|
||||
}
|
||||
|
||||
func main() {
|
||||
params := parseFlags()
|
||||
if len(params.Products) == 0 {
|
||||
flag.PrintDefaults()
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
logger := osutil.NewLogger()
|
||||
|
||||
db := postgres.NewClient()
|
||||
db.Name = params.Name
|
||||
db.User = params.User
|
||||
db.Password = params.Password
|
||||
db.Host = params.Host
|
||||
db.Port = params.Port
|
||||
db.SslMode = params.SSL
|
||||
db.Name = "gdax"
|
||||
db.User = "gdax"
|
||||
db.SetLogger(logger)
|
||||
err := db.Open()
|
||||
if err != nil {
|
||||
logger.Info("Error openning database connection: %v", err)
|
||||
return
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
ws := websocket.NewClient()
|
||||
ls := &gdax.ListenerService{}
|
||||
@ -95,50 +31,31 @@ func main() {
|
||||
r := ops.Registrar{}
|
||||
r.Database = db
|
||||
r.Listener = ls
|
||||
r.SetLogger(logger)
|
||||
if err := r.Record(params.Products...); err != nil {
|
||||
r.Logger = logger
|
||||
if err := r.Record("ETH-USD", "BTC-USD"); err != nil {
|
||||
logger.Info("Error: %v", err)
|
||||
os.Exit(3)
|
||||
return
|
||||
}
|
||||
|
||||
defer ls.Close()
|
||||
|
||||
a := ops.Aggregator{}
|
||||
a.Database = db
|
||||
a.Logger = logger
|
||||
a.Products = params.Products
|
||||
a.Products = []string{"ETH-USD", "BTC-USD"}
|
||||
a.Start(30 * time.Second)
|
||||
|
||||
v := ops.Validator{}
|
||||
v.Database = db
|
||||
v.Logger = logger
|
||||
v.Products = params.Products
|
||||
v.Products = []string{"ETH-USD", "BTC-USD"}
|
||||
v.Start(30 * time.Second)
|
||||
|
||||
av := ops.AggregationValidator{}
|
||||
av.Database = db
|
||||
av.Logger = logger
|
||||
av.Products = params.Products
|
||||
av.Products = []string{"ETH-USD", "BTC-USD"}
|
||||
av.Intervals = []time.Duration{30 * time.Second}
|
||||
av.Start(30 * time.Second)
|
||||
|
||||
c := make(chan os.Signal)
|
||||
signal.Notify(c, os.Interrupt)
|
||||
|
||||
var done bool
|
||||
for !done {
|
||||
select {
|
||||
case <-c:
|
||||
done = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (l *StringList) String() string {
|
||||
return "List of strings"
|
||||
}
|
||||
|
||||
func (l *StringList) Set(value string) error {
|
||||
*l = append(*l, value)
|
||||
return nil
|
||||
t := make(chan bool)
|
||||
<-t
|
||||
}
|
||||
|
@ -1,9 +1,9 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus/http"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/osutil"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/postgres"
|
||||
"github.com/kcotugno/tacitus/http"
|
||||
"github.com/kcotugno/tacitus/osutil"
|
||||
"github.com/kcotugno/tacitus/postgres"
|
||||
"log"
|
||||
)
|
||||
|
||||
|
@ -1,12 +1,13 @@
|
||||
package gdax
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
|
||||
"errors"
|
||||
"sync"
|
||||
)
|
||||
|
||||
const buf = 1024
|
||||
|
||||
type ListenerService struct {
|
||||
Client ListenerClient
|
||||
Logger tacitus.Logger
|
||||
@ -14,19 +15,11 @@ type ListenerService struct {
|
||||
trades chan tacitus.Trade
|
||||
err chan error
|
||||
|
||||
subs []string
|
||||
|
||||
closed bool
|
||||
|
||||
shouldRestart bool
|
||||
restMu sync.Mutex
|
||||
|
||||
errorsMu sync.Mutex
|
||||
sendErrors bool
|
||||
dirty bool
|
||||
}
|
||||
|
||||
func (s *ListenerService) Open() error {
|
||||
if s.closed {
|
||||
if s.dirty {
|
||||
return errors.New("Already used")
|
||||
}
|
||||
|
||||
@ -36,9 +29,8 @@ func (s *ListenerService) Open() error {
|
||||
}
|
||||
s.Logger.Info("GDAX listener started")
|
||||
|
||||
s.trades = make(chan tacitus.Trade, 1024)
|
||||
s.trades = make(chan tacitus.Trade, buf)
|
||||
s.err = make(chan error)
|
||||
s.setRestart(true)
|
||||
|
||||
s.listen()
|
||||
|
||||
@ -46,15 +38,22 @@ func (s *ListenerService) Open() error {
|
||||
}
|
||||
|
||||
func (s *ListenerService) Close() error {
|
||||
s.closed = true
|
||||
s.setRestart(false)
|
||||
s.Client.Close()
|
||||
s.Logger.Info("GDAX listener stopped")
|
||||
|
||||
close(s.trades)
|
||||
close(s.err)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *ListenerService) Subscribe(products ...string) {
|
||||
if s.dirty {
|
||||
return
|
||||
}
|
||||
|
||||
s.dirty = true
|
||||
|
||||
req := Request{}
|
||||
req.Type = Subscribe
|
||||
|
||||
@ -62,8 +61,6 @@ func (s *ListenerService) Subscribe(products ...string) {
|
||||
req.Channels[0].Name = "matches"
|
||||
req.Channels[0].ProductIds = products
|
||||
|
||||
s.subs = products
|
||||
|
||||
s.Client.Send(req)
|
||||
}
|
||||
|
||||
@ -75,10 +72,6 @@ func (s *ListenerService) Error() <-chan error {
|
||||
return s.err
|
||||
}
|
||||
|
||||
func (s *ListenerService) SetLogger(logger tacitus.Logger) {
|
||||
s.Logger = logger
|
||||
}
|
||||
|
||||
func (s *ListenerService) listen() {
|
||||
go func() {
|
||||
for msg := range s.Client.Stream() {
|
||||
@ -106,50 +99,16 @@ func (s *ListenerService) listen() {
|
||||
for err := range s.Client.Error() {
|
||||
s.Logger.Info("Error from GDAX client: %v", err)
|
||||
|
||||
if s.SendErrors() {
|
||||
s.err <- err
|
||||
select {
|
||||
case s.err <- err:
|
||||
default:
|
||||
}
|
||||
|
||||
close(s.trades)
|
||||
close(s.err)
|
||||
}
|
||||
|
||||
s.Logger.Info("Error closed")
|
||||
|
||||
if s.restart() {
|
||||
err := s.Open()
|
||||
if err != nil {
|
||||
s.Logger.Info("GDAX Unable to reastablish connection")
|
||||
s.Close()
|
||||
return
|
||||
}
|
||||
s.Subscribe(s.subs...)
|
||||
}
|
||||
}()
|
||||
|
||||
}
|
||||
|
||||
func (s *ListenerService) restart() bool {
|
||||
s.restMu.Lock()
|
||||
defer s.restMu.Unlock()
|
||||
|
||||
return s.shouldRestart
|
||||
}
|
||||
|
||||
func (s *ListenerService) setRestart(restart bool) {
|
||||
s.restMu.Lock()
|
||||
defer s.restMu.Unlock()
|
||||
|
||||
s.shouldRestart = restart
|
||||
}
|
||||
|
||||
func (s *ListenerService) SendErrors() bool {
|
||||
s.errorsMu.Lock()
|
||||
defer s.errorsMu.Unlock()
|
||||
|
||||
return s.sendErrors
|
||||
}
|
||||
|
||||
func (s *ListenerService) SetSendErrors(send bool) {
|
||||
s.errorsMu.Lock()
|
||||
defer s.errorsMu.Unlock()
|
||||
|
||||
s.sendErrors = send
|
||||
}
|
||||
|
@ -1,7 +1,7 @@
|
||||
package gdax
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
@ -2,7 +2,7 @@ package websocket
|
||||
|
||||
import (
|
||||
"github.com/gorilla/websocket"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/gdax"
|
||||
"github.com/kcotugno/tacitus/gdax"
|
||||
|
||||
"sync"
|
||||
"time"
|
||||
|
@ -1,7 +1,7 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
|
||||
"net/http"
|
||||
)
|
||||
|
@ -4,7 +4,7 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
)
|
||||
|
||||
const DefaultAddr = ":8080"
|
||||
|
@ -6,7 +6,7 @@ import (
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
)
|
||||
|
||||
type TradeHandler struct {
|
||||
|
@ -6,7 +6,4 @@ type ListenerService interface {
|
||||
Subscribe(products ...string)
|
||||
Stream() <-chan Trade
|
||||
Error() <-chan error
|
||||
SendErrors() bool
|
||||
SetSendErrors(send bool)
|
||||
SetLogger(logger Logger)
|
||||
}
|
||||
|
@ -1,7 +1,7 @@
|
||||
package ops
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
|
||||
"sort"
|
||||
"time"
|
||||
|
@ -1,7 +1,7 @@
|
||||
package ops
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
"github.com/shopspring/decimal"
|
||||
|
||||
"sync"
|
||||
|
@ -1,14 +1,15 @@
|
||||
package ops
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
)
|
||||
|
||||
type Registrar struct {
|
||||
Database tacitus.DatabaseClientService
|
||||
Listener tacitus.ListenerService
|
||||
Logger tacitus.Logger
|
||||
|
||||
logger tacitus.Logger
|
||||
err chan error
|
||||
}
|
||||
|
||||
func (r *Registrar) Record(products ...string) error {
|
||||
@ -22,26 +23,38 @@ func (r *Registrar) Record(products ...string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Registrar) SetLogger(logger tacitus.Logger) {
|
||||
r.Listener.SetLogger(logger)
|
||||
r.logger = logger
|
||||
}
|
||||
|
||||
func (r *Registrar) record() {
|
||||
go func() {
|
||||
for t := range r.Listener.Stream() {
|
||||
go func(trade tacitus.Trade) {
|
||||
_, err := r.Database.TradeService().CreateTrade(trade)
|
||||
if err != nil {
|
||||
r.logger.Info("Error inserting trade: %v", err)
|
||||
r.Logger.Info("Error inserting trade: %v", err)
|
||||
|
||||
select {
|
||||
case r.err <- err:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}(t)
|
||||
}
|
||||
r.Logger.Info("Trade goroutine done")
|
||||
}()
|
||||
|
||||
go func() {
|
||||
for err := range r.Listener.Error() {
|
||||
r.logger.Info("Registrar received error: %v", err)
|
||||
r.Logger.Info("Registrar received error: %v", err)
|
||||
|
||||
select {
|
||||
case r.err <- err:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
r.Logger.Info("Error goroutine done")
|
||||
}()
|
||||
}
|
||||
|
||||
func (r *Registrar) Error() <-chan error {
|
||||
return r.err
|
||||
}
|
||||
|
@ -1,8 +1,8 @@
|
||||
package ops
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"git.kevincotugno.com/kcotugno/tacitus/gdax"
|
||||
"github.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus/gdax"
|
||||
|
||||
"sort"
|
||||
"time"
|
||||
|
@ -1,7 +1,7 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
"github.com/shopspring/decimal"
|
||||
|
||||
"database/sql"
|
||||
|
@ -1,7 +1,7 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
|
||||
"strings"
|
||||
"time"
|
||||
|
@ -7,19 +7,15 @@ import (
|
||||
"database/sql"
|
||||
"text/template"
|
||||
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
)
|
||||
|
||||
const connStr = `host={{.Host}} dbname={{.Name}} port={{.Port}} user={{.User}} ` +
|
||||
`password={{.Password}} sslmode={{.SslMode}}`
|
||||
const connStr = `user={{.User}} dbname={{.Name}} sslmode={{.SslMode}}`
|
||||
|
||||
type Client struct {
|
||||
Host string
|
||||
Port int
|
||||
Name string
|
||||
User string
|
||||
SslMode string
|
||||
Password string
|
||||
Name string
|
||||
User string
|
||||
SslMode string
|
||||
|
||||
tradeService TradeService
|
||||
confirmationService ConfirmationService
|
||||
@ -34,12 +30,6 @@ func NewClient() *Client {
|
||||
c.tradeService.client = &c
|
||||
c.confirmationService.client = &c
|
||||
c.aggregationService.client = &c
|
||||
|
||||
c.Host = "localhost"
|
||||
c.Port = 5432
|
||||
c.Name = "postgres"
|
||||
c.User = "postgres"
|
||||
c.Password = `""`
|
||||
c.SslMode = "disable"
|
||||
|
||||
return &c
|
||||
|
@ -1,7 +1,7 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
)
|
||||
|
||||
const (
|
||||
|
@ -1,7 +1,7 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
|
||||
"database/sql"
|
||||
"errors"
|
||||
|
@ -1,7 +1,7 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"git.kevincotugno.com/kcotugno/tacitus"
|
||||
"github.com/kcotugno/tacitus"
|
||||
|
||||
"database/sql"
|
||||
"strings"
|
||||
|
37
product.go
37
product.go
@ -1,37 +0,0 @@
|
||||
package tacitus
|
||||
|
||||
import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
type Product string
|
||||
|
||||
const (
|
||||
BtcUsd = "BTC-USD"
|
||||
BtcEur = "BTC-EUR"
|
||||
BtcGbp = "BTC-GBP"
|
||||
|
||||
BchUsd = "BCH-USD"
|
||||
|
||||
EthUsd = "ETH-USD"
|
||||
EthBtc = "ETH-BTC"
|
||||
EthEur = "ETH-EUR"
|
||||
|
||||
LtcUsd = "LTC-USD"
|
||||
LtcBtc = "LTC-BTC"
|
||||
LtcEur = "LTC-EUR"
|
||||
)
|
||||
|
||||
var products = [...]string{BtcUsd, BtcEur, BtcGbp, BchUsd, EthUsd, EthBtc,
|
||||
EthEur, LtcUsd, LtcBtc, LtcEur}
|
||||
|
||||
func ValidProduct(prod string) bool {
|
||||
prod = strings.ToUpper(prod)
|
||||
for _, p := range products {
|
||||
if p == prod {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
14
trade.go
14
trade.go
@ -3,9 +3,12 @@ package tacitus
|
||||
import (
|
||||
"github.com/shopspring/decimal"
|
||||
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
var PRODUCTS = [...]string{"BTC-USD", "ETH-USD"}
|
||||
|
||||
type Trade struct {
|
||||
Id int `json:"-"`
|
||||
TradeId int `json:"trade_id,"`
|
||||
@ -32,3 +35,14 @@ type TradeService interface {
|
||||
TradesAfterAll(product string, id int) (Results, error)
|
||||
TradesWhereResults(sql string, params ...interface{}) (Results, error)
|
||||
}
|
||||
|
||||
func ValidProduct(prod string) bool {
|
||||
prod = strings.ToUpper(prod)
|
||||
for _, p := range PRODUCTS {
|
||||
if p == prod {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
Reference in New Issue
Block a user