Merge pull request 'perf: parallelize token source price requests' (#20) from feat/parallel-price-lookup-v2 into feat/token-message-rich

Reviewed-on: #20

internal/data/imarket.go

@@ -13,6 +13,7 @@ type IMarket interface {
 	// Alpha token methods
 	IsAlphaToken(symbol string) bool
 	GetAlphaToken(symbol string) (market.AlphaTokenInfo, bool)
+	GetAlphaPrice(symbol string) (float64, bool)
 
 	// Trading pair methods
 	IsSpotPair(symbol string) bool

internal/data/market/alpha_tokens.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"net/url"
 	"strconv"
 	"time"
 
@@ -54,10 +55,21 @@ type AlphaTokenInfo struct {
 
 // AlphaTokenResponse represents the API response structure
 type AlphaTokenResponse struct {
-	Code         string           `json:"code"`
+	Code          string           `json:"code"`
-	Message      *string          `json:"message"`
+	Message       *string          `json:"message"`
+	MessageDetail *string          `json:"messageDetail"`
+	Data          []AlphaTokenInfo `json:"data"`
+}
+
+type AlphaTickerData struct {
+	Price string `json:"price"`
+}
+
+type AlphaTickerResponse struct {
+	Code          string          `json:"code"`
+	Message       *string         `json:"message"`
 	MessageDetail *string         `json:"messageDetail"`
-	Data         []AlphaTokenInfo `json:"data"`
+	Data          AlphaTickerData `json:"data"`
 }
 
 // GetPrice returns the price as float64
@@ -150,3 +162,37 @@ func (ms *MarketData) IsAlphaToken(symbol string) bool {
 	_, exists := ms.GetAlphaToken(symbol)
 	return exists
 }
+
+func (ms *MarketData) GetAlphaPrice(symbol string) (float64, bool) {
+	const alphaTickerURL = "https://www.binance.com/bapi/defi/v1/public/alpha-trade/ticker"
+
+	client := &http.Client{Timeout: 5 * time.Second}
+	endpoint := alphaTickerURL + "?" + url.Values{"symbol": []string{symbol}}.Encode()
+
+	resp, err := client.Get(endpoint)
+	if err != nil {
+		log.Error().Err(err).Str("symbol", symbol).Msg("Failed to fetch Alpha ticker")
+		return 0, false
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return 0, false
+	}
+
+	var tickerResp AlphaTickerResponse
+	if err := json.NewDecoder(resp.Body).Decode(&tickerResp); err != nil {
+		return 0, false
+	}
+
+	if tickerResp.Code != "000000" {
+		return 0, false
+	}
+
+	price, err := strconv.ParseFloat(tickerResp.Data.Price, 64)
+	if err != nil || price <= 0 {
+		return 0, false
+	}
+
+	return price, true
+}

internal/services/tele/commands/token.go

@@ -2,11 +2,11 @@ package commands
 
 import (
 	"strings"
+	"sync"
 
 	"gopkg.in/telebot.v3"
 	"me.thuanle/bbot/internal/configs/tele"
 	"me.thuanle/bbot/internal/data"
-	"me.thuanle/bbot/internal/helper/binancex"
 	"me.thuanle/bbot/internal/services/tele/chat"
 	"me.thuanle/bbot/internal/services/tele/view"
 )
@@ -62,68 +62,79 @@ func showStickerMode(context telebot.Context, token string) {
 	}
 }
 
+func collectRichTokenData(token string) buildRichTokenMessageArgs {
+	a := buildRichTokenMessageArgs{Token: token}
+
+	marginRates := data.Market.GetMarginInterestRates()
+	a.MarginAPRPercent = marginRates[token] * 365 * 100
+	a.HasMarginAPR = marginRates[token] != 0
+
+	futureSymbol := token + "USDT"
+	spotSymbol := token + "USDT"
+
+	var (
+		wg sync.WaitGroup
+		mu sync.Mutex
+	)
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+		if alphaToken, ok := data.Market.GetAlphaToken(token); ok {
+			alpha24h := alphaToken.GetPercentChange24h()
+			alphaPrice := alphaToken.GetPrice()
+			if p, ok := data.Market.GetAlphaPrice(token + "USDT"); ok {
+				alphaPrice = p
+			}
+			mu.Lock()
+			a.HasAlpha = true
+			a.HasAlpha24h = true
+			a.Alpha24h = alpha24h
+			a.AlphaPrice = alphaPrice
+			mu.Unlock()
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		if data.Market.IsFuturesPair(futureSymbol) {
+			if fp, fr, ft, ok := data.Market.GetFuturePrice(futureSymbol); ok {
+				mu.Lock()
+				a.HasFuture = true
+				a.FuturePrice = fp
+				a.FundingRate = fr
+				a.FundingTimeMs = ft
+				mu.Unlock()
+			}
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		if data.Market.IsSpotPair(spotSymbol) {
+			if sp, ok := data.Market.GetSpotPrice(spotSymbol); ok {
+				mu.Lock()
+				a.HasSpot = true
+				a.SpotPrice = sp
+				mu.Unlock()
+			}
+		}
+	}()
+
+	wg.Wait()
+	return a
+}
+
 func OnTokenInfoByToken(context telebot.Context, token string) error {
 	token = strings.ToUpper(token)
 	showStickerMode(context, token)
 
-	var (
+	args := collectRichTokenData(token)
-		hasAlpha, hasAlpha24h bool
+	if !args.HasSpot && !args.HasFuture && !args.HasAlpha {
-		alphaPrice, alpha24h  float64
-		hasFuture, hasSpot    bool
-		futurePrice           float64
-		fundingRate           float64
-		fundingTime           int64
-		spotPrice             float64
-	)
-
-	if alphaToken, ok := data.Market.GetAlphaToken(token); ok {
-		hasAlpha = true
-		alphaPrice = alphaToken.GetPrice()
-		hasAlpha24h = true
-		alpha24h = alphaToken.GetPercentChange24h()
-	}
-
-	symbols := binancex.Token2Symbols(token)
-	if len(symbols) > 0 {
-		fp, fr, ft, ok := data.Market.GetFuturePrice(symbols[0])
-		if ok {
-			hasFuture = true
-			futurePrice = fp
-			fundingRate = fr
-			fundingTime = ft
-
-			sSymbol := binancex.Future2SpotSymbol(symbols[0])
-			if sp, ok := data.Market.GetSpotPrice(sSymbol); ok {
-				hasSpot = true
-				spotPrice = sp
-			}
-		}
-	}
-
-	marginRates := data.Market.GetMarginInterestRates()
-	marginAPR := marginRates[token] * 365 * 100
-	hasMarginAPR := marginRates[token] != 0
-
-	if !hasSpot && !hasFuture && !hasAlpha {
 		return nil
 	}
 
-	msg := view.RenderRichTokenMessage(buildRichTokenMessageInput(buildRichTokenMessageArgs{
+	msg := view.RenderRichTokenMessage(buildRichTokenMessageInput(args))
-		Token:            token,
-		HasSpot:          hasSpot,
-		SpotPrice:        spotPrice,
-		HasFuture:        hasFuture,
-		FuturePrice:      futurePrice,
-		FundingRate:      fundingRate,
-		FundingTimeMs:    fundingTime,
-		HasAlpha:         hasAlpha,
-		AlphaPrice:       alphaPrice,
-		HasAlpha24h:      hasAlpha24h,
-		Alpha24h:         alpha24h,
-		HasMarginAPR:     hasMarginAPR,
-		MarginAPRPercent: marginAPR,
-	}))
-
 	_ = chat.ReplyMessage(context, msg)
 	return nil
 }

internal/services/tele/commands/token_test.go

@@ -1,6 +1,13 @@
 package commands
 
-import "testing"
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"me.thuanle/bbot/internal/data"
+	"me.thuanle/bbot/internal/data/market"
+)
 
 func TestBuildRichTokenMessageInput_AllSources(t *testing.T) {
 	in := buildRichTokenMessageInput(buildRichTokenMessageArgs{
@@ -40,3 +47,184 @@ func TestBuildRichTokenMessageInput_OnlyAlpha(t *testing.T) {
 		t.Fatalf("did not expect spot/future flags true: %+v", in)
 	}
 }
+
+type marketStub struct {
+	spotPairs    map[string]bool
+	futuresPairs map[string]bool
+	spotPrices   map[string]float64
+	futurePrices map[string]struct {
+		price float64
+		rate  float64
+		time  int64
+	}
+	alphaTokens map[string]market.AlphaTokenInfo
+	alphaPrices map[string]float64
+	marginRates map[string]float64
+
+	gate   *sync.WaitGroup
+	ready  chan struct{}
+	done   chan struct{}
+	mu     sync.Mutex
+	starts int
+}
+
+func (m *marketStub) waitConcurrentGate() {
+	if m.gate == nil {
+		return
+	}
+	m.mu.Lock()
+	m.starts++
+	m.mu.Unlock()
+	m.gate.Done()
+	<-m.ready
+}
+
+func (m *marketStub) GetFuturePrice(symbol string) (float64, float64, int64, bool) {
+	m.waitConcurrentGate()
+	v, ok := m.futurePrices[symbol]
+	if !ok {
+		return 0, 0, 0, false
+	}
+	return v.price, v.rate, v.time, true
+}
+func (m *marketStub) GetAllPremiumIndex() (map[string]market.PremiumIndex, error) { return nil, nil }
+func (m *marketStub) GetAllFundRate() (map[string]float64, map[string]int64)      { return nil, nil }
+func (m *marketStub) GetSpotPrice(symbol string) (float64, bool) {
+	m.waitConcurrentGate()
+	v, ok := m.spotPrices[symbol]
+	return v, ok
+}
+func (m *marketStub) GetMarginInterestRates() map[string]float64 { return m.marginRates }
+func (m *marketStub) IsAlphaToken(symbol string) bool {
+	_, ok := m.alphaTokens[symbol]
+	return ok
+}
+func (m *marketStub) GetAlphaToken(symbol string) (market.AlphaTokenInfo, bool) {
+	v, ok := m.alphaTokens[symbol]
+	return v, ok
+}
+func (m *marketStub) GetAlphaPrice(symbol string) (float64, bool) {
+	m.waitConcurrentGate()
+	v, ok := m.alphaPrices[symbol]
+	return v, ok
+}
+func (m *marketStub) IsSpotPair(symbol string) bool    { return m.spotPairs[symbol] }
+func (m *marketStub) IsFuturesPair(symbol string) bool { return m.futuresPairs[symbol] }
+func (m *marketStub) RefreshTradingPairCache() error   { return nil }
+
+func TestCollectRichTokenData_SpotOnlyReachable(t *testing.T) {
+	orig := data.Market
+	defer func() { data.Market = orig }()
+
+	data.Market = &marketStub{
+		spotPairs:  map[string]bool{"ABCUSDT": true},
+		spotPrices: map[string]float64{"ABCUSDT": 1.23},
+	}
+
+	args := collectRichTokenData("ABC")
+	if !args.HasSpot {
+		t.Fatalf("expected spot to be reachable for spot-only token: %+v", args)
+	}
+	if args.HasFuture || args.HasAlpha {
+		t.Fatalf("did not expect future/alpha for spot-only token: %+v", args)
+	}
+}
+
+func TestCollectRichTokenData_FutureFailureStillKeepsSpot(t *testing.T) {
+	orig := data.Market
+	defer func() { data.Market = orig }()
+
+	data.Market = &marketStub{
+		spotPairs:    map[string]bool{"ETHUSDT": true},
+		futuresPairs: map[string]bool{"ETHUSDT": true},
+		spotPrices:   map[string]float64{"ETHUSDT": 3245},
+		futurePrices: map[string]struct {
+			price float64
+			rate  float64
+			time  int64
+		}{},
+	}
+
+	args := collectRichTokenData("ETH")
+	if !args.HasSpot {
+		t.Fatalf("expected spot to remain available when future lookup fails: %+v", args)
+	}
+	if args.HasFuture {
+		t.Fatalf("did not expect future when future lookup fails: %+v", args)
+	}
+}
+
+func TestCollectRichTokenData_AlphaUsesSymbolPrice(t *testing.T) {
+	orig := data.Market
+	defer func() { data.Market = orig }()
+
+	data.Market = &marketStub{
+		alphaTokens: map[string]market.AlphaTokenInfo{
+			"ABC": {Symbol: "ABC", PercentChange24h: "12.4", Price: "0.1"},
+		},
+		alphaPrices: map[string]float64{"ABCUSDT": 0.1234},
+	}
+
+	args := collectRichTokenData("ABC")
+	if !args.HasAlpha {
+		t.Fatalf("expected alpha to be available: %+v", args)
+	}
+	if !args.HasAlpha24h {
+		t.Fatalf("expected alpha 24h to be available: %+v", args)
+	}
+	if args.AlphaPrice != 0.1234 {
+		t.Fatalf("expected alpha price from symbol lookup, got %.4f", args.AlphaPrice)
+	}
+}
+
+func TestCollectRichTokenData_RunsLookupsConcurrently(t *testing.T) {
+	orig := data.Market
+	defer func() { data.Market = orig }()
+
+	gate := &sync.WaitGroup{}
+	gate.Add(3)
+	st := &marketStub{
+		spotPairs:    map[string]bool{"ABCUSDT": true},
+		futuresPairs: map[string]bool{"ABCUSDT": true},
+		spotPrices:   map[string]float64{"ABCUSDT": 1.23},
+		futurePrices: map[string]struct {
+			price float64
+			rate  float64
+			time  int64
+		}{"ABCUSDT": {price: 1.24, rate: 0.0001, time: 1740000000000}},
+		alphaTokens: map[string]market.AlphaTokenInfo{
+			"ABC": {Symbol: "ABC", PercentChange24h: "2.5", Price: "1.22"},
+		},
+		alphaPrices: map[string]float64{"ABCUSDT": 1.22},
+		gate:        gate,
+		ready:       make(chan struct{}),
+	}
+	data.Market = st
+
+	finished := make(chan buildRichTokenMessageArgs, 1)
+	go func() {
+		finished <- collectRichTokenData("ABC")
+	}()
+
+	waitDone := make(chan struct{})
+	go func() {
+		gate.Wait()
+		close(waitDone)
+	}()
+
+	select {
+	case <-waitDone:
+		close(st.ready)
+	case <-time.After(200 * time.Millisecond):
+		t.Fatalf("expected price lookups to run concurrently")
+	}
+
+	select {
+	case args := <-finished:
+		if !args.HasSpot || !args.HasFuture || !args.HasAlpha {
+			t.Fatalf("expected all lookups present: %+v", args)
+		}
+	case <-time.After(200 * time.Millisecond):
+		t.Fatalf("collectRichTokenData did not finish")
+	}
+}