Less than 1000candles use the cache

Author: adrianceding

src/binance-proxy/handler/handler.go

@@ -10,14 +10,10 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-func NewHandler(
-	ctx context.Context, class service.Class,
-	enableFakeKline, startTickerWithKline, startDepthWithKline bool,
-) func(w http.ResponseWriter, r *http.Request) {
+func NewHandler(ctx context.Context, class service.Class) func(w http.ResponseWriter, r *http.Request) {
 	handler := &Handler{
-		srv:             service.NewService(ctx, class, startTickerWithKline, startDepthWithKline),
-		class:           class,
-		enableFakeKline: enableFakeKline,
+		srv:   service.NewService(ctx, class),
+		class: class,
 	}
 	handler.ctx, handler.cancel = context.WithCancel(ctx)
 
@@ -28,9 +24,8 @@ type Handler struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 
-	class           service.Class
-	srv             *service.Service
-	enableFakeKline bool
+	class service.Class
+	srv   *service.Service
 }
 
 func (s *Handler) Router(w http.ResponseWriter, r *http.Request) {

src/binance-proxy/handler/kline.go

@@ -5,71 +5,59 @@ import (
 	"net/http"
 	"strconv"
 	"time"
+
+	"binance-proxy/service"
 )
 
 func (s *Handler) klines(w http.ResponseWriter, r *http.Request) {
 	symbol := r.URL.Query().Get("symbol")
 	interval := r.URL.Query().Get("interval")
-	limit := r.URL.Query().Get("limit")
-	if limit == "" {
-		limit = "500"
+	limitInt, _ := strconv.Atoi(r.URL.Query().Get("limit"))
+	if limitInt == 0 {
+		limitInt = 500
 	}
-	limitInt, err := strconv.Atoi(limit)
+
+	startTimeUnix, _ := strconv.Atoi(r.URL.Query().Get("startTime"))
+	startTime := time.Unix(int64(startTimeUnix/1000), 0)
 
 	switch {
-	case err != nil, limitInt <= 0, limitInt > 1000,
-		r.URL.Query().Get("startTime") != "", r.URL.Query().Get("endTime") != "",
+	case limitInt <= 0, limitInt > 1000,
+		startTime.Unix() > 0 && startTime.Before(time.Now().Add(service.INTERVAL_2_DURATION[interval]*999*-1)),
+		r.URL.Query().Get("endTime") != "",
 		symbol == "", interval == "":
 		s.reverseProxy(w, r)
 		return
 	}
 
 	data := s.srv.Klines(symbol, interval)
-	if data == nil {
-		s.reverseProxy(w, r)
-		return
-	}
-
-	minLen := len(data)
-	if minLen > limitInt {
-		minLen = limitInt
+	klines := make([]interface{}, 0)
+	startTimeUnixMs := startTime.Unix() * 1000
+	if startTimeUnixMs == 0 && limitInt < len(data) {
+		data = data[len(data)-limitInt:]
 	}
+	for _, v := range data {
+		if len(klines) >= limitInt {
+			break
+		}
 
-	klines := make([]interface{}, minLen)
-	for i := minLen; i > 0; i-- {
-		ri := len(data) - i
-		klines[minLen-i] = []interface{}{
-			data[ri].OpenTime,
-			data[ri].Open,
-			data[ri].High,
-			data[ri].Low,
-			data[ri].Close,
-			data[ri].Volume,
-			data[ri].CloseTime,
-			data[ri].QuoteAssetVolume,
-			data[ri].TradeNum,
-			data[ri].TakerBuyBaseAssetVolume,
-			data[ri].TakerBuyQuoteAssetVolume,
-			"0",
+		if startTimeUnixMs > 0 && startTimeUnixMs > v.OpenTime {
+			continue
 		}
-	}
 
-	if s.enableFakeKline && len(data) > 0 && time.Now().UnixNano()/1e6 > data[len(data)-1].CloseTime {
 		klines = append(klines, []interface{}{
-			data[len(data)-1].CloseTime + 1,
-			data[len(data)-1].Close,
-			data[len(data)-1].Close,
-			data[len(data)-1].Close,
-			data[len(data)-1].Close,
-			"0.0",
-			data[len(data)-1].CloseTime + 1 + (data[len(data)-1].CloseTime - data[len(data)-1].OpenTime),
-			"0.0",
-			0,
-			"0.0",
-			"0.0",
+			v.OpenTime,
+			v.Open,
+			v.High,
+			v.Low,
+			v.Close,
+			v.Volume,
+			v.CloseTime,
+			v.QuoteAssetVolume,
+			v.TradeNum,
+			v.TakerBuyBaseAssetVolume,
+			v.TakerBuyQuoteAssetVolume,
 			"0",
 		})
-		klines = klines[len(klines)-minLen:]
 	}
 
 	w.Header().Set("Content-Type", "application/json")

src/binance-proxy/main.go

@@ -17,7 +17,7 @@ import (
 
 func startProxy(ctx context.Context, address string, class service.Class) {
 	mux := http.NewServeMux()
-	mux.HandleFunc("/", handler.NewHandler(ctx, class, flagEnableFakeKline, flagStartTickerWithKline, flagStartDepthWithKline))
+	mux.HandleFunc("/", handler.NewHandler(ctx, class))
 
 	log.Infof("Start %s proxy !Address: %s", class, address)
 	if err := http.ListenAndServe(address, mux); err != nil {
@@ -40,16 +40,10 @@ var ctx, cancel = context.WithCancel(context.Background())
 var flagSpotAddress string
 var flagFuturesAddress string
 var flagDebug bool
-var flagEnableFakeKline bool
-var flagStartTickerWithKline bool
-var flagStartDepthWithKline bool
 
 func main() {
 	flag.StringVar(&flagSpotAddress, "s", ":8090", "spot bind address.")
 	flag.StringVar(&flagFuturesAddress, "f", ":8091", "futures bind address.")
-	flag.BoolVar(&flagEnableFakeKline, "fakekline", false, "enable fake kline.")
-	flag.BoolVar(&flagStartTickerWithKline, "at", false, "auto monitor ticker with kline request.")
-	flag.BoolVar(&flagStartDepthWithKline, "ao", false, "auto monitor depth with kline request.")
 	flag.BoolVar(&flagDebug, "v", false, "print debug log.")
 	flag.Parse()
 

src/binance-proxy/service/service.go

@@ -12,9 +12,6 @@ type Service struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 
-	startTickerWithKline bool
-	startDepthWithKline  bool
-
 	class           Class
 	exchangeInfoSrv *ExchangeInfoSrv
 	klinesSrv       sync.Map // map[symbolInterval]*Klines
@@ -26,12 +23,8 @@ type Service struct {
 	lastGetTicker sync.Map // map[symbolInterval]time.Time
 }
 
-func NewService(ctx context.Context, class Class, startTickerWithKline, startDepthWithKline bool) *Service {
-	s := &Service{
-		class:                class,
-		startTickerWithKline: startTickerWithKline,
-		startDepthWithKline:  startDepthWithKline,
-	}
+func NewService(ctx context.Context, class Class) *Service {
+	s := &Service{class: class}
 	s.ctx, s.cancel = context.WithCancel(ctx)
 	s.exchangeInfoSrv = NewExchangeInfoSrv(s.ctx, NewSymbolInterval(s.class, "", ""))
 	s.exchangeInfoSrv.Start()
@@ -83,9 +76,7 @@ func (s *Service) autoRemoveExpired() {
 		srv := v.(*DepthSrv)
 
 		if t, ok := s.lastGetDepth.Load(si); ok {
-			_, isKlineAlive := aliveKlines[si.Symbol]
-
-			if ((s.startDepthWithKline && !isKlineAlive) || !s.startDepthWithKline) && time.Now().Sub(t.(time.Time)) > 2*time.Minute {
+			if time.Now().Sub(t.(time.Time)) > 2*time.Minute {
 				log.Debugf("%s.Depth srv expired!Removed", si)
 				s.lastGetDepth.Delete(si)
 
@@ -103,9 +94,7 @@ func (s *Service) autoRemoveExpired() {
 		srv := v.(*TickerSrv)
 
 		if t, ok := s.lastGetTicker.Load(si); ok {
-			_, isKlineAlive := aliveKlines[si.Symbol]
-
-			if ((s.startTickerWithKline && !isKlineAlive) || !s.startTickerWithKline) && time.Now().Sub(t.(time.Time)) > 2*time.Minute {
+			if time.Now().Sub(t.(time.Time)) > 2*time.Minute {
 				log.Debugf("%s.Ticker srv expired!Removed", si)
 				s.lastGetTicker.Delete(si)
 
@@ -136,12 +125,6 @@ func (s *Service) ExchangeInfo() []byte {
 func (s *Service) Klines(symbol, interval string) []*Kline {
 	si := NewSymbolInterval(s.class, symbol, interval)
 	srv := s.StartKlineSrv(si)
-	if s.startTickerWithKline {
-		s.StartTickerSrv(NewSymbolInterval(s.class, symbol, ""))
-	}
-	if s.startDepthWithKline {
-		s.StartDepthSrv(NewSymbolInterval(s.class, symbol, ""))
-	}
 
 	s.lastGetKlines.Store(*si, time.Now())