diff --git a/agent/agent.go b/agent/agent.go new file mode 100644 index 00000000..6986c107 --- /dev/null +++ b/agent/agent.go @@ -0,0 +1,499 @@ +// Package agent implements the NOFXi Agent Core. +// +// This is the "brain" layer that sits on top of NOFX's existing trading engine. +// It adds: natural language interaction, proactive market monitoring, +// trading memory/learning, and autonomous decision-making. +// +// Architecture: +// NOFX (engine) provides: kernel, trader, market, store, mcp +// Agent (brain) adds: perception, thinking, memory, interaction +// +// The agent does NOT replace any NOFX functionality — it enhances it. +package agent + +import ( + "context" + "fmt" + "log/slog" + "strconv" + "strings" + "time" + + "nofx/manager" + "nofx/market" + "nofx/store" +) + +// Agent is the NOFXi intelligence layer on top of NOFX. +type Agent struct { + // NOFX core (injected) + traderManager *manager.TraderManager + store *store.Store + + // Agent components + config *Config + sentinel *Sentinel + brain *Brain + scheduler *Scheduler + router *Router + logger *slog.Logger + + // Notification callback (set by telegram/web) + NotifyFunc func(userID int64, text string) error +} + +// Config holds agent-specific configuration. +type Config struct { + Language string `json:"language"` // "zh" or "en" + WatchSymbols []string `json:"watch_symbols"` // Default symbols to watch + EnableBriefs bool `json:"enable_briefs"` // Morning/evening market briefs + EnableNews bool `json:"enable_news"` // News scanning + EnableSentinel bool `json:"enable_sentinel"` // Market anomaly detection + BriefTimes []int `json:"brief_times"` // Hours to send briefs (e.g. [8, 20]) +} + +// DefaultConfig returns sensible defaults. +func DefaultConfig() *Config { + return &Config{ + Language: "zh", + WatchSymbols: []string{"BTCUSDT", "ETHUSDT", "SOLUSDT"}, + EnableBriefs: true, + EnableNews: true, + EnableSentinel: true, + BriefTimes: []int{8, 20}, + } +} + +// New creates a new NOFXi Agent. +func New( + traderMgr *manager.TraderManager, + st *store.Store, + cfg *Config, + logger *slog.Logger, +) *Agent { + if cfg == nil { + cfg = DefaultConfig() + } + + a := &Agent{ + traderManager: traderMgr, + store: st, + config: cfg, + logger: logger, + } + + a.router = NewRouter() + a.scheduler = NewScheduler(a, logger) + + return a +} + + + +// Start starts all agent services. +func (a *Agent) Start() { + a.logger.Info("starting NOFXi agent...") + + // Start sentinel (market anomaly detection) + if a.config.EnableSentinel { + a.sentinel = NewSentinel(a.config.WatchSymbols, a.handleSignal, a.logger) + a.sentinel.Start() + a.logger.Info("sentinel started", "symbols", a.config.WatchSymbols) + } + + // Start brain (proactive intelligence) + a.brain = NewBrain(a, a.logger) + if a.config.EnableNews { + a.brain.StartNewsScan(5 * time.Minute) + a.logger.Info("news scanner started") + } + if a.config.EnableBriefs { + a.brain.StartMarketBriefs(a.config.BriefTimes) + a.logger.Info("market briefs enabled", "hours", a.config.BriefTimes) + } + + // Start scheduler + a.scheduler.Start(context.Background()) + a.logger.Info("scheduler started") + + a.logger.Info("NOFXi agent is online 🚀") +} + +// Stop stops all agent services. +func (a *Agent) Stop() { + if a.sentinel != nil { + a.sentinel.Stop() + } + if a.brain != nil { + a.brain.Stop() + } + a.scheduler.Stop() + a.logger.Info("NOFXi agent stopped") +} + +// HandleMessage processes a user message and returns a response. +// This is the main entry point for Telegram/Web interaction. +func (a *Agent) HandleMessage(ctx context.Context, userID int64, text string) (string, error) { + // Extract language from prefix [lang:xx] + lang := a.config.Language + if strings.HasPrefix(text, "[lang:") { + if end := strings.Index(text, "] "); end > 0 { + lang = text[6:end] + text = text[end+2:] + } + } + + a.logger.Info("agent message", "user_id", userID, "text", text) + + intent := a.router.Route(text) + + switch intent.Type { + case IntentHelp: + return a.msg(lang, "help"), nil + case IntentStatus: + return a.handleStatus(lang), nil + case IntentQuery: + return a.handleQuery(lang, intent) + case IntentAnalyze: + return a.handleAnalyze(ctx, lang, intent) + case IntentTrade: + return a.handleTrade(lang, intent) + case IntentWatch: + return a.handleWatch(lang, intent), nil + case IntentStrategy: + return a.handleStrategyCmd(lang, intent), nil + default: + return a.handleChat(ctx, lang, userID, text) + } +} + +// --- Handlers using NOFX core --- + +func (a *Agent) handleStatus(L string) string { + traderCount := 0 + runningCount := 0 + if a.traderManager != nil { + all := a.traderManager.GetAllTraders() + traderCount = len(all) + for _, t := range all { + status := t.GetStatus() + if isRunning, ok := status["is_running"].(bool); ok && isRunning { + runningCount++ + } + } + } + watchCount := 0 + if a.sentinel != nil { + watchCount = a.sentinel.SymbolCount() + } + + return fmt.Sprintf(a.msg(L, "status"), + runningCount, traderCount, watchCount, time.Now().Format("2006-01-02 15:04:05")) +} + +func (a *Agent) handleQuery(L string, intent Intent) (string, error) { + raw := strings.ToLower(intent.Raw) + + // Get live data from NOFX trader manager + if a.traderManager == nil { + return a.msg(L, "no_traders"), nil + } + + // List all positions across all traders + if strings.Contains(raw, "position") || strings.Contains(raw, "持仓") { + return a.queryAllPositions(L) + } + if strings.Contains(raw, "balance") || strings.Contains(raw, "余额") { + return a.queryAllBalances(L) + } + if strings.Contains(raw, "trader") || strings.Contains(raw, "交易员") { + return a.queryTraders(L, nil) + } + + return a.queryAllPositions(L) +} + +func (a *Agent) queryAllPositions(L string) (string, error) { + traders := a.traderManager.GetAllTraders() + if len(traders) == 0 { + return a.msg(L, "no_traders"), nil + } + + var sb strings.Builder + sb.WriteString(a.msg(L, "positions_header")) + totalPnL := 0.0 + hasPosition := false + + for id, t := range traders { + positions, err := t.GetPositions() + if err != nil { + continue + } + for _, p := range positions { + size := toFloat(p["size"]) + if size == 0 { + continue + } + hasPosition = true + pnl := toFloat(p["unrealizedPnl"]) + e := "🟢" + if pnl < 0 { + e = "🔴" + } + sb.WriteString(fmt.Sprintf("%s *%s* %s\n Entry: $%.4f → $%.4f | P/L: $%.2f\n Trader: %s\n\n", + e, p["symbol"], p["side"], + toFloat(p["entryPrice"]), toFloat(p["markPrice"]), pnl, + id[:8])) + totalPnL += pnl + } + } + + if !hasPosition { + return a.msg(L, "no_positions"), nil + } + sb.WriteString(fmt.Sprintf(a.msg(L, "total_pnl"), totalPnL)) + return sb.String(), nil +} + +func (a *Agent) queryAllBalances(L string) (string, error) { + traders := a.traderManager.GetAllTraders() + if len(traders) == 0 { + return a.msg(L, "no_traders"), nil + } + + var sb strings.Builder + sb.WriteString(a.msg(L, "balance_header")) + + for id, t := range traders { + info, err := t.GetAccountInfo() + if err != nil { + continue + } + sb.WriteString(fmt.Sprintf("*%s* (%s)\n Total: $%.2f | Available: $%.2f | P/L: $%.2f\n\n", + t.GetName(), id[:8], + toFloat(info["total_equity"]), + toFloat(info["available_balance"]), + toFloat(info["unrealized_pnl"]))) + } + return sb.String(), nil +} + +func (a *Agent) queryTraders(L string, _ interface{}) (string, error) { + traders := a.traderManager.GetAllTraders() + if len(traders) == 0 { + return a.msg(L, "no_traders"), nil + } + + var sb strings.Builder + sb.WriteString(a.msg(L, "traders_header")) + + for id, t := range traders { + s := t.GetStatus() + running, _ := s["is_running"].(bool) + icon := "⏹" + if running { + icon = "▶️" + } + sb.WriteString(fmt.Sprintf("%s *%s* `%s`\n Exchange: %s | AI: %s\n\n", + icon, t.GetName(), id[:8], t.GetExchange(), t.GetAIModel())) + } + return sb.String(), nil +} + +func toFloat(v interface{}) float64 { + switch x := v.(type) { + case float64: + return x + case float32: + return float64(x) + case int: + return float64(x) + case int64: + return float64(x) + case string: + f, _ := strconv.ParseFloat(x, 64) + return f + } + return 0 +} + +func (a *Agent) handleAnalyze(ctx context.Context, L string, intent Intent) (string, error) { + symbol := "BTCUSDT" + if d, ok := intent.Params["detail"]; ok && d != "" { + symbol = strings.ToUpper(strings.TrimSpace(d)) + if !strings.HasSuffix(symbol, "USDT") { + symbol += "USDT" + } + } + + // Use NOFX's market data — full technical indicators + md, err := market.Get(symbol) + if err != nil { + a.logger.Error("get market data", "symbol", symbol, "error", err) + if L == "zh" { + return fmt.Sprintf("🔍 *%s 分析*\n\n⚠️ 市场数据暂不可用，请稍后再试。", strings.TrimSuffix(symbol, "USDT")), nil + } + return fmt.Sprintf("🔍 *%s Analysis*\n\n⚠️ Market data unavailable. Try again later.", strings.TrimSuffix(symbol, "USDT")), nil + } + + // Build rich analysis prompt with real data + prompt := buildAnalysisPrompt(symbol, md, L) + + // For now, format the raw market data as analysis + return fmt.Sprintf(a.msg(L, "analysis_header"), strings.TrimSuffix(symbol, "USDT")) + "\n\n" + prompt, nil +} + +func buildAnalysisPrompt(symbol string, md *market.Data, L string) string { + var sb strings.Builder + sb.WriteString(fmt.Sprintf("Analyze %s for trading.\n\n", symbol)) + sb.WriteString(fmt.Sprintf("Current Price: $%.4f\n", md.CurrentPrice)) + sb.WriteString(fmt.Sprintf("1h Change: %.2f%%\n", md.PriceChange1h)) + sb.WriteString(fmt.Sprintf("4h Change: %.2f%%\n", md.PriceChange4h)) + sb.WriteString(fmt.Sprintf("EMA20: %.4f\n", md.CurrentEMA20)) + sb.WriteString(fmt.Sprintf("MACD: %.4f\n", md.CurrentMACD)) + sb.WriteString(fmt.Sprintf("RSI7: %.2f\n", md.CurrentRSI7)) + sb.WriteString(fmt.Sprintf("Funding Rate: %.4f%%\n", md.FundingRate*100)) + if md.OpenInterest != nil { + sb.WriteString(fmt.Sprintf("OI: %.0f (avg: %.0f)\n", md.OpenInterest.Latest, md.OpenInterest.Average)) + } + if L == "zh" { + sb.WriteString("\n请用中文给出交易建议，包括方向、入场价、止损、止盈。简洁专业。") + } else { + sb.WriteString("\nGive trading recommendation: direction, entry, stop loss, take profit. Be concise and professional.") + } + return sb.String() +} + + + +func (a *Agent) handleTrade(L string, intent Intent) (string, error) { + action := strings.ToLower(intent.Params["action"]) + detail := intent.Params["detail"] + + traders := a.traderManager.GetAllTraders() + if len(traders) == 0 { + return a.msg(L, "no_traders"), nil + } + + // Parse symbol and quantity + parts := strings.Fields(detail) + if len(parts) < 1 { + return a.msg(L, "trade_usage"), nil + } + + symbol := strings.ToUpper(parts[0]) + if !strings.HasSuffix(symbol, "USDT") { + symbol += "USDT" + } + + qty := 0.0 + if len(parts) >= 2 { + q, err := strconv.ParseFloat(parts[1], 64) + if err != nil { + return fmt.Sprintf(a.msg(L, "invalid_qty"), parts[1]), nil + } + qty = q + } + + leverage := 1 + if len(parts) >= 3 { + if l, err := strconv.Atoi(strings.TrimSuffix(strings.ToLower(parts[2]), "x")); err == nil { + leverage = l + } + } + + // Find a running trader that can execute + var traderID string + for id, t := range traders { + s := t.GetStatus() + if running, ok := s["is_running"].(bool); ok && running { + traderID = id + break + } + } + if traderID == "" { + return a.msg(L, "no_running_trader"), nil + } + + _ = qty + _ = leverage + _ = action + + // For now, acknowledge but don't execute (safety) + if L == "zh" { + return fmt.Sprintf("⚡ *交易请求*\n\n• 操作: %s\n• 交易对: %s\n• 数量: %.6f\n• 杠杆: %dx\n• Trader: %s\n\n⚠️ 自动执行功能开发中。请在 Web UI 中操作。", + strings.ToUpper(action), symbol, qty, leverage, traderID[:8]), nil + } + return fmt.Sprintf("⚡ *Trade Request*\n\n• Action: %s\n• Symbol: %s\n• Qty: %.6f\n• Leverage: %dx\n• Trader: %s\n\n⚠️ Auto-execution coming soon. Please use Web UI.", + strings.ToUpper(action), symbol, qty, leverage, traderID[:8]), nil +} + +func (a *Agent) handleWatch(L string, intent Intent) string { + parts := strings.Fields(intent.Raw) + if len(parts) < 2 { + if a.sentinel == nil { + return a.msg(L, "sentinel_off") + } + return a.sentinel.FormatWatchlist(L) + } + + cmd := strings.ToLower(parts[0]) + symbol := strings.ToUpper(parts[1]) + if !strings.HasSuffix(symbol, "USDT") { + symbol += "USDT" + } + + if a.sentinel == nil { + return a.msg(L, "sentinel_off") + } + + switch cmd { + case "/watch": + a.sentinel.AddSymbol(symbol) + if L == "zh" { + return fmt.Sprintf("👁️ 已添加 *%s* 到监控列表", symbol) + } + return fmt.Sprintf("👁️ Now watching *%s*", symbol) + case "/unwatch": + a.sentinel.RemoveSymbol(symbol) + if L == "zh" { + return fmt.Sprintf("🚫 已移除 *%s*", symbol) + } + return fmt.Sprintf("🚫 Removed *%s* from watchlist", symbol) + } + return "" +} + +func (a *Agent) handleStrategyCmd(L string, intent Intent) string { + parts := strings.Fields(intent.Raw) + if len(parts) < 2 { + result, _ := a.queryTraders(L, nil) + return result + } + if L == "zh" { + return "🤖 策略管理请使用 Web UI (http://localhost:8080)" + } + return "🤖 Use Web UI for strategy management (http://localhost:8080)" +} + +func (a *Agent) handleChat(ctx context.Context, L string, userID int64, text string) (string, error) { + // TODO: integrate with NOFX's MCP client for AI chat + // For now, route to the existing Telegram agent or return guidance + if L == "zh" { + return "🤖 收到。AI 对话功能正在接入 NOFX 的 AI 引擎。\n\n你可以试试:\n• `/analyze BTC` — 市场分析\n• `/positions` — 查看持仓\n• `/status` — 系统状态", nil + } + return "🤖 Got it. AI chat is being integrated with NOFX's AI engine.\n\nTry:\n• `/analyze BTC` — market analysis\n• `/positions` — view positions\n• `/status` — system status", nil +} + +func (a *Agent) handleSignal(sig Signal) { + if a.brain != nil { + a.brain.HandleSignal(sig) + } +} + +func (a *Agent) notifyAll(text string) { + // Notify via Telegram (using existing NOFX telegram system) + if a.NotifyFunc != nil { + a.NotifyFunc(0, text) + } +} diff --git a/agent/brain.go b/agent/brain.go new file mode 100644 index 00000000..4325ddf3 --- /dev/null +++ b/agent/brain.go @@ -0,0 +1,146 @@ +package agent + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "sync" + "time" +) + +// Brain handles proactive intelligence: signals, news, market briefs. +type Brain struct { + agent *Agent + logger *slog.Logger + stopCh chan struct{} + recentSignals sync.Map // debounce +} + +func NewBrain(agent *Agent, logger *slog.Logger) *Brain { + return &Brain{agent: agent, logger: logger, stopCh: make(chan struct{})} +} + +func (b *Brain) Stop() { close(b.stopCh) } + +func (b *Brain) HandleSignal(sig Signal) { + key := fmt.Sprintf("%s:%s", sig.Type, sig.Symbol) + if v, ok := b.recentSignals.Load(key); ok { + if time.Since(v.(time.Time)) < 10*time.Minute { + return + } + } + b.recentSignals.Store(key, time.Now()) + + emoji := map[string]string{"info": "ℹ️", "warning": "⚠️", "critical": "🚨"} + e := emoji[sig.Severity] + if e == "" { e = "📊" } + + b.agent.notifyAll(fmt.Sprintf("%s *%s*\n\n%s", e, sig.Title, sig.Detail)) +} + +func (b *Brain) StartNewsScan(interval time.Duration) { + seen := make(map[string]bool) + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-b.stopCh: return + case <-ticker.C: + b.scanNews(seen) + } + } + }() +} + +func (b *Brain) scanNews(seen map[string]bool) { + resp, err := http.Get("https://min-api.cryptocompare.com/data/v2/news/?lang=EN&sortOrder=latest") + if err != nil { return } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + + var result struct { + Data []struct { + Title string `json:"title"` + Source string `json:"source"` + URL string `json:"url"` + Body string `json:"body"` + Categories string `json:"categories"` + PublishedOn int64 `json:"published_on"` + } `json:"Data"` + } + json.Unmarshal(body, &result) + + bullish := []string{"surge", "rally", "bullish", "breakout", "ath", "pump", "adoption"} + bearish := []string{"crash", "dump", "bearish", "sell-off", "plunge", "hack", "ban", "fraud"} + + for _, d := range result.Data { + if seen[d.URL] { continue } + seen[d.URL] = true + if time.Since(time.Unix(d.PublishedOn, 0)) > 10*time.Minute { continue } + + lower := strings.ToLower(d.Title + " " + d.Body) + bc, brc := 0, 0 + for _, w := range bullish { if strings.Contains(lower, w) { bc++ } } + for _, w := range bearish { if strings.Contains(lower, w) { brc++ } } + + if bc == 0 && brc == 0 { continue } + + emoji := "📰" + sentiment := "NEUTRAL" + if bc > brc { emoji = "🟢"; sentiment = "BULLISH" } + if brc > bc { emoji = "🔴"; sentiment = "BEARISH" } + + b.agent.notifyAll(fmt.Sprintf("%s *News*\n\n%s\n\n• Source: %s\n• Sentiment: %s", + emoji, d.Title, d.Source, sentiment)) + } + + if len(seen) > 1000 { for k := range seen { delete(seen, k) } } +} + +func (b *Brain) StartMarketBriefs(hours []int) { + go func() { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + sent := make(map[string]bool) + for { + select { + case <-b.stopCh: return + case now := <-ticker.C: + key := now.Format("2006-01-02-15") + for _, h := range hours { + if now.Hour() == h && now.Minute() == 30 && !sent[key] { + sent[key] = true + b.sendBrief(h) + } + } + } + } + }() +} + +func (b *Brain) sendBrief(hour int) { + title := "☀️ *早间市场简报*" + if hour >= 18 { title = "🌙 *晚间市场简报*" } + + // Fetch BTC/ETH prices for the brief + var btcPrice, ethPrice, btcChg, ethChg string + for _, sym := range []string{"BTCUSDT", "ETHUSDT"} { + resp, err := http.Get(fmt.Sprintf("https://fapi.binance.com/fapi/v1/ticker/24hr?symbol=%s", sym)) + if err != nil { continue } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + var t map[string]string + json.Unmarshal(body, &t) + if sym == "BTCUSDT" { btcPrice = t["lastPrice"]; btcChg = t["priceChangePercent"] } + if sym == "ETHUSDT" { ethPrice = t["lastPrice"]; ethChg = t["priceChangePercent"] } + } + + brief := fmt.Sprintf("%s\n\n• BTC: $%s (%s%%)\n• ETH: $%s (%s%%)\n\n_%s_", + title, btcPrice, btcChg, ethPrice, ethChg, time.Now().Format("2006-01-02 15:04")) + + b.agent.notifyAll(brief) +} diff --git a/agent/i18n.go b/agent/i18n.go new file mode 100644 index 00000000..d262112f --- /dev/null +++ b/agent/i18n.go @@ -0,0 +1,86 @@ +package agent + +var i18nMessages = map[string]map[string]string{ + "help": { + "zh": "🤖 *NOFXi — 你的 AI 交易 Agent*\n\n" + + "*交易:* /buy /sell /long /short + 交易对 数量 杠杆\n" + + "*查询:* /positions /balance /pnl /traders\n" + + "*分析:* /analyze BTC\n" + + "*监控:* /watch BTC · /unwatch BTC\n" + + "*策略:* /strategy\n" + + "*系统:* /status /help\n\n" + + "直接跟我说话就行，中英文都可以 💬", + "en": "🤖 *NOFXi — Your AI Trading Agent*\n\n" + + "*Trade:* /buy /sell /long /short + symbol qty leverage\n" + + "*Query:* /positions /balance /pnl /traders\n" + + "*Analyze:* /analyze BTC\n" + + "*Monitor:* /watch BTC · /unwatch BTC\n" + + "*Strategy:* /strategy\n" + + "*System:* /status /help\n\n" + + "Just talk to me in any language 💬", + }, + "status": { + "zh": "📊 *NOFXi 状态*\n\n• 运行中 Traders: %d/%d\n• 监控交易对: %d\n• 时间: %s", + "en": "📊 *NOFXi Status*\n\n• Running Traders: %d/%d\n• Watching: %d symbols\n• Time: %s", + }, + "no_traders": { + "zh": "📭 暂无 Trader。请在 Web UI 中创建和配置。", + "en": "📭 No traders configured. Create one in Web UI.", + }, + "no_running_trader": { + "zh": "⚠️ 没有运行中的 Trader。请在 Web UI 中启动。", + "en": "⚠️ No running trader. Start one in Web UI.", + }, + "no_positions": { + "zh": "📭 当前没有持仓。", + "en": "📭 No open positions.", + }, + "positions_header": { + "zh": "📊 *当前持仓*\n\n", + "en": "📊 *Open Positions*\n\n", + }, + "total_pnl": { + "zh": "💰 *总未实现盈亏: $%.2f*", + "en": "💰 *Total Unrealized P/L: $%.2f*", + }, + "balance_header": { + "zh": "💰 *账户余额*\n\n", + "en": "💰 *Account Balances*\n\n", + }, + "traders_header": { + "zh": "🤖 *Traders*\n\n", + "en": "🤖 *Traders*\n\n", + }, + "trade_usage": { + "zh": "用法: `/buy BTC 0.01` 或 `/sell ETH 0.5 3x`", + "en": "Usage: `/buy BTC 0.01` or `/sell ETH 0.5 3x`", + }, + "invalid_qty": { + "zh": "❓ 无效数量: %s", + "en": "❓ Invalid quantity: %s", + }, + "analysis_header": { + "zh": "🔍 *%s 市场分析*", + "en": "🔍 *%s Analysis*", + }, + "sentinel_off": { + "zh": "⚠️ Sentinel 未启用。", + "en": "⚠️ Sentinel not enabled.", + }, + "system_prompt": { + "zh": "你是 NOFXi，一个专业的 AI 交易 Agent。简洁、专业、用中文回复。使用交易相关 emoji。", + "en": "You are NOFXi, a professional AI trading agent. Be concise, professional. Use trading emojis.", + }, +} + +func (a *Agent) msg(lang, key string) string { + if m, ok := i18nMessages[key]; ok { + if s, ok := m[lang]; ok { + return s + } + if s, ok := m["en"]; ok { + return s + } + } + return key +} diff --git a/agent/router.go b/agent/router.go new file mode 100644 index 00000000..e463cec3 --- /dev/null +++ b/agent/router.go @@ -0,0 +1,102 @@ +package agent + +import ( + "regexp" + "strings" +) + +type IntentType int + +const ( + IntentChat IntentType = iota + IntentTrade + IntentQuery + IntentAnalyze + IntentSettings + IntentHelp + IntentStatus + IntentWatch + IntentStrategy +) + +type Intent struct { + Type IntentType + Params map[string]string + Raw string +} + +type Router struct { + tradePatterns []*regexp.Regexp + queryPatterns []*regexp.Regexp + analyzePatterns []*regexp.Regexp +} + +func NewRouter() *Router { + return &Router{ + tradePatterns: []*regexp.Regexp{ + regexp.MustCompile(`(?i)(buy|sell|long|short|open|close)\s+(.+)`), + regexp.MustCompile(`(?i)(做多|做空|买入|卖出|开仓|平仓)\s*(.+)`), + }, + queryPatterns: []*regexp.Regexp{ + regexp.MustCompile(`(?i)(position|balance|pnl|profit|loss|持仓|余额|盈亏|trader|交易员)`), + }, + analyzePatterns: []*regexp.Regexp{ + regexp.MustCompile(`(?i)(analyze|analysis|分析|看看)\s+(.+)`), + regexp.MustCompile(`(?i)(what.*think|怎么看|你觉得)\s*(.+)?`), + }, + } +} + +func (r *Router) Route(text string) Intent { + text = strings.TrimSpace(text) + + if strings.HasPrefix(text, "/") { + return r.routeCommand(text) + } + + for _, p := range r.tradePatterns { + if m := p.FindStringSubmatch(text); m != nil { + return Intent{Type: IntentTrade, Params: map[string]string{"action": strings.ToLower(m[1]), "detail": m[2]}, Raw: text} + } + } + for _, p := range r.queryPatterns { + if p.MatchString(text) { + return Intent{Type: IntentQuery, Raw: text} + } + } + for _, p := range r.analyzePatterns { + if m := p.FindStringSubmatch(text); m != nil { + d := "" + if len(m) > 2 { d = m[2] } + return Intent{Type: IntentAnalyze, Params: map[string]string{"detail": d}, Raw: text} + } + } + + return Intent{Type: IntentChat, Raw: text} +} + +func (r *Router) routeCommand(text string) Intent { + cmd := strings.ToLower(strings.Fields(text)[0]) + parts := strings.SplitN(text, " ", 2) + detail := "" + if len(parts) > 1 { detail = parts[1] } + + switch cmd { + case "/start", "/help": + return Intent{Type: IntentHelp, Raw: text} + case "/status": + return Intent{Type: IntentStatus, Raw: text} + case "/buy", "/sell", "/long", "/short", "/open", "/close": + return Intent{Type: IntentTrade, Params: map[string]string{"action": strings.TrimPrefix(cmd, "/"), "detail": detail}, Raw: text} + case "/positions", "/balance", "/pnl", "/traders": + return Intent{Type: IntentQuery, Raw: text} + case "/analyze": + return Intent{Type: IntentAnalyze, Params: map[string]string{"detail": detail}, Raw: text} + case "/watch", "/unwatch": + return Intent{Type: IntentWatch, Raw: text} + case "/strategy": + return Intent{Type: IntentStrategy, Raw: text} + default: + return Intent{Type: IntentChat, Raw: text} + } +} diff --git a/agent/scheduler.go b/agent/scheduler.go new file mode 100644 index 00000000..c5570aea --- /dev/null +++ b/agent/scheduler.go @@ -0,0 +1,99 @@ +package agent + +import ( + "context" + "fmt" + "log/slog" + "strings" + "time" +) + +type Scheduler struct { + agent *Agent + logger *slog.Logger + stopCh chan struct{} +} + +func NewScheduler(a *Agent, l *slog.Logger) *Scheduler { + return &Scheduler{agent: a, logger: l, stopCh: make(chan struct{})} +} + +func (s *Scheduler) Start(ctx context.Context) { + go func() { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + lastReport := time.Time{} + lastCheck := time.Time{} + + for { + select { + case <-ctx.Done(): return + case <-s.stopCh: return + case now := <-ticker.C: + // Daily report at 21:00 + if now.Hour() == 21 && now.Sub(lastReport) > 12*time.Hour { + s.dailyReport() + lastReport = now + } + // Position risk check every 4h + if now.Sub(lastCheck) > 4*time.Hour { + s.riskCheck() + lastCheck = now + } + } + } + }() +} + +func (s *Scheduler) Stop() { close(s.stopCh) } + +func (s *Scheduler) dailyReport() { + if s.agent.traderManager == nil { return } + + traders := s.agent.traderManager.GetAllTraders() + if len(traders) == 0 { return } + + var sb strings.Builder + sb.WriteString(fmt.Sprintf("📊 *NOFXi 每日报告 — %s*\n\n", time.Now().Format("2006-01-02"))) + + totalPnL := 0.0 + for _, t := range traders { + info, err := t.GetAccountInfo() + if err != nil { continue } + equity := toFloat(info["total_equity"]) + pnl := toFloat(info["unrealized_pnl"]) + sb.WriteString(fmt.Sprintf("• %s: $%.2f (P/L: $%.2f)\n", t.GetName(), equity, pnl)) + totalPnL += pnl + } + e := "📈" + if totalPnL < 0 { e = "📉" } + sb.WriteString(fmt.Sprintf("\n%s Total P/L: $%.2f", e, totalPnL)) + + s.agent.notifyAll(sb.String()) +} + +func (s *Scheduler) riskCheck() { + if s.agent.traderManager == nil { return } + + var alerts []string + for _, t := range s.agent.traderManager.GetAllTraders() { + positions, err := t.GetPositions() + if err != nil { continue } + for _, p := range positions { + pnl := toFloat(p["unrealizedPnl"]) + size := toFloat(p["size"]) + if size == 0 { continue } + entry := toFloat(p["entryPrice"]) + if entry > 0 { + pnlPct := (pnl / (entry * size)) * 100 + if pnlPct < -5 { + alerts = append(alerts, fmt.Sprintf("⚠️ *%s* %s: %.1f%% ($%.2f)", + p["symbol"], p["side"], pnlPct, pnl)) + } + } + } + } + if len(alerts) > 0 { + s.agent.notifyAll("🚨 *持仓风险提醒*\n\n" + strings.Join(alerts, "\n")) + } +} diff --git a/agent/sentinel.go b/agent/sentinel.go new file mode 100644 index 00000000..5397696e --- /dev/null +++ b/agent/sentinel.go @@ -0,0 +1,167 @@ +package agent + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "math" + "net/http" + "strconv" + "strings" + "sync" + "time" +) + +type SignalType string + +const ( + SignalPriceBreakout SignalType = "price_breakout" + SignalVolumeSpike SignalType = "volume_spike" + SignalFundingRate SignalType = "funding_rate" +) + +type Signal struct { + Type SignalType + Symbol string + Severity string + Title string + Detail string + Price float64 + Change float64 +} + +type SignalCallback func(Signal) + +type Sentinel struct { + mu sync.RWMutex + symbols []string + history map[string][]pricePt + onSignal SignalCallback + http *http.Client + logger *slog.Logger + stopCh chan struct{} +} + +type pricePt struct { + Price float64 + Volume float64 + Time time.Time +} + +func NewSentinel(symbols []string, cb SignalCallback, logger *slog.Logger) *Sentinel { + return &Sentinel{ + symbols: symbols, + history: make(map[string][]pricePt), + onSignal: cb, + http: &http.Client{Timeout: 10 * time.Second}, + logger: logger, + stopCh: make(chan struct{}), + } +} + +func (s *Sentinel) Start() { + go func() { + ticker := time.NewTicker(60 * time.Second) + defer ticker.Stop() + s.scan() + for { + select { + case <-s.stopCh: + return + case <-ticker.C: + s.scan() + } + } + }() +} + +func (s *Sentinel) Stop() { close(s.stopCh) } +func (s *Sentinel) SymbolCount() int { s.mu.RLock(); defer s.mu.RUnlock(); return len(s.symbols) } +func (s *Sentinel) AddSymbol(sym string) { s.mu.Lock(); defer s.mu.Unlock(); for _, x := range s.symbols { if x == sym { return } }; s.symbols = append(s.symbols, sym) } +func (s *Sentinel) RemoveSymbol(sym string) { s.mu.Lock(); defer s.mu.Unlock(); for i, x := range s.symbols { if x == sym { s.symbols = append(s.symbols[:i], s.symbols[i+1:]...); return } } } + +func (s *Sentinel) FormatWatchlist(L string) string { + s.mu.RLock() + defer s.mu.RUnlock() + if len(s.symbols) == 0 { + if L == "zh" { return "📭 监控列表为空。用 `/watch BTC` 添加。" } + return "📭 Watchlist empty. Use `/watch BTC` to add." + } + var sb strings.Builder + if L == "zh" { sb.WriteString("👁️ *监控列表*\n\n") } else { sb.WriteString("👁️ *Watchlist*\n\n") } + for _, sym := range s.symbols { + if pts, ok := s.history[sym]; ok && len(pts) > 0 { + last := pts[len(pts)-1] + sb.WriteString(fmt.Sprintf("• *%s*: $%.4f (%s)\n", sym, last.Price, last.Time.Format("15:04"))) + } else { + sb.WriteString(fmt.Sprintf("• *%s*: waiting...\n", sym)) + } + } + return sb.String() +} + +func (s *Sentinel) scan() { + s.mu.RLock() + syms := make([]string, len(s.symbols)) + copy(syms, s.symbols) + s.mu.RUnlock() + for _, sym := range syms { + s.check(sym) + } +} + +func (s *Sentinel) check(symbol string) { + resp, err := s.http.Get(fmt.Sprintf("https://fapi.binance.com/fapi/v1/ticker/24hr?symbol=%s", symbol)) + if err != nil { return } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + var t map[string]interface{} + json.Unmarshal(body, &t) + + price, _ := strconv.ParseFloat(fmt.Sprint(t["lastPrice"]), 64) + vol, _ := strconv.ParseFloat(fmt.Sprint(t["quoteVolume"]), 64) + chg, _ := strconv.ParseFloat(fmt.Sprint(t["priceChangePercent"]), 64) + + pt := pricePt{Price: price, Volume: vol, Time: time.Now()} + s.mu.Lock() + h := s.history[symbol] + h = append(h, pt) + if len(h) > 60 { h = h[len(h)-60:] } + s.history[symbol] = h + s.mu.Unlock() + + if len(h) < 5 { return } + + // Price breakout (>3% in 5 min) + old := h[len(h)-5] + pct := ((price - old.Price) / old.Price) * 100 + if math.Abs(pct) >= 3.0 { + sev := "warning" + if math.Abs(pct) >= 6.0 { sev = "critical" } + dir := "📈 拉升" + if pct < 0 { dir = "📉 下跌" } + s.emit(Signal{Type: SignalPriceBreakout, Symbol: symbol, Severity: sev, + Title: fmt.Sprintf("%s %s %.1f%%", symbol, dir, math.Abs(pct)), + Detail: fmt.Sprintf("5min: $%.2f → $%.2f (24h: %.1f%%)", old.Price, price, chg), + Price: price, Change: pct}) + } + + // Volume spike (>3x avg) + if len(h) >= 10 { + var avg float64 + for i := 0; i < len(h)-1; i++ { avg += h[i].Volume } + avg /= float64(len(h) - 1) + if avg > 0 && vol > avg*3 { + s.emit(Signal{Type: SignalVolumeSpike, Symbol: symbol, Severity: "warning", + Title: fmt.Sprintf("%s 成交量异常 %.1fx", symbol, vol/avg), + Detail: fmt.Sprintf("Price: $%.2f (24h: %.1f%%)", price, chg), + Price: price, Change: chg}) + } + } +} + +func (s *Sentinel) emit(sig Signal) { + s.logger.Info("signal", "type", sig.Type, "symbol", sig.Symbol, "title", sig.Title) + if s.onSignal != nil { s.onSignal(sig) } +} diff --git a/agent/web.go b/agent/web.go new file mode 100644 index 00000000..af6b853d --- /dev/null +++ b/agent/web.go @@ -0,0 +1,141 @@ +package agent + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "time" +) + +// WebHandler provides HTTP endpoints for the NOFXi agent. +// These are registered on the existing NOFX API server. +type WebHandler struct { + agent *Agent + logger *slog.Logger +} + +// NewWebHandler creates a new web handler. +func NewWebHandler(agent *Agent, logger *slog.Logger) *WebHandler { + return &WebHandler{agent: agent, logger: logger} +} + +// RegisterRoutes registers agent API routes on an existing mux or gin router. +// For now we use a standalone http server on a separate port. +func (w *WebHandler) StartStandalone(port int) { + mux := http.NewServeMux() + + // Health + mux.HandleFunc("/api/agent/health", func(rw http.ResponseWriter, r *http.Request) { + writeJSON(rw, 200, map[string]string{"status": "ok", "agent": "NOFXi", "time": time.Now().Format(time.RFC3339)}) + }) + + // Chat endpoint + mux.HandleFunc("/api/agent/chat", func(rw http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(rw, "method not allowed", 405) + return + } + var req struct { + Message string `json:"message"` + UserID int64 `json:"user_id"` + Lang string `json:"lang"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(rw, 400, map[string]string{"error": "invalid request"}) + return + } + if req.Message == "" { + writeJSON(rw, 400, map[string]string{"error": "message required"}) + return + } + if req.UserID == 0 { + req.UserID = 1 + } + + msg := req.Message + if req.Lang != "" { + msg = "[lang:" + req.Lang + "] " + msg + } + + ctx, cancel := context.WithTimeout(r.Context(), 55*time.Second) + defer cancel() + + resp, err := w.agent.HandleMessage(ctx, req.UserID, msg) + if err != nil { + writeJSON(rw, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(rw, 200, map[string]string{"response": resp}) + }) + + // Kline data + mux.HandleFunc("/api/agent/klines", handleKlines) + + // Ticker + mux.HandleFunc("/api/agent/ticker", handleTicker) + + go func() { + addr := fmt.Sprintf(":%d", port) + w.logger.Info("NOFXi agent web API starting", "port", port) + if err := http.ListenAndServe(addr, mux); err != nil { + w.logger.Error("agent web server error", "error", err) + } + }() +} + +func writeJSON(w http.ResponseWriter, status int, v interface{}) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Access-Control-Allow-Origin", "*") + w.WriteHeader(status) + json.NewEncoder(w).Encode(v) +} + +func handleKlines(w http.ResponseWriter, r *http.Request) { + symbol := r.URL.Query().Get("symbol") + if symbol == "" { symbol = "BTCUSDT" } + interval := r.URL.Query().Get("interval") + if interval == "" { interval = "1h" } + + url := fmt.Sprintf("https://fapi.binance.com/fapi/v1/klines?symbol=%s&interval=%s&limit=300", symbol, interval) + client := &http.Client{Timeout: 10 * time.Second} + resp, err := client.Get(url) + if err != nil { + writeJSON(w, 502, map[string]string{"error": err.Error()}) + return + } + defer resp.Body.Close() + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Access-Control-Allow-Origin", "*") + w.WriteHeader(200) + buf := make([]byte, 32768) + for { + n, err := resp.Body.Read(buf) + if n > 0 { w.Write(buf[:n]) } + if err != nil { break } + } +} + +func handleTicker(w http.ResponseWriter, r *http.Request) { + symbol := r.URL.Query().Get("symbol") + if symbol == "" { symbol = "BTCUSDT" } + + url := fmt.Sprintf("https://fapi.binance.com/fapi/v1/ticker/24hr?symbol=%s", symbol) + client := &http.Client{Timeout: 10 * time.Second} + resp, err := client.Get(url) + if err != nil { + writeJSON(w, 502, map[string]string{"error": err.Error()}) + return + } + defer resp.Body.Close() + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Access-Control-Allow-Origin", "*") + w.WriteHeader(200) + buf := make([]byte, 4096) + for { + n, err := resp.Body.Read(buf) + if n > 0 { w.Write(buf[:n]) } + if err != nil { break } + } +} diff --git a/main.go b/main.go index 7fa7295c..8ed526b2 100644 --- a/main.go +++ b/main.go @@ -1,7 +1,9 @@ package main import ( + "log/slog" "nofx/api" + nofxiagent "nofx/agent" "nofx/auth" "nofx/config" "nofx/crypto" @@ -29,7 +31,8 @@ func main() { logger.Init(nil) logger.Info("╔════════════════════════════════════════════════════════════╗") - logger.Info("║ 🚀 NOFX - AI-Powered Trading System ║") + logger.Info("║ 🚀 NOFXi - AI Trading Agent ║") + logger.Info("║ Powered by NOFX Engine ║") logger.Info("╚════════════════════════════════════════════════════════════╝") // Initialize global configuration (loaded from .env) @@ -140,6 +143,17 @@ func main() { // Start Telegram bot (if TELEGRAM_BOT_TOKEN is configured) go telegram.Start(cfg, st, telegramReloadCh) + // Start NOFXi Agent (proactive intelligence layer) + nofxiAgent := nofxiagent.New(traderManager, st, nil, slog.Default()) + nofxiAgent.Start() + defer nofxiAgent.Stop() + + // Start NOFXi Agent Web API (port 8900) + agentWeb := nofxiagent.NewWebHandler(nofxiAgent, slog.Default()) + agentWeb.StartStandalone(8900) + + logger.Info("🧠 NOFXi Agent started (sentinel + brain + scheduler + web:8900)") + // Wait for interrupt signal quit := make(chan os.Signal, 1) signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) diff --git a/web/nofxi.html b/web/nofxi.html new file mode 100644 index 00000000..a34b1fb6 --- /dev/null +++ b/web/nofxi.html @@ -0,0 +1,418 @@ + + + + + + NOFXi — AI Trading Agent + + + + + + + +
+
+ +
+ + + +
+
+
+
--
+ +
+
+ + +
+ +
+
+
+
Analysis
+
+ + + + +
+
+
+
Portfolio
+
+ + + +
+
+
+
Monitor
+
+ + +
+
+
+
Strategy
+
+ + +
+
+
+ + +
+
+
Thinking...
+
+
+ + +
+
Enter to send · /help, /analyze BTC, /buy ETH 0.1
+
+
+ + +
+
+
+
+ +
+ -- + -- +
+
+ + + + + + +
+
+
+
+
+ + + +