mirror of
https://github.com/NoFxAiOS/nofx.git
synced 2026-07-07 13:00:59 +08:00
Applied safe.GoNamed to: - 9 exchange order_sync goroutines (OKX, Hyperliquid, Aster, Bybit, KuCoin, Gate, Bitget, Lighter, Binance×2) - Drawdown monitor (auto_trader_risk.go) - Brain news scanner + market briefs (agent/brain.go) - Sentinel scanner (agent/sentinel.go) - Agent scheduler (agent/scheduler.go) - x402 idle watchdog (mcp/payment/x402.go) - MCP stream idle watchdog (mcp/client.go) - Rate limiter cleanup (api/rate_limiter.go) - 3 telemetry fire-and-forget sends (telemetry/experience.go) - CoinAnk WS handler (provider/coinank/coinank_api/kline_ws.go) - API server goroutine (main.go) Added manual defer/recover with error reporting to: - Telegram AI agent handler (sends error msg to user on panic) - Trader data fetch (returns error result on panic to prevent deadlock) Before: a panic in ANY of these 27 goroutines would crash the entire trading process with zero diagnostics. Now all panics are caught, logged with stack traces, and the process continues running.
109 lines
2.6 KiB
Go
109 lines
2.6 KiB
Go
package agent
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"nofx/safe"
|
|
"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) {
|
|
safe.GoNamed("agent-scheduler", func() {
|
|
ticker := time.NewTicker(1 * time.Minute)
|
|
defer ticker.Stop()
|
|
lastReport := time.Time{}
|
|
lastCheck := time.Time{}
|
|
lastCleanup := 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
|
|
}
|
|
// Clean stale chat history every hour (sessions idle > 24h)
|
|
if now.Sub(lastCleanup) > 1*time.Hour {
|
|
if s.agent.history != nil {
|
|
s.agent.history.CleanOld(24 * time.Hour)
|
|
}
|
|
lastCleanup = 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"))
|
|
}
|
|
}
|