mirror of
https://github.com/NoFxAiOS/nofx.git
synced 2026-07-11 06:46:59 +08:00
7302f96e8e
流程为下： 1. 启动时使用所有交易员设置的币种(去重) 如果交易员未配置，则使用系统默认 2. 在决策获取K线时 如果没有缓存 则先实时获取后再添加订阅. ps: 适用于Api方式的币种列表
521 lines
14 KiB
Go
521 lines
14 KiB
Go
package market
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"math"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
type WSMonitor struct {
|
||
wsClient *WSClient
|
||
combinedClient *CombinedStreamsClient
|
||
featureEngine *FeatureEngine
|
||
symbols []string
|
||
featuresMap sync.Map
|
||
alertsChan chan Alert
|
||
klineDataMap3m sync.Map // 存储每个交易对的K线历史数据
|
||
klineDataMap4h sync.Map // 存储每个交易对的K线历史数据
|
||
tickerDataMap sync.Map // 存储每个交易对的ticker数据
|
||
batchSize int
|
||
filterSymbols sync.Map // 使用sync.Map来存储需要监控的币种和其状态
|
||
symbolStats sync.Map // 存储币种统计信息
|
||
FilterSymbol []string //经过筛选的币种
|
||
}
|
||
type SymbolStats struct {
|
||
LastActiveTime time.Time
|
||
AlertCount int
|
||
VolumeSpikeCount int
|
||
LastAlertTime time.Time
|
||
Score float64 // 综合评分
|
||
}
|
||
|
||
var WSMonitorCli *WSMonitor
|
||
var subKlineTime = []string{"3m", "4h"} // 管理订阅流的K线周期
|
||
|
||
func NewWSMonitor(batchSize int) *WSMonitor {
|
||
WSMonitorCli = &WSMonitor{
|
||
wsClient: NewWSClient(),
|
||
combinedClient: NewCombinedStreamsClient(batchSize),
|
||
featureEngine: NewFeatureEngine(config.AlertThresholds),
|
||
alertsChan: make(chan Alert, 1000),
|
||
batchSize: batchSize,
|
||
}
|
||
return WSMonitorCli
|
||
}
|
||
|
||
func (m *WSMonitor) Initialize(coins []string) error {
|
||
log.Println("初始化WebSocket监控器...")
|
||
// 获取交易对信息
|
||
apiClient := NewAPIClient()
|
||
// 如果不指定交易对，则使用market市场的所有交易对币种
|
||
if len(coins) == 0 {
|
||
exchangeInfo, err := apiClient.GetExchangeInfo()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// 筛选永续合约交易对 --仅测试时使用
|
||
//exchangeInfo.Symbols = exchangeInfo.Symbols[0:2]
|
||
for _, symbol := range exchangeInfo.Symbols {
|
||
if symbol.Status == "TRADING" && symbol.ContractType == "PERPETUAL" && strings.ToUpper(symbol.Symbol[len(symbol.Symbol)-4:]) == "USDT" {
|
||
m.symbols = append(m.symbols, symbol.Symbol)
|
||
}
|
||
}
|
||
} else {
|
||
m.symbols = coins
|
||
}
|
||
|
||
log.Printf("找到 %d 个交易对", len(m.symbols))
|
||
// 初始化历史数据
|
||
if err := m.initializeHistoricalData(); err != nil {
|
||
log.Printf("初始化历史数据失败: %v", err)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (m *WSMonitor) initializeHistoricalData() error {
|
||
apiClient := NewAPIClient()
|
||
|
||
var wg sync.WaitGroup
|
||
semaphore := make(chan struct{}, 5) // 限制并发数
|
||
|
||
for _, symbol := range m.symbols {
|
||
wg.Add(1)
|
||
semaphore <- struct{}{}
|
||
|
||
go func(s string) {
|
||
defer wg.Done()
|
||
defer func() { <-semaphore }()
|
||
|
||
// 获取历史K线数据
|
||
klines, err := apiClient.GetKlines(s, "3m", 100)
|
||
if err != nil {
|
||
log.Printf("获取 %s 历史数据失败: %v", s, err)
|
||
return
|
||
}
|
||
if len(klines) > 0 {
|
||
m.klineDataMap3m.Store(s, klines)
|
||
log.Printf("已加载 %s 的历史K线数据-3m: %d 条", s, len(klines))
|
||
}
|
||
// 获取历史K线数据
|
||
klines4h, err := apiClient.GetKlines(s, "4h", 100)
|
||
if err != nil {
|
||
log.Printf("获取 %s 历史数据失败: %v", s, err)
|
||
return
|
||
}
|
||
if len(klines4h) > 0 {
|
||
m.klineDataMap4h.Store(s, klines)
|
||
log.Printf("已加载 %s 的历史K线数据-4h: %d 条", s, len(klines))
|
||
}
|
||
}(symbol)
|
||
}
|
||
|
||
wg.Wait()
|
||
return nil
|
||
}
|
||
|
||
func (m *WSMonitor) Start(coins []string) {
|
||
log.Printf("启动WebSocket实时监控...")
|
||
// 初始化交易对
|
||
err := m.Initialize(coins)
|
||
if err != nil {
|
||
log.Fatalf("❌ 初始化币种: %v", err)
|
||
return
|
||
}
|
||
|
||
err = m.combinedClient.Connect()
|
||
if err != nil {
|
||
log.Fatalf("❌ 批量订阅流: %v", err)
|
||
return
|
||
}
|
||
// 启动警报处理器
|
||
//go m.handleAlerts()
|
||
// 启动定期清理任务
|
||
//go m.cleanupInactiveSymbols()
|
||
// 输出监控统计 - 评分前十名
|
||
//go m.printFilterStats(20)
|
||
// 订阅所有交易对
|
||
err = m.subscribeAll()
|
||
if err != nil {
|
||
log.Fatalf("❌ 订阅币种交易对: %v", err)
|
||
return
|
||
}
|
||
}
|
||
|
||
// subscribeSymbol 注册监听
|
||
func (m *WSMonitor) subscribeSymbol(symbol, st string) []string {
|
||
var streams []string
|
||
stream := fmt.Sprintf("%s@kline_%s", strings.ToLower(symbol), st)
|
||
ch := m.combinedClient.AddSubscriber(stream, 100)
|
||
streams = append(streams, stream)
|
||
go m.handleKlineData(symbol, ch, st)
|
||
|
||
return streams
|
||
}
|
||
func (m *WSMonitor) subscribeAll() error {
|
||
// 执行批量订阅
|
||
log.Println("开始订阅所有交易对...")
|
||
for _, symbol := range m.symbols {
|
||
for _, st := range subKlineTime {
|
||
m.subscribeSymbol(symbol, st)
|
||
}
|
||
}
|
||
for _, st := range subKlineTime {
|
||
err := m.combinedClient.BatchSubscribeKlines(m.symbols, st)
|
||
if err != nil {
|
||
log.Fatalf("❌ 订阅3m K线: %v", err)
|
||
return err
|
||
}
|
||
}
|
||
log.Println("所有交易对订阅完成")
|
||
return nil
|
||
}
|
||
|
||
func (m *WSMonitor) handleKlineData(symbol string, ch <-chan []byte, _time string) {
|
||
for data := range ch {
|
||
var klineData KlineWSData
|
||
if err := json.Unmarshal(data, &klineData); err != nil {
|
||
log.Printf("解析Kline数据失败: %v", err)
|
||
continue
|
||
}
|
||
m.processKlineUpdate(symbol, klineData, _time)
|
||
}
|
||
}
|
||
|
||
func (m *WSMonitor) getKlineDataMap(_time string) *sync.Map {
|
||
var klineDataMap *sync.Map
|
||
if _time == "3m" {
|
||
klineDataMap = &m.klineDataMap3m
|
||
} else if _time == "4h" {
|
||
klineDataMap = &m.klineDataMap4h
|
||
} else {
|
||
klineDataMap = &sync.Map{}
|
||
}
|
||
return klineDataMap
|
||
}
|
||
func (m *WSMonitor) processKlineUpdate(symbol string, wsData KlineWSData, _time string) {
|
||
// 转换WebSocket数据为Kline结构
|
||
kline := Kline{
|
||
OpenTime: wsData.Kline.StartTime,
|
||
CloseTime: wsData.Kline.CloseTime,
|
||
Trades: wsData.Kline.NumberOfTrades,
|
||
}
|
||
kline.Open, _ = parseFloat(wsData.Kline.OpenPrice)
|
||
kline.High, _ = parseFloat(wsData.Kline.HighPrice)
|
||
kline.Low, _ = parseFloat(wsData.Kline.LowPrice)
|
||
kline.Close, _ = parseFloat(wsData.Kline.ClosePrice)
|
||
kline.Volume, _ = parseFloat(wsData.Kline.Volume)
|
||
kline.High, _ = parseFloat(wsData.Kline.HighPrice)
|
||
kline.QuoteVolume, _ = parseFloat(wsData.Kline.QuoteVolume)
|
||
kline.TakerBuyBaseVolume, _ = parseFloat(wsData.Kline.TakerBuyBaseVolume)
|
||
kline.TakerBuyQuoteVolume, _ = parseFloat(wsData.Kline.TakerBuyQuoteVolume)
|
||
// 更新K线数据
|
||
var klineDataMap = m.getKlineDataMap(_time)
|
||
value, exists := klineDataMap.Load(symbol)
|
||
var klines []Kline
|
||
if exists {
|
||
klines = value.([]Kline)
|
||
|
||
// 检查是否是新的K线
|
||
if len(klines) > 0 && klines[len(klines)-1].OpenTime == kline.OpenTime {
|
||
// 更新当前K线
|
||
klines[len(klines)-1] = kline
|
||
} else {
|
||
// 添加新K线
|
||
klines = append(klines, kline)
|
||
|
||
// 保持数据长度
|
||
if len(klines) > 100 {
|
||
klines = klines[1:]
|
||
}
|
||
}
|
||
} else {
|
||
klines = []Kline{kline}
|
||
}
|
||
|
||
klineDataMap.Store(symbol, klines)
|
||
// 计算特征并检测警报
|
||
if len(klines) >= 20 {
|
||
features := m.featureEngine.CalculateFeatures(symbol, klines)
|
||
if features != nil {
|
||
m.featuresMap.Store(symbol, features)
|
||
|
||
alerts := m.featureEngine.DetectAlerts(features)
|
||
hasAlert := len(alerts) > 0
|
||
|
||
// 更新统计信息
|
||
m.updateSymbolStats(symbol, features, hasAlert)
|
||
|
||
for _, alert := range alerts {
|
||
m.alertsChan <- alert
|
||
}
|
||
|
||
// 实时日志输出重要特征
|
||
if len(alerts) > 0 || features.VolumeRatio5 > 2.0 || math.Abs(features.PriceChange15Min) > 0.02 {
|
||
//log.Printf("📊 %s - 价格: %.4f, 15分钟变动: %.2f%%, 交易量倍数: %.2f, RSI: %.1f",
|
||
// symbol, features.Price, features.PriceChange15Min*100,
|
||
// features.VolumeRatio5, features.RSI14)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (m *WSMonitor) processTickerUpdate(symbol string, tickerData TickerWSData) {
|
||
// 存储ticker数据
|
||
m.tickerDataMap.Store(symbol, tickerData)
|
||
}
|
||
|
||
func (m *WSMonitor) handleAlerts() {
|
||
alertCounts := make(map[string]int)
|
||
lastReset := time.Now()
|
||
|
||
for alert := range m.alertsChan {
|
||
// 重置计数器（每小时）
|
||
if time.Since(lastReset) > time.Hour {
|
||
alertCounts = make(map[string]int)
|
||
lastReset = time.Now()
|
||
}
|
||
|
||
// 警报去重和频率控制
|
||
alertKey := fmt.Sprintf("%s_%s", alert.Symbol, alert.Type)
|
||
alertCounts[alertKey]++
|
||
m.filterSymbols.Store(alert.Symbol, true)
|
||
|
||
//log.Printf("✅ 自动添加监控: %s (因警报: %s)", alert.Symbol, alert.Message)
|
||
if alertCounts[alertKey] <= 3 { // 每小时最多3次相同警报
|
||
//log.Printf("🚨 实时警报: %s", alert.Message)
|
||
|
||
// 这里可以添加其他警报处理逻辑
|
||
}
|
||
}
|
||
}
|
||
|
||
func (m *WSMonitor) GetCurrentKlines(symbol string, _time string) ([]Kline, error) {
|
||
// 对每一个进来的symbol检测是否存在内类 是否的话就订阅它
|
||
value, exists := m.getKlineDataMap(_time).Load(symbol)
|
||
if !exists {
|
||
// 如果Ws数据未初始化完成时,单独使用api获取 - 兼容性代码 (防止在未初始化完成是,已经有交易员运行)
|
||
apiClient := NewAPIClient()
|
||
klines, err := apiClient.GetKlines(symbol, _time, 100)
|
||
m.getKlineDataMap(_time).Store(strings.ToUpper(symbol), klines) //动态缓存进缓存
|
||
subStr := m.subscribeSymbol(symbol, _time)
|
||
subErr := m.combinedClient.subscribeStreams(subStr)
|
||
log.Printf("动态订阅流: %v", subStr)
|
||
if subErr != nil {
|
||
return nil, fmt.Errorf("动态订阅%v分钟K线失败: %v", _time, subErr)
|
||
}
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取%v分钟K线失败: %v", _time, err)
|
||
}
|
||
return klines, fmt.Errorf("symbol不存在")
|
||
}
|
||
return value.([]Kline), nil
|
||
}
|
||
|
||
func (m *WSMonitor) GetCurrentFeatures(symbol string) (*SymbolFeatures, bool) {
|
||
value, exists := m.featuresMap.Load(symbol)
|
||
if !exists {
|
||
return nil, false
|
||
}
|
||
return value.(*SymbolFeatures), true
|
||
}
|
||
|
||
func (m *WSMonitor) GetAllFeatures() map[string]*SymbolFeatures {
|
||
features := make(map[string]*SymbolFeatures)
|
||
m.featuresMap.Range(func(key, value interface{}) bool {
|
||
features[key.(string)] = value.(*SymbolFeatures)
|
||
return true
|
||
})
|
||
return features
|
||
}
|
||
|
||
func (m *WSMonitor) Close() {
|
||
m.wsClient.Close()
|
||
close(m.alertsChan)
|
||
}
|
||
func (m *WSMonitor) printFilterStats(nember int) {
|
||
ticker := time.NewTicker(2 * time.Minute)
|
||
defer ticker.Stop()
|
||
|
||
for range ticker.C {
|
||
var monitoredSymbols []string
|
||
m.filterSymbols.Range(func(key, value interface{}) bool {
|
||
monitoredSymbols = append(monitoredSymbols, key.(string))
|
||
return true
|
||
})
|
||
|
||
log.Printf("🎯 监控统计 - 总数: %d, 币种: %v",
|
||
len(monitoredSymbols), monitoredSymbols)
|
||
|
||
// 打印前5个评分最高的币种
|
||
type symbolScore struct {
|
||
symbol string
|
||
score float64
|
||
}
|
||
var topScores []symbolScore
|
||
|
||
m.symbolStats.Range(func(key, value interface{}) bool {
|
||
symbol := key.(string)
|
||
stats := value.(*SymbolStats)
|
||
topScores = append(topScores, symbolScore{symbol, stats.Score})
|
||
return true
|
||
})
|
||
|
||
// 按评分排序
|
||
sort.Slice(topScores, func(i, j int) bool {
|
||
return topScores[i].score > topScores[j].score
|
||
})
|
||
m.FilterSymbol = nil
|
||
if len(topScores) > 0 {
|
||
log.Printf("🏆 评分TOP%v:", nember)
|
||
for i := 0; i < len(topScores) && i < nember; i++ {
|
||
m.FilterSymbol = append(m.FilterSymbol, topScores[i].symbol)
|
||
log.Printf(" %d. %s: %.1f分", i+1, topScores[i].symbol, topScores[i].score)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// evaluateSymbolScore 评估币种得分，决定是否保留
|
||
func (m *WSMonitor) evaluateSymbolScore(symbol string, features *SymbolFeatures) float64 {
|
||
score := 0.0
|
||
|
||
// 交易量活跃度评分 (权重: 40%)
|
||
if features.VolumeRatio5 > 1.5 {
|
||
score += 40 * math.Min(features.VolumeRatio5/5.0, 1.0)
|
||
}
|
||
|
||
// 价格波动评分 (权重: 30%)
|
||
volatilityScore := math.Abs(features.PriceChange15Min) * 1000 // 放大系数
|
||
score += 30 * math.Min(volatilityScore/10.0, 1.0) // 最大10%波动得满分
|
||
|
||
// RSI活跃度评分 (权重: 20%)
|
||
if features.RSI14 < 30 || features.RSI14 > 70 {
|
||
score += 20 // RSI在极端区域
|
||
} else if features.RSI14 < 40 || features.RSI14 > 60 {
|
||
score += 10 // RSI在活跃区域
|
||
}
|
||
|
||
// 交易量趋势评分 (权重: 10%)
|
||
if features.VolumeTrend > 1.2 {
|
||
score += 10 * math.Min(features.VolumeTrend/3.0, 1.0)
|
||
}
|
||
|
||
return score
|
||
}
|
||
|
||
// shouldRemoveFromFilter 判断是否应该从FilterSymbols中移除
|
||
func (m *WSMonitor) shouldRemoveFromFilter(symbol string) bool {
|
||
value, exists := m.symbolStats.Load(symbol)
|
||
if !exists {
|
||
return true // 没有统计信息，移除
|
||
}
|
||
|
||
stats := value.(*SymbolStats)
|
||
|
||
// 规则1: 超过30分钟没有活跃迹象
|
||
if time.Since(stats.LastActiveTime) > 30*time.Minute {
|
||
log.Printf("🔻 %s 因长时间不活跃被移除", symbol)
|
||
return true
|
||
}
|
||
|
||
// 规则2: 评分持续低于阈值 (最近5次评分平均)
|
||
if stats.Score < 15 { // 调整这个阈值
|
||
log.Printf("🔻 %s 因评分过低(%.1f)被移除", symbol, stats.Score)
|
||
return true
|
||
}
|
||
|
||
// 规则3: 超过2小时没有产生警报
|
||
if time.Since(stats.LastAlertTime) > 2*time.Hour && stats.AlertCount > 0 {
|
||
log.Printf("🔻 %s 因长时间无新警报被移除", symbol)
|
||
return true
|
||
}
|
||
|
||
return false
|
||
}
|
||
|
||
// updateSymbolStats 更新币种统计信息
|
||
func (m *WSMonitor) updateSymbolStats(symbol string, features *SymbolFeatures, hasAlert bool) {
|
||
now := time.Now()
|
||
|
||
value, exists := m.symbolStats.Load(symbol)
|
||
var stats *SymbolStats
|
||
|
||
if !exists {
|
||
stats = &SymbolStats{
|
||
LastActiveTime: now,
|
||
Score: m.evaluateSymbolScore(symbol, features),
|
||
}
|
||
} else {
|
||
stats = value.(*SymbolStats)
|
||
stats.LastActiveTime = now
|
||
|
||
// 平滑更新评分 (指数移动平均)
|
||
newScore := m.evaluateSymbolScore(symbol, features)
|
||
stats.Score = 0.7*stats.Score + 0.3*newScore
|
||
}
|
||
|
||
if hasAlert {
|
||
stats.AlertCount++
|
||
stats.LastAlertTime = now
|
||
}
|
||
|
||
if features.VolumeRatio5 > 2.0 {
|
||
stats.VolumeSpikeCount++
|
||
}
|
||
|
||
m.symbolStats.Store(symbol, stats)
|
||
}
|
||
|
||
// removeFromFilter 从FilterSymbols中移除币种
|
||
func (m *WSMonitor) removeFromFilter(symbol string) {
|
||
|
||
// 从filterSymbols中移除
|
||
m.filterSymbols.Delete(symbol)
|
||
m.symbolStats.Delete(symbol)
|
||
|
||
log.Printf("🗑️ 已移除币种监控: %s", symbol)
|
||
}
|
||
|
||
// cleanupInactiveSymbols 定期清理不活跃的币种
|
||
func (m *WSMonitor) cleanupInactiveSymbols() {
|
||
ticker := time.NewTicker(5 * time.Minute) // 每5分钟检查一次
|
||
defer ticker.Stop()
|
||
|
||
for range ticker.C {
|
||
var symbolsToRemove []string
|
||
|
||
// 收集需要移除的币种
|
||
m.filterSymbols.Range(func(key, value interface{}) bool {
|
||
symbol := key.(string)
|
||
if m.shouldRemoveFromFilter(symbol) {
|
||
symbolsToRemove = append(symbolsToRemove, symbol)
|
||
}
|
||
return true
|
||
})
|
||
|
||
// 执行移除操作
|
||
for _, symbol := range symbolsToRemove {
|
||
m.removeFromFilter(symbol)
|
||
}
|
||
|
||
if len(symbolsToRemove) > 0 {
|
||
log.Printf("🧹 清理完成，移除了 %d 个不活跃币种", len(symbolsToRemove))
|
||
}
|
||
}
|
||
}
|
||
|
||
// getSymbolScore 获取币种当前评分
|
||
func (m *WSMonitor) getSymbolScore(symbol string) float64 {
|
||
value, exists := m.symbolStats.Load(symbol)
|
||
if !exists {
|
||
return 0
|
||
}
|
||
return value.(*SymbolStats).Score
|
||
}
|