mirror of
https://github.com/NoFxAiOS/nofx.git
synced 2026-07-17 17:34:39 +08:00
fix: goroutine leak on shutdown, unsafe type assertions, redundant API calls
- 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)
This commit is contained in:
@@ -34,6 +34,7 @@ type Agent struct {
|
|||||||
logger *slog.Logger
|
logger *slog.Logger
|
||||||
history *chatHistory
|
history *chatHistory
|
||||||
pending *pendingTrades
|
pending *pendingTrades
|
||||||
|
stopCh chan struct{} // signals background goroutines to stop
|
||||||
NotifyFunc func(userID int64, text string) error
|
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 {
|
func New(tm *manager.TraderManager, st *store.Store, cfg *Config, logger *slog.Logger) *Agent {
|
||||||
if cfg == nil { cfg = DefaultConfig() }
|
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 }
|
func (a *Agent) SetAIClient(c mcp.AIClient) { a.aiClient = c }
|
||||||
@@ -102,8 +103,13 @@ func (a *Agent) Start() {
|
|||||||
safe.GoNamed("chat-history-cleanup", func() {
|
safe.GoNamed("chat-history-cleanup", func() {
|
||||||
ticker := time.NewTicker(30 * time.Minute)
|
ticker := time.NewTicker(30 * time.Minute)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
for range ticker.C {
|
for {
|
||||||
a.history.CleanOld(4 * time.Hour)
|
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() {
|
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.sentinel != nil { a.sentinel.Stop() }
|
||||||
if a.brain != nil { a.brain.Stop() }
|
if a.brain != nil { a.brain.Stop() }
|
||||||
if a.scheduler != nil { a.scheduler.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)
|
// A-share / stocks — only call Sina API when text likely references stocks.
|
||||||
stockCode, stockName := resolveStockCodeDynamic(text)
|
// Skip for purely crypto conversations to avoid unnecessary external API calls.
|
||||||
if stockCode != "" {
|
if looksLikeStockQuery(text) {
|
||||||
quote, err := fetchStockQuote(stockCode)
|
stockCode, stockName := resolveStockCodeDynamic(text)
|
||||||
if err == nil && quote.Price > 0 {
|
if stockCode != "" {
|
||||||
parts = append(parts, fmt.Sprintf("[%s(%s) Real-time A-share Data]\n%s", quote.Name, quote.Code, formatStockQuote(quote)))
|
quote, err := fetchStockQuote(stockCode)
|
||||||
} else if err != nil {
|
if err == nil && quote.Price > 0 {
|
||||||
a.logger.Error("fetch stock quote", "code", stockCode, "name", stockName, "error", err)
|
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) }
|
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 {
|
func toFloat(v interface{}) float64 {
|
||||||
switch x := v.(type) {
|
switch x := v.(type) {
|
||||||
case float64: return x
|
case float64: return x
|
||||||
|
|||||||
@@ -121,18 +121,40 @@ func parseKline(kr KlineResponse) (Kline, error) {
|
|||||||
return kline, fmt.Errorf("invalid kline data")
|
return kline, fmt.Errorf("invalid kline data")
|
||||||
}
|
}
|
||||||
|
|
||||||
// Parse each field
|
// Parse each field with safe type assertions to prevent panics on unexpected API responses
|
||||||
kline.OpenTime = int64(kr[0].(float64))
|
if v, ok := kr[0].(float64); ok {
|
||||||
kline.Open, _ = strconv.ParseFloat(kr[1].(string), 64)
|
kline.OpenTime = int64(v)
|
||||||
kline.High, _ = strconv.ParseFloat(kr[2].(string), 64)
|
}
|
||||||
kline.Low, _ = strconv.ParseFloat(kr[3].(string), 64)
|
if v, ok := kr[1].(string); ok {
|
||||||
kline.Close, _ = strconv.ParseFloat(kr[4].(string), 64)
|
kline.Open, _ = strconv.ParseFloat(v, 64)
|
||||||
kline.Volume, _ = strconv.ParseFloat(kr[5].(string), 64)
|
}
|
||||||
kline.CloseTime = int64(kr[6].(float64))
|
if v, ok := kr[2].(string); ok {
|
||||||
kline.QuoteVolume, _ = strconv.ParseFloat(kr[7].(string), 64)
|
kline.High, _ = strconv.ParseFloat(v, 64)
|
||||||
kline.Trades = int(kr[8].(float64))
|
}
|
||||||
kline.TakerBuyBaseVolume, _ = strconv.ParseFloat(kr[9].(string), 64)
|
if v, ok := kr[3].(string); ok {
|
||||||
kline.TakerBuyQuoteVolume, _ = strconv.ParseFloat(kr[10].(string), 64)
|
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
|
return kline, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -30,7 +30,14 @@ func GetKlinesRange(symbol string, timeframe string, start, end time.Time) ([]Kl
|
|||||||
var all []Kline
|
var all []Kline
|
||||||
cursor := startMs
|
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 {
|
for cursor < endMs {
|
||||||
req, err := http.NewRequest("GET", binanceFuturesKlinesURL, nil)
|
req, err := http.NewRequest("GET", binanceFuturesKlinesURL, nil)
|
||||||
@@ -68,25 +75,37 @@ func GetKlinesRange(symbol string, timeframe string, start, end time.Time) ([]Kl
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
batch := make([]Kline, len(raw))
|
batch := make([]Kline, 0, len(raw))
|
||||||
for i, item := range raw {
|
for _, item := range raw {
|
||||||
openTime := int64(item[0].(float64))
|
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])
|
open, _ := parseFloat(item[1])
|
||||||
high, _ := parseFloat(item[2])
|
high, _ := parseFloat(item[2])
|
||||||
low, _ := parseFloat(item[3])
|
low, _ := parseFloat(item[3])
|
||||||
close, _ := parseFloat(item[4])
|
cls, _ := parseFloat(item[4])
|
||||||
volume, _ := parseFloat(item[5])
|
volume, _ := parseFloat(item[5])
|
||||||
closeTime := int64(item[6].(float64))
|
|
||||||
|
|
||||||
batch[i] = Kline{
|
batch = append(batch, Kline{
|
||||||
OpenTime: openTime,
|
OpenTime: int64(openTimeF),
|
||||||
Open: open,
|
Open: open,
|
||||||
High: high,
|
High: high,
|
||||||
Low: low,
|
Low: low,
|
||||||
Close: close,
|
Close: cls,
|
||||||
Volume: volume,
|
Volume: volume,
|
||||||
CloseTime: closeTime,
|
CloseTime: int64(closeTimeF),
|
||||||
}
|
})
|
||||||
|
}
|
||||||
|
if len(batch) == 0 {
|
||||||
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
all = append(all, batch...)
|
all = append(all, batch...)
|
||||||
|
|||||||
@@ -154,7 +154,9 @@ func (t *BybitTrader) CloseShort(symbol string, quantity float64) (map[string]in
|
|||||||
for _, pos := range positions {
|
for _, pos := range positions {
|
||||||
side, _ := pos["side"].(string)
|
side, _ := pos["side"].(string)
|
||||||
if pos["symbol"] == symbol && strings.ToLower(side) == "short" {
|
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
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user