Merge pull request #15 from itsphat/wip/futures

Futures market
heyphat · Jan 23, 2022 · a551f34 · a551f34
2 parents 1509495 + 8197382
commit a551f34
Show file tree

Hide file tree

Showing 13 changed files with 328 additions and 72 deletions.
diff --git a/cmd/app/watcher.go b/cmd/app/watcher.go
@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"net/http"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/gorilla/mux"
@@ -31,12 +30,16 @@ func watchlist(w http.ResponseWriter, req *http.Request) {
 }
 
 func watch(w http.ResponseWriter, req *http.Request) {
-	str, ok := mux.Vars(req)["ticker"]
+	tickers, ok := parseVars(mux.Vars(req), "ticker")
 	if !ok {
 		w.WriteHeader(http.StatusBadRequest)
 	}
-	for _, t := range strings.Split(str, ",") {
-		if err := market.Watch(t); err != nil {
+	mk, ok := parseOptions(req.URL.Query(), "market")
+	if !ok {
+		mk = []string{"CASH"}
+	}
+	for _, t := range tickers {
+		if err := market.Watch(t, mk[0]); err != nil {
 			logger.Error.Println(err)
 			InternalError(w)
 			return

diff --git a/configs/configs.json b/configs/configs.json
@@ -43,7 +43,7 @@
     "watcher": {
       "base_market": "USDT",
       "watchlist": [
-       "(?=(?<!(SUSD|BUSD|BVND|PAX|DAI|TUSD|USDC|VAI|BRL|AUD|BIRD|EUR|GBP|BIDR|DOWN|UP|BEAR|BULL))USDT)(?=USDT$)"
+       "(?=(?<!(USDP|SUSD|BUSD|BVND|PAX|DAI|TUSD|USDC|VAI|BRL|AUD|BIRD|EUR|GBP|BIDR|DOWN|UP|BEAR|BULL))USDT)(?=USDT$)"
       ],
       "runner": {
         "frames": [

diff --git a/internal/cmd/market/evaluator.go b/internal/cmd/market/evaluator.go
@@ -171,7 +171,7 @@ func (e *evaluator) processingWatcherRequest(msg *message) {
 	signals := e.getByTicker(r.GetName())
 	for _, s := range signals {
 		if s.Evaluate(r, nil) {
-			e.communicator.evaluator2Notifier <- e.communicator.newMessageWithPayloadID(r.GetName()+"-"+s.Name, s, nil)
+			e.communicator.evaluator2Notifier <- e.communicator.newMessageWithPayloadID(r.GetUniqueName()+"-"+s.Name, s, nil)
 			if s.IsOnetime() {
 				_ = e.drop(s.Name)
 			}

diff --git a/internal/cmd/market/market.go b/internal/cmd/market/market.go
@@ -2,6 +2,7 @@ package market
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"strings"
@@ -112,8 +113,11 @@ func NewMarket(configFilePath *string) (*MarketStruct, error) {
 	return Market, nil
 }
 
-func (m *MarketStruct) parseRunnerConfigs() *runner.RunnerConfigs {
+func (m *MarketStruct) parseRunnerConfigs(market runner.MarketType) *runner.RunnerConfigs {
 	out := runner.NewRunnerDefaultConfigs()
+	if market == runner.Cash || market == runner.Futures {
+		out.Market = runner.MarketType(market)
+	}
 	frames := []time.Duration{}
 	if len(m.configs.Market.Watcher.Runner.Frames) > 0 {
 		for _, f := range m.configs.Market.Watcher.Runner.Frames {
@@ -142,6 +146,10 @@ func (m *MarketStruct) initWatchlist() error {
 	if err != nil {
 		return err
 	}
+	futuStats, err := m.watcher.provider.binFutu.NewListPriceChangeStatsService().Do(context.Background())
+	if err != nil {
+		return err
+	}
 	limit := 1
 	if m.configs.IsProduction() {
 		limit = 5000
@@ -165,8 +173,26 @@ func (m *MarketStruct) initWatchlist() error {
 				if val, ok := listings[s.Symbol]; ok {
 					fd = &val
 				}
-				if err := m.watcher.watch(s.Symbol, m.parseRunnerConfigs(), fd); err != nil {
-					m.watcher.logger.Error.Println(m.watcher.newLog(s.Symbol, err.Error()))
+				if err := m.watcher.watch(s.Symbol, m.parseRunnerConfigs(runner.Cash), fd); err != nil {
+					m.watcher.logger.Error.Println(m.watcher.newLog(s.Symbol+"-"+string(runner.Cash), err.Error()))
+				}
+			}
+		}
+		for _, s := range futuStats {
+			if len(strings.Split(s.Symbol, "_")) > 1 {
+				continue
+			}
+			isMatched, err := re.MatchString(s.Symbol)
+			if err != nil {
+				return err
+			}
+			if isMatched {
+				var fd *runner.Fundamental
+				if val, ok := listings[s.Symbol]; ok {
+					fd = &val
+				}
+				if err := m.watcher.watch(s.Symbol, m.parseRunnerConfigs(runner.Futures), fd); err != nil {
+					m.watcher.logger.Error.Println(m.watcher.newLog(s.Symbol+"-"+string(runner.Futures), err.Error()))
 				}
 			}
 		}
@@ -206,8 +232,12 @@ func (m *MarketStruct) connect() {
 }
 
 // watcher endpoints
-func (m *MarketStruct) Watch(ticker string) error {
-	return m.watcher.watch(ticker+m.configs.Market.Watcher.BaseMarket, m.parseRunnerConfigs(), nil)
+func (m *MarketStruct) Watch(ticker, market string) error {
+	mk, ok := runner.ValidateMarket(market)
+	if !ok {
+		return errors.New("unsupported market")
+	}
+	return m.watcher.watch(ticker+m.configs.Market.Watcher.BaseMarket, m.parseRunnerConfigs(mk), nil)
 }
 
 func (m *MarketStruct) Watchlist() []string {

diff --git a/internal/cmd/market/provider.go b/internal/cmd/market/provider.go
@@ -21,6 +21,7 @@ const (
 
 type provider struct {
 	binSpot *bn.Client
+	binFutu *bnf.Client
 	coinCap *cc.Client
 }
 
@@ -69,7 +70,7 @@ type fetchOptions struct {
 	end   *time.Time
 }
 
-func (p *provider) fetchBinanceKlinesV3(ticker string, d time.Duration, opt *fetchOptions) ([]*ta.Candle, error) {
+func (p *provider) fetchBinanceSpotKlinesV3(ticker string, d time.Duration, opt *fetchOptions) ([]*ta.Candle, error) {
 	lmt := 1000
 	if opt != nil && opt.limit > 0 && opt.limit <= 1000 {
 		lmt = opt.limit
@@ -105,6 +106,42 @@ func (p *provider) fetchBinanceKlinesV3(ticker string, d time.Duration, opt *fet
 	return candles, nil
 }
 
+func (p *provider) fetchBinanceFuturesKlinesV3(ticker string, d time.Duration, opt *fetchOptions) ([]*ta.Candle, error) {
+	lmt := 1000
+	if opt != nil && opt.limit > 0 && opt.limit <= 1000 {
+		lmt = opt.limit
+	}
+	re, _ := regexp.Compile(timeFramePattern)
+	interval := re.FindString(d.String())
+	if d >= time.Hour*24 {
+		interval = "1d"
+	}
+	if d == time.Minute*10 {
+		interval = "5m"
+	}
+	end := time.Now().Unix() * 1000
+	if opt != nil && opt.end != nil {
+		end = opt.end.Unix() * 1000
+	}
+	var klines []*bnf.Kline
+	for len(klines) < lmt || (opt.start != nil && len(klines) > 0 && klines[0].OpenTime > opt.start.Unix()*1000) {
+		service := p.binFutu.NewKlinesService().Symbol(ticker).Interval(interval).EndTime(end).Limit(lmt)
+		kls, err := service.Do(context.Background())
+		if err != nil {
+			return nil, err
+		}
+		klines = append(kls, klines...)
+		if len(kls) > 0 {
+			end = kls[0].CloseTime
+		}
+	}
+	var candles []*ta.Candle
+	for _, kline := range klines {
+		candles = append(candles, tax.ConvertBinanceFuturesKline(kline, &d))
+	}
+	return candles, nil
+}
+
 func (p *provider) fetchCoinFundamentals(base string, limit int) (map[string]runner.Fundamental, error) {
 	out := make(map[string]runner.Fundamental)
 	listings, err := p.coinCap.Cryptocurrency.LatestListings(&cc.ListingOptions{Limit: limit})

diff --git a/internal/cmd/market/provider_test.go b/internal/cmd/market/provider_test.go
@@ -1,7 +1,9 @@
 package market
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"follow.markets/pkg/config"
 	"github.com/stretchr/testify/assert"
@@ -42,4 +44,14 @@ func Test_Provider(t *testing.T) {
 	listings, err := provider.fetchCoinFundamentals(configs.Market.Watcher.BaseMarket, 1)
 	assert.EqualValues(t, nil, err)
 	assert.EqualValues(t, true, len(listings) == 1)
+
+	_, err := provider.binFutu.NewListPriceChangeStatsService().Do(context.Background())
+	assert.EqualValues(t, nil, err)
+	//for _, s := range stats {
+	//	fmt.Println(fmt.Sprintf("%v", s.Symbol))
+	//}
+
+	klines, err := provider.fetchBinanceFuturesKlinesV3("BTCUSDT", time.Minute, &fetchOptions{limit: 60})
+	assert.EqualValues(t, nil, err)
+	assert.EqualValues(t, 60, len(klines))
 }