diff --git a/node/pkg/fetcher/fetcher.go b/node/pkg/fetcher/fetcher.go index 7315d1bdf..9850727e4 100644 --- a/node/pkg/fetcher/fetcher.go +++ b/node/pkg/fetcher/fetcher.go @@ -100,6 +100,8 @@ func (f *Fetcher) fetch(chainHelpers map[string]ChainHelper, proxies []Proxy) ([ errChan <- fetchErr return } + case *definition.Type == "websocket": + return default: errChan <- errorSentinel.ErrFetcherInvalidType } diff --git a/node/pkg/wfetcher/app.go b/node/pkg/wfetcher/app.go new file mode 100644 index 000000000..a64aed684 --- /dev/null +++ b/node/pkg/wfetcher/app.go @@ -0,0 +1 @@ +package wfetcher diff --git a/node/pkg/wfetcher/binance/binance.go b/node/pkg/wfetcher/binance/binance.go new file mode 100644 index 000000000..596465e2a --- /dev/null +++ b/node/pkg/wfetcher/binance/binance.go @@ -0,0 +1,73 @@ +package binance + +import ( + "context" + "fmt" + "strings" + "time" + + "bisonai.com/orakl/node/pkg/wfetcher" + "bisonai.com/orakl/node/pkg/wss" + "github.com/rs/zerolog/log" +) + +type BinanceFetcher wfetcher.Fetcher + +// expected to recieve feedmap with key having format "" +func New(ctx context.Context, opts ...wfetcher.FetcherOption) (*BinanceFetcher, error) { + config := &wfetcher.FetcherConfig{} + for _, opt := range opts { + opt(config) + } + + fetcher := &BinanceFetcher{} + if len(config.FeedMap) == 0 { + log.Error().Str("Player", "Binance").Msg("no feed map") + return nil, fmt.Errorf("no feed map") + } + fetcher.FeedMap = config.FeedMap + + streams := []Stream{} + for feed := range config.FeedMap { + streams = append(streams, Stream(strings.ToLower(feed)+"@miniTicker")) + } + subscription := Subscription{"SUBSCRIBE", streams, 1} + + ws, err := wss.NewWebsocketHelper(ctx, + wss.WithEndpoint(URL), + wss.WithSubscriptions([]any{subscription}), + wss.WithProxyUrl(config.Proxy)) + if err != nil { + log.Error().Str("Player", "Binance").Err(err).Msg("error in NewWebsocketHelper") + return nil, err + } + fetcher.Ws = ws + + return fetcher, nil +} + +func (b *BinanceFetcher) handleMessage(ctx context.Context, message map[string]any) error { + ticker, err := MessageToTicker(message) + if err != nil { + log.Error().Str("Player", "Binance").Err(err).Msg("error in MessageToTicker") + return err + } + + feedData, err := TickerToFeedData(ticker, b.FeedMap) + if err != nil { + log.Error().Str("Player", "Binance").Err(err).Msg("error in MiniTickerToFeedData") + return err + } + + err = wfetcher.StoreFeed(ctx, feedData, 2*time.Second) + if err != nil { + log.Error().Str("Player", "Binance").Err(err).Msg("error in StoreFeed") + return err + } + + return nil +} + +func (b *BinanceFetcher) Run(ctx context.Context) { + b.Ws.Run(ctx, b.handleMessage) +} diff --git a/node/pkg/wfetcher/binance/type.go b/node/pkg/wfetcher/binance/type.go new file mode 100644 index 000000000..5cbc08834 --- /dev/null +++ b/node/pkg/wfetcher/binance/type.go @@ -0,0 +1,25 @@ +package binance + +const ( + URL = "wss://stream.binance.com:443/ws" +) + +type Stream string + +type Subscription struct { + Method string `json:"method"` + Params []Stream `json:"params"` + Id uint32 `json:"id"` +} + +type MiniTicker struct { + EventType string `json:"e"` + EventTime int64 `json:"E"` + Symbol string `json:"s"` + Price string `json:"c"` + OpenPrice string `json:"o"` + HighPrice string `json:"h"` + LowPrice string `json:"l"` + Volume string `json:"v"` + QuoteVolume string `json:"q"` +} diff --git a/node/pkg/wfetcher/binance/utils.go b/node/pkg/wfetcher/binance/utils.go new file mode 100644 index 000000000..6979808f9 --- /dev/null +++ b/node/pkg/wfetcher/binance/utils.go @@ -0,0 +1,57 @@ +package binance + +import ( + "encoding/json" + "fmt" + "time" + + "bisonai.com/orakl/node/pkg/wfetcher" + "github.com/rs/zerolog/log" +) + +func GetSubscription(names []string) Subscription { + streams := []Stream{} + + for _, name := range names { + streams = append(streams, Stream(name+"@miniTicker")) + } + + return Subscription{ + Method: "SUBSCRIBE", + Params: streams, + Id: 1, + } +} +func MessageToTicker(msg map[string]any) (MiniTicker, error) { + var miniTicker MiniTicker + jsonData, err := json.Marshal(msg) + if err != nil { + log.Error().Str("Player", "Binance").Err(err).Msg("error in json.Marshal") + return miniTicker, err + } + + err = json.Unmarshal(jsonData, &miniTicker) + if err != nil { + log.Error().Str("Player", "Binance").Err(err).Msg("error in json.Unmarshal") + return miniTicker, err + } + return miniTicker, nil +} + +func TickerToFeedData(miniTicker MiniTicker, feedMap map[string]int32) (*wfetcher.FeedData, error) { + feedData := new(wfetcher.FeedData) + timestamp := time.Unix(miniTicker.EventTime/1000, 0) + value, err := wfetcher.PriceStringToFloat64(miniTicker.Price) + if err != nil { + return feedData, err + } + + id, exists := feedMap[miniTicker.Symbol] + if !exists { + return feedData, fmt.Errorf("feed not found") + } + feedData.FeedId = id + feedData.Value = value + feedData.Timestamp = timestamp + return feedData, nil +} diff --git a/node/pkg/wfetcher/coinone/coinone.go b/node/pkg/wfetcher/coinone/coinone.go new file mode 100644 index 000000000..90a033235 --- /dev/null +++ b/node/pkg/wfetcher/coinone/coinone.go @@ -0,0 +1,90 @@ +package coinone + +import ( + "context" + "fmt" + "strings" + "time" + + "bisonai.com/orakl/node/pkg/wfetcher" + "bisonai.com/orakl/node/pkg/wss" + "github.com/rs/zerolog/log" +) + +type CoinoneFetcher wfetcher.Fetcher + +// expected to recieve feedmap with key having format "-" +func New(ctx context.Context, opts ...wfetcher.FetcherOption) (*CoinoneFetcher, error) { + config := &wfetcher.FetcherConfig{} + for _, opt := range opts { + opt(config) + } + + fetcher := &CoinoneFetcher{} + + if len(config.FeedMap) == 0 { + log.Error().Str("Player", "Coinone").Msg("no feed map") + return nil, fmt.Errorf("no feed map") + } + fetcher.FeedMap = config.FeedMap + + subscriptions := []any{} + for feed := range config.FeedMap { + raw := strings.Split(feed, "-") + if len(raw) < 2 { + log.Error().Str("Player", "Coinone").Msg("invalid feed name") + return nil, fmt.Errorf("invalid feed name") + } + base := raw[0] + quote := raw[1] + + subscriptions = append(subscriptions, Subscription{ + RequestType: "SUBSCRIBE", + Channel: "TICKER", + Topic: Topic{ + QuoteCurrency: quote, + TargetCurrency: base, + }, + Format: "SHORT", + }) + } + ws, err := wss.NewWebsocketHelper(ctx, + wss.WithEndpoint(URL), + wss.WithSubscriptions(subscriptions), + wss.WithProxyUrl(config.Proxy)) + if err != nil { + log.Error().Str("Player", "Coinone").Err(err).Msg("error in NewWebsocketHelper") + return nil, err + } + fetcher.Ws = ws + + return fetcher, nil +} + +func (c *CoinoneFetcher) handleMessage(ctx context.Context, message map[string]any) error { + raw, err := MessageToRawResponse(message) + if err != nil { + log.Error().Str("Player", "Coinone").Err(err).Msg("error in MessageToRawResponse") + return err + } + + if raw.ResponseType != "DATA" { + return nil + } + feedData, err := DataToFeedData(raw.Data, c.FeedMap) + if err != nil { + log.Error().Str("Player", "Coinone").Err(err).Msg("error in DataToFeedData") + return err + } + + err = wfetcher.StoreFeed(ctx, &feedData, 2*time.Second) + if err != nil { + log.Error().Str("Player", "Coinone").Err(err).Msg("error in StoreFeed") + return err + } + return nil +} + +func (c *CoinoneFetcher) Run(ctx context.Context) { + c.Ws.Run(ctx, c.handleMessage) +} diff --git a/node/pkg/wfetcher/coinone/type.go b/node/pkg/wfetcher/coinone/type.go new file mode 100644 index 000000000..dec132f9f --- /dev/null +++ b/node/pkg/wfetcher/coinone/type.go @@ -0,0 +1,47 @@ +package coinone + +const ( + URL = "wss://stream.coinone.co.kr" +) + +type Topic struct { + QuoteCurrency string `json:"quote_currency"` + TargetCurrency string `json:"target_currency"` +} + +type Subscription struct { + RequestType string `json:"request_type"` + Channel string `json:"channel"` + Topic Topic `json:"topic"` + Format string `json:"format"` +} + +type Data struct { + QuoteCurrency string `json:"qc"` + TargetCurrency string `json:"tc"` + Timestamp int64 `json:"t"` + QuoteVolume string `json:"qv"` + TargetVolume string `json:"tv"` + First string `json:"fi"` + Low string `json:"lo"` + High string `json:"hi"` + Last string `json:"la"` + VolumePower string `json:"vp"` + AskBestPrice string `json:"abp"` + AskBestQty string `json:"abq"` + BidBestPrice string `json:"bbp"` + BidBestQty string `json:"bbq"` + ID string `json:"i"` + YesterdayFirst string `json:"yfi"` + YesterdayLow string `json:"ylo"` + YesterdayHigh string `json:"yhi"` + YesterdayLast string `json:"yla"` + YesterdayQuoteVolume string `json:"yqv"` + YesterdayTargetVolume string `json:"ytv"` +} + +type Raw struct { + ResponseType string `json:"r"` + Channel string `json:"c"` + Data Data `json:"d"` +} diff --git a/node/pkg/wfetcher/coinone/utils.go b/node/pkg/wfetcher/coinone/utils.go new file mode 100644 index 000000000..91bb4a69b --- /dev/null +++ b/node/pkg/wfetcher/coinone/utils.go @@ -0,0 +1,47 @@ +package coinone + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "bisonai.com/orakl/node/pkg/wfetcher" + "github.com/rs/zerolog/log" +) + +func MessageToRawResponse(msg map[string]any) (Raw, error) { + var rawResponse Raw + jsonData, err := json.Marshal(msg) + if err != nil { + log.Error().Str("Player", "Coinone").Err(err).Msg("error in json.Marshal") + return rawResponse, err + } + + err = json.Unmarshal(jsonData, &rawResponse) + if err != nil { + log.Error().Str("Player", "Coinone").Err(err).Msg("error in json.Unmarshal") + return rawResponse, err + } + return rawResponse, nil +} + +func DataToFeedData(data Data, feedMap map[string]int32) (wfetcher.FeedData, error) { + feedData := wfetcher.FeedData{} + + timestamp := time.Unix(data.Timestamp/1000, 0) + value, err := wfetcher.PriceStringToFloat64(data.Last) + + if err != nil { + return feedData, err + } + + id, exists := feedMap[strings.ToUpper(data.TargetCurrency)+"-"+strings.ToUpper(data.QuoteCurrency)] + if !exists { + return feedData, fmt.Errorf("feed not found") + } + feedData.FeedId = id + feedData.Value = value + feedData.Timestamp = timestamp + return feedData, nil +} diff --git a/node/pkg/wfetcher/korbit/korbit.go b/node/pkg/wfetcher/korbit/korbit.go new file mode 100644 index 000000000..2eaae52a3 --- /dev/null +++ b/node/pkg/wfetcher/korbit/korbit.go @@ -0,0 +1,93 @@ +package korbit + +import ( + "context" + "fmt" + "strings" + "time" + + "bisonai.com/orakl/node/pkg/wfetcher" + "bisonai.com/orakl/node/pkg/wss" + "github.com/rs/zerolog/log" +) + +type KorbitFetcher wfetcher.Fetcher + +// expected to recieve feedmap with key having format "-" +func New(ctx context.Context, opts ...wfetcher.FetcherOption) (*KorbitFetcher, error) { + config := &wfetcher.FetcherConfig{} + for _, opt := range opts { + opt(config) + } + + fetcher := &KorbitFetcher{} + + if len(config.FeedMap) == 0 { + log.Error().Str("Player", "Korbit").Msg("no feed map") + return nil, fmt.Errorf("no feed map") + } + fetcher.FeedMap = config.FeedMap + + pairListString := []string{} + for feed := range config.FeedMap { + raw := strings.Split(feed, "-") + if len(raw) < 2 { + log.Error().Str("Player", "Coinone").Msg("invalid feed name") + return nil, fmt.Errorf("invalid feed name") + } + base := raw[0] + quote := raw[1] + pairListString = append(pairListString, fmt.Sprintf("%s_%s", strings.ToLower(base), strings.ToLower(quote))) + } + + subscription := Subscription{ + AccessToken: nil, + Timestamp: time.Now().Unix(), + Event: "korbit:subscribe", + Data: Data{Channels: []string{ + "ticker:" + strings.Join(pairListString, ","), + }}, + } + + ws, err := wss.NewWebsocketHelper(ctx, + wss.WithEndpoint(URL), + wss.WithSubscriptions([]any{subscription}), + wss.WithProxyUrl(config.Proxy)) + if err != nil { + log.Error().Str("Player", "Korbit").Err(err).Msg("error in NewWebsocketHelper") + return nil, err + } + fetcher.Ws = ws + return fetcher, nil + +} + +func (k *KorbitFetcher) handleMessage(ctx context.Context, message map[string]any) error { + raw, err := MessageToRawResponse(message) + if err != nil { + log.Error().Str("Player", "Korbit").Err(err).Msg("error in MessageToRawResponse") + return err + } + + if raw.Event != "korbit:push-ticker" { + return nil + } + + feedData, err := DataToFeedData(raw.Data, k.FeedMap) + if err != nil { + log.Error().Str("Player", "Korbit").Err(err).Msg("error in DataToFeedData") + return err + } + + err = wfetcher.StoreFeed(ctx, &feedData, 2*time.Second) + if err != nil { + log.Error().Str("Player", "Korbit").Err(err).Msg("error in StoreFeed") + return err + } + + return nil +} + +func (k *KorbitFetcher) Run(ctx context.Context) { + k.Ws.Run(ctx, k.handleMessage) +} diff --git a/node/pkg/wfetcher/korbit/type.go b/node/pkg/wfetcher/korbit/type.go new file mode 100644 index 000000000..70c3b6171 --- /dev/null +++ b/node/pkg/wfetcher/korbit/type.go @@ -0,0 +1,57 @@ +package korbit + +const ( + URL = "wss://ws2.korbit.co.kr/v1/user/push" +) + +/* +{ + "accessToken": null, + "event": "korbit:push-ticker", + "timestamp" : 1389678052000, + "data": + { + "channel": "ticker", + "currency_pair": "btc_krw", + "timestamp": 1558590089274, + "last": "9198500.1235789" + "open": "9500000.3445783", + "bid": "9192500.4578344", + "ask": "9198000.32148556", + "low": "9171500.23785685", + "high": "9599000.34876458", + "volume": "1539.18571988", + "change": "-301500.234578934" + } +} +*/ + +type Raw struct { + Event string `json:"event"` + Data Ticker `json:"data"` +} + +type Ticker struct { + Channel string `json:"channel"` + CurrencyPair string `json:"currency_pair"` + Timestamp int64 `json:"timestamp"` + Last string `json:"last"` + Open string `json:"open"` + Bid string `json:"bid"` + Ask string `json:"ask"` + Low string `json:"low"` + High string `json:"high"` + Volume string `json:"volume"` + Change string `json:"change"` +} + +type Data struct { + Channels []string `json:"channels"` +} + +type Subscription struct { + AccessToken *string `json:"accessToken"` + Timestamp int64 `json:"timestamp"` + Event string `json:"event"` + Data Data `json:"data"` +} diff --git a/node/pkg/wfetcher/korbit/utils.go b/node/pkg/wfetcher/korbit/utils.go new file mode 100644 index 000000000..7a2638d0d --- /dev/null +++ b/node/pkg/wfetcher/korbit/utils.go @@ -0,0 +1,52 @@ +package korbit + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "bisonai.com/orakl/node/pkg/wfetcher" + "github.com/rs/zerolog/log" +) + +func MessageToRawResponse(msg map[string]any) (Raw, error) { + var rawResponse Raw + jsonData, err := json.Marshal(msg) + if err != nil { + log.Error().Str("Player", "Korbit").Err(err).Msg("error in json.Marshal") + return rawResponse, err + } + + err = json.Unmarshal(jsonData, &rawResponse) + if err != nil { + log.Error().Str("Player", "Korbit").Err(err).Msg("error in json.Unmarshal") + return rawResponse, err + } + return rawResponse, nil +} + +func DataToFeedData(data Ticker, feedMap map[string]int32) (wfetcher.FeedData, error) { + feedData := wfetcher.FeedData{} + + timestamp := time.Unix(data.Timestamp/1000, 0) + value, err := wfetcher.PriceStringToFloat64(data.Last) + if err != nil { + return feedData, err + } + rawPair := strings.Split(data.CurrencyPair, "_") + if len(rawPair) < 2 { + return feedData, fmt.Errorf("invalid feed name") + } + target := rawPair[0] + quote := rawPair[1] + + id, exists := feedMap[strings.ToUpper(target)+"-"+strings.ToUpper(quote)] + if !exists { + return feedData, fmt.Errorf("feed not found") + } + feedData.FeedId = id + feedData.Value = value + feedData.Timestamp = timestamp + return feedData, nil +} diff --git a/node/pkg/wfetcher/type.go b/node/pkg/wfetcher/type.go new file mode 100644 index 000000000..8b525d547 --- /dev/null +++ b/node/pkg/wfetcher/type.go @@ -0,0 +1,58 @@ +package wfetcher + +import ( + "encoding/json" + "fmt" + "time" + + "bisonai.com/orakl/node/pkg/wss" +) + +const DECIMALS = 8 + +type Feed struct { + ID int32 `db:"id"` + Name string `db:"name"` + Definition json.RawMessage `db:"definition"` + ConfigID int32 `db:"config_id"` +} + +type Definition struct { + Type *string `json:"type"` + Provider *string `json:"provider"` +} + +type FeedData struct { + FeedId int32 `db:"feed_id"` + Value float64 `db:"value"` + Timestamp time.Time `db:"timestamp"` +} + +func (f *FeedData) String() string { + return fmt.Sprintf("FeedId: %d, Value: %f, Timestamp: %s", f.FeedId, f.Value, f.Timestamp) +} + +type FetcherConfig struct { + FeedMap map[string]int32 + Proxy string +} + +type FetcherOption func(*FetcherConfig) + +func WithFeedMap(feedMap map[string]int32) FetcherOption { + return func(c *FetcherConfig) { + c.FeedMap = feedMap + } +} + +func WithProxy(proxy string) FetcherOption { + return func(c *FetcherConfig) { + c.Proxy = proxy + } +} + +type Fetcher struct { + Name string + FeedMap map[string]int32 + Ws *wss.WebsocketHelper +} diff --git a/node/pkg/wfetcher/utils.go b/node/pkg/wfetcher/utils.go new file mode 100644 index 000000000..924c051f1 --- /dev/null +++ b/node/pkg/wfetcher/utils.go @@ -0,0 +1,78 @@ +package wfetcher + +import ( + "context" + "encoding/json" + "math" + "strconv" + "strings" + "time" + + "bisonai.com/orakl/node/pkg/db" + "github.com/rs/zerolog/log" +) + +// Returns two mappings of currency pair to feed id. +// combined : map['btcusdt'] = 1 +// separated : map['btc-usdt'] = 1 +func MakeFeedMap(feeds []Feed, provider string) (map[string]int32, map[string]int32) { + feedMapCombined := make(map[string]int32) + feedMapSeparated := make(map[string]int32) + + for _, feed := range feeds { + definition := new(Definition) + err := json.Unmarshal(feed.Definition, &definition) + if err != nil { + continue + } + if definition.Type != nil && + definition.Provider != nil && + *definition.Type == "websocket" && + *definition.Provider == provider { + // expected feed name to have format "-wss--" + + raw := strings.Split(feed.Name, "-") + if len(raw) < 3 { + log.Warn().Str("Player", provider).Str("name", feed.Name).Msg("invalid name") + continue + } + base := strings.ToUpper(raw[len(raw)-2]) + quote := strings.ToUpper(raw[len(raw)-1]) + combined := base + quote + + feedMapCombined[combined] = feed.ID + feedMapSeparated[base+"-"+quote] = feed.ID + } + } + + return feedMapCombined, feedMapSeparated +} + +func StoreFeed(ctx context.Context, feedData *FeedData, expiration time.Duration) error { + err := db.SetObject(ctx, "latestFeedData:"+strconv.Itoa(int(feedData.FeedId)), feedData, expiration) + if err != nil { + return err + } + return db.LPushObject(ctx, "feedDataBuffer", []FeedData{*feedData}) +} + +func StoreFeeds(ctx context.Context, feedData []FeedData, expiration time.Duration) error { + latestData := make(map[string]any) + for _, data := range feedData { + latestData["latestFeedData:"+strconv.Itoa(int(data.FeedId))] = data + } + err := db.MSetObjectWithExp(ctx, latestData, expiration) + if err != nil { + return err + } + return db.LPushObject(ctx, "feedDataBuffer", feedData) +} + +func PriceStringToFloat64(price string) (float64, error) { + f, err := strconv.ParseFloat(price, 64) + if err != nil { + return 0, err + } + + return f * float64(math.Pow10(int(DECIMALS))), nil +} diff --git a/node/pkg/wss/utils.go b/node/pkg/wss/utils.go index d0b72770a..46ad04253 100644 --- a/node/pkg/wss/utils.go +++ b/node/pkg/wss/utils.go @@ -38,6 +38,9 @@ func WithEndpoint(endpoint string) ConnectionOption { func WithProxyUrl(proxyUrl string) ConnectionOption { return func(c *ConnectionConfig) { + if proxyUrl == "" { + return + } c.Proxy = proxyUrl } } diff --git a/node/script/test_websocket/main.go b/node/script/test_websocket/main.go new file mode 100644 index 000000000..11ed49c46 --- /dev/null +++ b/node/script/test_websocket/main.go @@ -0,0 +1,25 @@ +package main + +import ( + "context" + "sync" + + "bisonai.com/orakl/node/pkg/wfetcher" + "bisonai.com/orakl/node/pkg/wfetcher/korbit" +) + +func main() { + ctx := context.Background() + var wg sync.WaitGroup + feedMap := make(map[string]int32) + feedMap["BTC-KRW"] = 1 + feedMap["ETH-KRW"] = 2 + + wg.Add(1) + c, err := korbit.New(ctx, wfetcher.WithFeedMap(feedMap)) + if err != nil { + panic(err) + } + c.Run(ctx) + wg.Wait() +}