golang protobuf livecoin

gistfile1.go

package main

import (
	"awesomeProject/protobuf_ws"
	"crypto/hmac"
	"crypto/sha256"
	"errors"
	"fmt"
	"github.com/golang/protobuf/proto"
	"github.com/gorilla/websocket"
	"log"
	"net/url"
	"os"
	"os/signal"
	"strconv"
	"time"
)

const (
	API_KEY    = ""
	SECRET_KEY = ""

	BTC_USD  = "BTC/USD"
	EUR_USD  = "EUR/USD"
	USD_RUR  = "USD/RUR"
	BTC_RUR  = "BTC/RUR"
	BTC_EUR  = "BTC/EUR"
	ZEC_BTC  = "ZEC/BTC"
	ZEC_ETH  = "ZEC/ETH"
	ZEC_USD  = "ZEC/USD"
	ETH_RUR  = "ETH/RUR"
	ETH_USD  = "ETH/USD"
	ETH_BTC  = "ETH/BTC"
	LTC_BTC  = "LTC/BTC"
	LTC_USD  = "LTC/USD"
	BCH_RUR  = "BCH/RUR"
	BCH_USD  = "BCH/USD"
	BCH_BTC  = "BCH/BTC"
	BCH_ETH  = "BCH/ETH"
	TOR_BTC  = "TOR/BTC"
	DIG_USD  = "DIG/USD"
	DIG_BTC  = "DIG/BTC"
	DIG_LTC  = "DIG/LTC"
	DIG_ETH  = "DIG/ETH"
	DASH_USD = "DASH/USD"
	DASH_BTC = "DASH/BTC"
)

type Order struct {
	Price    float64
	Quantity float64
	Id       int64
	Type     string
	Pair     string
}

type Orders map[int64]*Order
type Types map[string]Orders
type Pairs map[string]Types

var pairs Pairs

func main() {
	currencyPairs := []string{BTC_USD, EUR_USD, BTC_RUR, BTC_EUR, ZEC_BTC, ETH_BTC, BCH_BTC, TOR_BTC, DIG_BTC, DASH_BTC,USD_RUR, ZEC_ETH, ZEC_USD, ETH_RUR, ETH_USD, LTC_BTC, LTC_USD, BCH_RUR, BCH_USD, BCH_ETH, DIG_USD, DIG_LTC, DIG_ETH, DASH_USD,}

	pairs = Pairs{}

	log.SetFlags(0)

	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt)

	u := url.URL{Scheme: "wss", Host: "ws.api.livecoin.net", Path: "/ws/beta2"}
	log.Printf("connecting to %s", u.String())

	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
	if err != nil {
		log.Fatal("dial:", err)
	}
	defer c.Close()

	done := make(chan struct{})

	go func() {
		defer close(done)
		for {
			_, message, err := c.ReadMessage()
			if err != nil {
				log.Fatalln(err)
				return
			}
			if len(message) == 0 {
				err = ApiPingRequest(c)
				if err != nil {
					log.Fatalln(err)
					return
				}
				continue
			}

			response := &protobuf_ws.WsResponse{}
			err = proto.Unmarshal(message, response)
			if err != nil {
				log.Fatalln(err)
				return
			}
			switch response.GetMeta().GetResponseType() {
			case protobuf_ws.WsResponseMetaData_ERROR:
				err = ApiErrorResponse(response.GetMsg())
			case protobuf_ws.WsResponseMetaData_PONG_RESPONSE:
				err = ApiPongResponse(response.GetMsg())
			case protobuf_ws.WsResponseMetaData_BALANCES_RESPONSE:
				err = ApiBalancesResponse(response.GetMsg())
			case protobuf_ws.WsResponseMetaData_ORDER_BOOK_RAW_CHANNEL_SUBSCRIBED:
				err = ApiOrderBookRawChannelSubscribedResponse(response.GetMsg())
			case protobuf_ws.WsResponseMetaData_ORDER_BOOK_RAW_NOTIFY:
				err = ApiOrderBookRawNotifyResponse(response.GetMsg())
			case protobuf_ws.WsResponseMetaData_LOGIN_RESPONSE:
				{
					err = ApiLoginResponse(response.GetMsg())
					go func() {
						err = ApiBalancesRequest(c, true, 10000)
						if err != nil {
							log.Fatalln(err)
						}
					}()
				}
			default:
				err = errors.New(protobuf_ws.WsResponseMetaData_WsResponseMsgType_name[int32(response.GetMeta().GetResponseType())])
			}
			if err != nil {
				log.Fatalln(err)
				return
			}
		}
	}()

	go func() {
		err := ApiPingRequest(c)
		if err != nil {
			log.Fatalln(err)
		}
		err = ApiLoginRequest(c, API_KEY, 10000)
		if err != nil {
			log.Fatalln(err)
		}

		for _, currencyPair := range currencyPairs {
			err = ApiSubscribeOrderBookRawChannelRequest(c, currencyPair)
			if err != nil {
				log.Fatalln(err)
			}
		}
	}()

	for {
		select {
		case <-done:
			return
		case <-interrupt:
			log.Println("interrupt")
			err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
			if err != nil {
				log.Println("write close:", err)
				return
			}
			select {
			case <-done:
			case <-time.After(time.Second):
			}
			return
		}
	}
}

