From 1f8493c0cf4222ce5767a962674f97bdc819de7a Mon Sep 17 00:00:00 2001 From: shinchan-zhai Date: Mon, 23 Mar 2026 19:20:04 +0800 Subject: [PATCH] fix: goroutine leak on shutdown, unsafe type assertions, redundant API calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Agent: add stopCh to cleanly stop chat-history-cleanup goroutine on Agent.Stop() (previously leaked a goroutine+ticker forever on every restart) - Agent: add looksLikeStockQuery() guard to avoid hitting Sina search API on every single message — only calls external API when text contains stock-related content (saves ~200ms latency + API call on crypto-only queries) - market/historical.go: safe type assertions in kline parsing (was bare .(float64) which panics on unexpected API responses), reuse HTTP client for connection pooling - market/api_client.go: safe comma-ok type assertions for all kline field parsing (11 bare assertions → all guarded) - trader/bybit: fix unsafe type assertion in CloseShort — pos["positionAmt"].(float64) could panic if field is nil/wrong type (critical: handles real money) --- agent/agent.go | 88 ++++++++++++++++++++++++++++++----- market/api_client.go | 46 +++++++++++++----- market/historical.go | 41 +++++++++++----- trader/bybit/trader_orders.go | 4 +- 4 files changed, 144 insertions(+), 35 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index c55b7c8d..f14b8e51 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -34,6 +34,7 @@ type Agent struct { logger *slog.Logger history *chatHistory pending *pendingTrades + stopCh chan struct{} // signals background goroutines to stop NotifyFunc func(userID int64, text string) error } @@ -55,7 +56,7 @@ func DefaultConfig() *Config { func New(tm *manager.TraderManager, st *store.Store, cfg *Config, logger *slog.Logger) *Agent { if cfg == nil { cfg = DefaultConfig() } - return &Agent{traderManager: tm, store: st, config: cfg, logger: logger, history: newChatHistory(20), pending: newPendingTrades()} + return &Agent{traderManager: tm, store: st, config: cfg, logger: logger, history: newChatHistory(20), pending: newPendingTrades(), stopCh: make(chan struct{})} } func (a *Agent) SetAIClient(c mcp.AIClient) { a.aiClient = c } @@ -102,8 +103,13 @@ func (a *Agent) Start() { safe.GoNamed("chat-history-cleanup", func() { ticker := time.NewTicker(30 * time.Minute) defer ticker.Stop() - for range ticker.C { - a.history.CleanOld(4 * time.Hour) + for { + select { + case <-ticker.C: + a.history.CleanOld(4 * time.Hour) + case <-a.stopCh: + return + } } }) @@ -111,6 +117,13 @@ func (a *Agent) Start() { } func (a *Agent) Stop() { + // Signal all background goroutines (e.g. chat-history-cleanup) to exit. + select { + case <-a.stopCh: + // Already closed + default: + close(a.stopCh) + } if a.sentinel != nil { a.sentinel.Stop() } if a.brain != nil { a.brain.Stop() } if a.scheduler != nil { a.scheduler.Stop() } @@ -614,14 +627,17 @@ func (a *Agent) gatherContext(text string) string { } } - // A-share / stocks — try Sina Finance (dynamic search as fallback) - stockCode, stockName := resolveStockCodeDynamic(text) - if stockCode != "" { - quote, err := fetchStockQuote(stockCode) - if err == nil && quote.Price > 0 { - parts = append(parts, fmt.Sprintf("[%s(%s) Real-time A-share Data]\n%s", quote.Name, quote.Code, formatStockQuote(quote))) - } else if err != nil { - a.logger.Error("fetch stock quote", "code", stockCode, "name", stockName, "error", err) + // A-share / stocks — only call Sina API when text likely references stocks. + // Skip for purely crypto conversations to avoid unnecessary external API calls. + if looksLikeStockQuery(text) { + stockCode, stockName := resolveStockCodeDynamic(text) + if stockCode != "" { + quote, err := fetchStockQuote(stockCode) + if err == nil && quote.Price > 0 { + parts = append(parts, fmt.Sprintf("[%s(%s) Real-time A-share Data]\n%s", quote.Name, quote.Code, formatStockQuote(quote))) + } else if err != nil { + a.logger.Error("fetch stock quote", "code", stockCode, "name", stockName, "error", err) + } } } @@ -746,6 +762,56 @@ func (a *Agent) notifyAll(text string) { if a.NotifyFunc != nil { a.NotifyFunc(0, text) } } +// looksLikeStockQuery returns true if the text likely references stocks rather +// than being a pure crypto/general query. This avoids hitting the Sina search +// API on every single message (saves ~200ms latency + external API call). +func looksLikeStockQuery(text string) bool { + upper := strings.ToUpper(text) + + // Check for known stock-related Chinese keywords + stockKeywords := []string{ + "股", "A股", "港股", "美股", "股票", "涨停", "跌停", "大盘", + "沪指", "深指", "恒指", "纳指", "标普", "道琼斯", + "茅台", "比亚迪", "宁德", "腾讯", "阿里", "美团", "小米", + "京东", "百度", "苹果", "特斯拉", "英伟达", "微软", "谷歌", + "盘前", "盘后", "开盘", "收盘", "涨幅", "跌幅", + } + for _, kw := range stockKeywords { + if strings.Contains(text, kw) { + return true + } + } + + // Check for US stock ticker patterns (1-5 uppercase letters not matching crypto) + for _, word := range strings.Fields(upper) { + word = strings.Trim(word, ".,!?;:()[]{}\"'") + if len(word) >= 1 && len(word) <= 5 { + allLetter := true + for _, c := range word { + if c < 'A' || c > 'Z' { allLetter = false; break } + } + if allLetter { + // Check if it's in the known US ticker map + if _, ok := usTickerMap[word]; ok { + return true + } + } + } + } + + // Check for 6-digit A-share codes or 5-digit HK codes + for _, w := range strings.Fields(text) { + w = strings.TrimSpace(w) + if (len(w) == 5 || len(w) == 6) { + if _, err := strconv.Atoi(w); err == nil { + return true + } + } + } + + return false +} + func toFloat(v interface{}) float64 { switch x := v.(type) { case float64: return x diff --git a/market/api_client.go b/market/api_client.go index f3058482..7e751d7d 100644 --- a/market/api_client.go +++ b/market/api_client.go @@ -121,18 +121,40 @@ func parseKline(kr KlineResponse) (Kline, error) { return kline, fmt.Errorf("invalid kline data") } - // Parse each field - kline.OpenTime = int64(kr[0].(float64)) - kline.Open, _ = strconv.ParseFloat(kr[1].(string), 64) - kline.High, _ = strconv.ParseFloat(kr[2].(string), 64) - kline.Low, _ = strconv.ParseFloat(kr[3].(string), 64) - kline.Close, _ = strconv.ParseFloat(kr[4].(string), 64) - kline.Volume, _ = strconv.ParseFloat(kr[5].(string), 64) - kline.CloseTime = int64(kr[6].(float64)) - kline.QuoteVolume, _ = strconv.ParseFloat(kr[7].(string), 64) - kline.Trades = int(kr[8].(float64)) - kline.TakerBuyBaseVolume, _ = strconv.ParseFloat(kr[9].(string), 64) - kline.TakerBuyQuoteVolume, _ = strconv.ParseFloat(kr[10].(string), 64) + // Parse each field with safe type assertions to prevent panics on unexpected API responses + if v, ok := kr[0].(float64); ok { + kline.OpenTime = int64(v) + } + if v, ok := kr[1].(string); ok { + kline.Open, _ = strconv.ParseFloat(v, 64) + } + if v, ok := kr[2].(string); ok { + kline.High, _ = strconv.ParseFloat(v, 64) + } + if v, ok := kr[3].(string); ok { + kline.Low, _ = strconv.ParseFloat(v, 64) + } + if v, ok := kr[4].(string); ok { + kline.Close, _ = strconv.ParseFloat(v, 64) + } + if v, ok := kr[5].(string); ok { + kline.Volume, _ = strconv.ParseFloat(v, 64) + } + if v, ok := kr[6].(float64); ok { + kline.CloseTime = int64(v) + } + if v, ok := kr[7].(string); ok { + kline.QuoteVolume, _ = strconv.ParseFloat(v, 64) + } + if v, ok := kr[8].(float64); ok { + kline.Trades = int(v) + } + if v, ok := kr[9].(string); ok { + kline.TakerBuyBaseVolume, _ = strconv.ParseFloat(v, 64) + } + if v, ok := kr[10].(string); ok { + kline.TakerBuyQuoteVolume, _ = strconv.ParseFloat(v, 64) + } return kline, nil } diff --git a/market/historical.go b/market/historical.go index 18d8f51a..3ae82874 100644 --- a/market/historical.go +++ b/market/historical.go @@ -30,7 +30,14 @@ func GetKlinesRange(symbol string, timeframe string, start, end time.Time) ([]Kl var all []Kline cursor := startMs - client := &http.Client{Timeout: 15 * time.Second} + client := &http.Client{ + Timeout: 15 * time.Second, + Transport: &http.Transport{ + MaxIdleConns: 5, + MaxIdleConnsPerHost: 5, + IdleConnTimeout: 90 * time.Second, + }, + } for cursor < endMs { req, err := http.NewRequest("GET", binanceFuturesKlinesURL, nil) @@ -68,25 +75,37 @@ func GetKlinesRange(symbol string, timeframe string, start, end time.Time) ([]Kl break } - batch := make([]Kline, len(raw)) - for i, item := range raw { - openTime := int64(item[0].(float64)) + batch := make([]Kline, 0, len(raw)) + for _, item := range raw { + if len(item) < 7 { + continue // skip malformed entries + } + openTimeF, ok := item[0].(float64) + if !ok { + continue + } + closeTimeF, ok := item[6].(float64) + if !ok { + continue + } open, _ := parseFloat(item[1]) high, _ := parseFloat(item[2]) low, _ := parseFloat(item[3]) - close, _ := parseFloat(item[4]) + cls, _ := parseFloat(item[4]) volume, _ := parseFloat(item[5]) - closeTime := int64(item[6].(float64)) - batch[i] = Kline{ - OpenTime: openTime, + batch = append(batch, Kline{ + OpenTime: int64(openTimeF), Open: open, High: high, Low: low, - Close: close, + Close: cls, Volume: volume, - CloseTime: closeTime, - } + CloseTime: int64(closeTimeF), + }) + } + if len(batch) == 0 { + break } all = append(all, batch...) diff --git a/trader/bybit/trader_orders.go b/trader/bybit/trader_orders.go index ab62ee45..7f6d1881 100644 --- a/trader/bybit/trader_orders.go +++ b/trader/bybit/trader_orders.go @@ -154,7 +154,9 @@ func (t *BybitTrader) CloseShort(symbol string, quantity float64) (map[string]in for _, pos := range positions { side, _ := pos["side"].(string) if pos["symbol"] == symbol && strings.ToLower(side) == "short" { - quantity = -pos["positionAmt"].(float64) // Short position is negative + if amt, ok := pos["positionAmt"].(float64); ok { + quantity = -amt // Short position is negative + } break } }