func ApiOrderBookRawNotifyResponse(msg []byte) error {
	response := &protobuf_ws.OrderBookRawNotification{}
	err := proto.Unmarshal(msg, response)
	if err != nil {
		return err
	}
	return processOrderBookRows(response.GetCurrencyPair(), response.GetData())
}

func processOrderBookRows(currencyPair string, data []*protobuf_ws.OrderBookRawEvent) error {
	l := len(data)
	for i := 0; i < l; i++ {
		item := data[i]
		price, err := strconv.ParseFloat(item.GetPrice(), 64)
		if err != nil {
			return err
		}
		quantity, err := strconv.ParseFloat(item.GetQuantity(), 64)
		if err != nil {
			return err
		}
		orderType := protobuf_ws.OrderBookRawEvent_OrderType_name[int32(item.GetOrderType())]
		order := &Order{
			Id:       item.GetId(),
			Type:     orderType,
			Pair:     currencyPair,
			Price:    price,
			Quantity: quantity,
		}
		if _, ok := pairs[currencyPair]; !ok {
			pairs[currencyPair] = Types{}
		}
		if _, ok := pairs[currencyPair][order.Type]; !ok {
			pairs[currencyPair][order.Type] = Orders{}
		}
		if order.Quantity == 0 {
			delete(pairs[currencyPair][order.Type], order.Id)
		} else {
			pairs[currencyPair][order.Type][order.Id] = order
		}
	}
	fmt.Println(">>>>")
	for currencyPair, currencyPairData := range pairs {
		for orderType, orderTypeData := range currencyPairData {
			minPrice, maxPrice := -1.0, -1.0
			for _, orderData := range orderTypeData {
				if minPrice == -1.0 || minPrice > orderData.Price {
					minPrice = orderData.Price
				}
				if maxPrice == -1.0 || maxPrice < orderData.Price {
					maxPrice = orderData.Price
				}
			}
			if orderType == "BID" {
				fmt.Printf("%10s %3s %5d %30.15f\n", currencyPair, orderType, len(orderTypeData), maxPrice)
			} else {
				fmt.Printf("%10s %3s %5d %30.15f\n", currencyPair, orderType, len(orderTypeData), minPrice)
			}
		}
	}
	fmt.Println("<<<<")
	return nil
}

func ApiOrderBookRawChannelSubscribedResponse(msg []byte) error {
	response := &protobuf_ws.OrderBookRawChannelSubscribedResponse{}
	err := proto.Unmarshal(msg, response)
	if err != nil {
		return err
	}
	return processOrderBookRows(response.GetCurrencyPair(), response.GetData())
}

func ApiPongResponse(msg []byte) error {
	response := &protobuf_ws.PongResponse{}
	err := proto.Unmarshal(msg, response)
	if err != nil {
		return err
	}
	fmt.Println(response.String())
	return nil
}

func ApiBalancesResponse(msg []byte) error {
	response := &protobuf_ws.BalancesResponse{}
	err := proto.Unmarshal(msg, response)
	if err != nil {
		return err
	}
	balances:= response.GetBalances()
	for i:=0; i< len(balances) ; i++  {
		b := balances[i]
		fmt.Println(b.GetCurrency(), b.GetType(), b.GetValue())
	}
	return nil
}

func ApiErrorResponse(msg []byte) error {
	response := &protobuf_ws.ErrorResponse{}
	err := proto.Unmarshal(msg, response)
	if err != nil {
		return err
	}
	fmt.Println(response.String())
	return nil
}

func ApiSendRequest(c *websocket.Conn, requestType protobuf_ws.WsRequestMetaData_WsRequestMsgType, requestBody proto.Message) error {
	meta := protobuf_ws.WsRequestMetaData{RequestType: &requestType}
	requestMessage, err := proto.Marshal(requestBody)
	if err != nil {
		return err
	}
	request := &protobuf_ws.WsRequest{Meta: &meta, Msg: requestMessage}
	data, err := proto.Marshal(request)
	if err != nil {
		return err
	}
	return c.WriteMessage(websocket.BinaryMessage, data)
}

func ApiPingRequest(c *websocket.Conn) error {
	request := &protobuf_ws.PingRequest{}

	return ApiSendRequest(c, protobuf_ws.WsRequestMetaData_PING_REQUEST, request)
}

func ApiSubscribeOrderBookRawChannelRequest(c *websocket.Conn, currencyPair string) error {
	request := &protobuf_ws.SubscribeOrderBookRawChannelRequest{
		CurrencyPair: proto.String(currencyPair),
	}

	return ApiSendRequest(c, protobuf_ws.WsRequestMetaData_SUBSCRIBE_ORDER_BOOK_RAW, request)
}

func ApiBalancesRequest(c *websocket.Conn, onlyNotZero bool, ttl int32) error {
	request := &protobuf_ws.BalancesRequest{
		ExpireControl: &protobuf_ws.RequestExpired{
			Now: proto.Int64(time.Now().Unix() * 1000),
			Ttl: proto.Int32(ttl),
		},
		OnlyNotZero: proto.Bool(onlyNotZero),
	}

	return ApiSendSignedRequest(c, protobuf_ws.WsRequestMetaData_BALANCES, request)
}

func ApiLoginRequest(c *websocket.Conn, apiKey string, ttl int32) error {
	request := &protobuf_ws.LoginRequest{
		ApiKey: proto.String(apiKey),
		ExpireControl: &protobuf_ws.RequestExpired{
			Now: proto.Int64(time.Now().Unix() * 1000),
			Ttl: proto.Int32(ttl),
		},
	}

	return ApiSendSignedRequest(c, protobuf_ws.WsRequestMetaData_LOGIN, request)
}

func ApiLoginResponse(msg []byte) error {
	response := &protobuf_ws.LoginResponse{}
	err := proto.Unmarshal(msg, response)
	if err != nil {
		return err
	}
	return nil
}

func ApiSendSignedRequest(c *websocket.Conn, requestType protobuf_ws.WsRequestMetaData_WsRequestMsgType, requestBody proto.Message) error {

	requestMessage, err := proto.Marshal(requestBody)
	if err != nil {
		return err
	}
	mac := hmac.New(sha256.New, []byte(SECRET_KEY))
	mac.Write(requestMessage)
	sign := mac.Sum(nil)

	meta := protobuf_ws.WsRequestMetaData{RequestType: &requestType, Sign: sign}
	request := &protobuf_ws.WsRequest{Meta: &meta, Msg: requestMessage}
	data, err := proto.Marshal(request)
	if err != nil {
		return err
	}
	return c.WriteMessage(websocket.BinaryMessage, data)
}