package spotservice import ( "bytes" "fmt" "go-admin/common/const/rediskey" "go-admin/common/global" "go-admin/common/helper" "go-admin/config" "go-admin/pkg/utility" "go-admin/services/binanceservice" "go-admin/services/excservice" "strings" "sync" "go-admin/models" log "github.com/go-admin-team/go-admin-core/logger" "github.com/shopspring/decimal" "github.com/bytedance/sonic" "go.uber.org/zap" ) var ( baseBinanceWsUrlAll = "wss://stream.binance.com:9443/stream?streams=" wsBin *excservice.BinanceWs binSetKey = make(map[string]bool) binSetKeyMu sync.RWMutex quoteAssetSymbols = []string{"USDT", "ETH", "BTC", "SOL", "BNB", "DOGE"} ) type BaseWsDepthStream struct { Stream string `json:"stream"` // Data models.DepthBin `json:"data"` //数据 } // GetBinance24hr 获取币安24小时价格变动信息 func GetBinance24hr(coin, curr string) (models.Ticker24, error) { ticker, err := excservice.GetTicker(coin, curr) if err != nil { log.Error("获取ticker失败", zap.String("coin", coin), zap.Error(err)) return models.Ticker24{}, err } return ticker, nil } // GetProLastDeal 获取最新交易记录 func GetProLastDeal(coin, currency string) ([]models.NewDealPush, error) { resp, err := excservice.GetTrades(coin, currency) if err != nil { log.Error("获取交易记录失败", zap.Error(err)) return nil, err } return resp, nil } // GetProDepth 获取交易对的市场深度 func GetProDepth(coin, currency string) (models.FiveItem, error) { depth, err := excservice.GetDepth(50, coin, currency) if err != nil { log.Error("获取市场深度失败", zap.String("coin", coin), zap.String("currency", currency), zap.Error(err)) return models.FiveItem{}, err } return createFive2(depth), nil } // 组合买卖5结果 func createFive2(result models.DepthBin) models.FiveItem { five := models.FiveItem{} //卖5单 slice1 := make([][]string, 0, 20) var sum1 float64 for _, item := range result.Asks { if len(item) > 1 { num0 := utility.FloatToStringZero(item[0]) num1 := utility.FloatToStringZero(item[1]) slice1 = append(slice1, []string{num0, num1}) } if len(item) == 2 { sum1 = utility.FloatAdd(sum1, utility.StringAsFloat(item[1])) } } tempNumAsk := 0.0 for k := 0; k < len(slice1); k++ { //(前数量总值+当前数量值)/总数量值 nowNum := utility.StringAsFloat(slice1[k][1]) add := utility.FloatAdd(tempNumAsk, nowNum) sc1 := utility.FloatDiv(add, sum1) tempNumAsk = add sc2 := decimal.NewFromFloat(sc1).Truncate(3).String() //.Float64() slice1[k] = append(slice1[k], sc2) } five.Sell = slice1 five.SellNum = sum1 //买5 slice2 := make([][]string, 0, 20) var sum2 float64 for _, item := range result.Bids { if len(item) > 1 { num0 := utility.FloatToStringZero(item[0]) num1 := utility.FloatToStringZero(item[1]) slice2 = append(slice2, []string{num0, num1}) } //slice2 = append(slice2, item) //item.Price, item.Amount}) if len(item) == 2 { sum2 = utility.FloatAdd(sum2, utility.StringAsFloat(item[1])) //sum2 + item.Amount } } tempNumBid := 0.0 for k := 0; k < len(slice2); k++ { //(前数量总值+当前数量值)/总数量值 nowNum := utility.StringAsFloat(slice2[k][1]) add := utility.FloatAdd(tempNumBid, nowNum) sc1 := utility.FloatDiv(add, sum2) tempNumBid = add //sc1 := utility.FloatDiv(utility.StringAsFloat(slice2[k][1]), sum2) //(slice2[k][1]) / sum2 sc2 := decimal.NewFromFloat(sc1).Truncate(3).String() slice2[k] = append(slice2[k], sc2) } five.Buy = slice2 five.BuyNum = sum2 return five } // StartBinanceProWs 启动币安现货市场推送 // workType normal-正常任务 trigger-主动触发任务 func StartBinanceProWs(workType string) { if wsBin == nil { wsBin = excservice.NewBinanceWs(baseBinanceWsUrlAll, "") } if wsBin == nil { log.Error("实例化wsBin失败") return } if wsBin != nil && config.ExtConfig.ProxyUrl != "" { wsBin.SetProxyUrl(config.ExtConfig.ProxyUrl) } wsBin.WorkType = workType wsBin.SetAllCallbacks(HandleWsAll, HandleWsAllKline) subscribeToTradeSet(wsBin) } // subscribeToTradeSet 订阅给定交易集合的所有流 func subscribeToTradeSet(ws *excservice.BinanceWs) { // 订阅ticker、深度和交易的所有流 streamNames := []string{ "!miniTicker@arr", } err := ws.SubscribeAll(strings.Join(streamNames, "/")) if err != nil { log.Error("订阅流失败", zap.String("streams", strings.Join(streamNames, ",")), zap.Error(err)) } } // HandleWsAll 处理从WebSocket接收到的消息 func HandleWsAll(msg []byte) { if bytes.Contains(msg, []byte("miniTicker@arr")) { handleTickerMessage(msg) } } // handleTickerMessage 处理ticker消息 func handleTickerMessage(msg []byte) { var streamResp tickerAllMessage if err := sonic.Unmarshal(msg, &streamResp); err != nil { log.Error("解码ticker消息失败", zap.Error(err)) return } if len(streamResp.Data) == 0 { log.Error("ticker消息为空") return } pairs := make([]map[string]interface{}, 0) trades := make([]models.TradeSet, 0) pairsVal, _ := helper.DefaultRedis.GetString(rediskey.SpotSymbolTicker) if err := sonic.UnmarshalString(pairsVal, &pairs); err != nil { log.Error("解码ticker消息失败", zap.Error(err)) return } for _, dataMap := range streamResp.Data { symbolName, ok := dataMap["s"].(string) if !ok { log.Error("ticker消息不包含有效symbol字段") return } if !utility.HasSuffix(symbolName, quoteAssetSymbols) { continue } key := fmt.Sprintf(global.TICKER_SPOT, global.EXCHANGE_BINANCE, symbolName) lastPriceKey := fmt.Sprintf(rediskey.SpotTickerLastPrice, global.EXCHANGE_BINANCE, symbolName) tcVal, _ := helper.DefaultRedis.GetString(key) tradeSet := models.TradeSet{} if err := sonic.UnmarshalString(tcVal, &tradeSet); err != nil { if tcVal != "" { log.Error("解析tradeSet失败", zap.Error(err)) } continue } var utcTime int64 var lastPrice decimal.Decimal if e, ok := dataMap["E"]; ok { utcTime = utility.ToInt64(e) } if e, ok := dataMap["c"]; ok { lastPrice = utility.StrToDecimal(e.(string)).Truncate(int32(tradeSet.PriceDigit)) } tradeSet.LastPrice = lastPrice.String() tradeSet.OpenPrice = utility.StrToFloatCut(dataMap["o"].(string), int32(tradeSet.PriceDigit)) tradeSet.HighPrice = utility.StringFloat64Cut(dataMap["h"].(string), int32(tradeSet.PriceDigit)) tradeSet.LowPrice = utility.StringFloat64Cut(dataMap["l"].(string), int32(tradeSet.PriceDigit)) tradeSet.Volume = utility.StringFloat64Cut(dataMap["v"].(string), int32(tradeSet.AmountDigit)) tradeSet.QuoteVolume = utility.StringFloat64Cut(dataMap["q"].(string), 5) for index := range pairs { if symbol, ok := pairs[index]["symbol"].(string); ok { if symbol == symbolName { pairs[index]["price"] = tradeSet.LastPrice break } } } trades = append(trades, tradeSet) tcVal, _ = sonic.MarshalString(&tradeSet) if tcVal != "" { if err := helper.DefaultRedis.SetString(key, tcVal); err != nil { log.Error("redis保存交易对失败", tradeSet.Coin, tradeSet.Currency) } } val, _ := helper.DefaultRedis.GetAllList(rediskey.CacheSymbolLastPrice) if utility.ContainsStr(val, symbolName) { //行情存储时间 lastUtc := utcTime - 1000*60*60 content := fmt.Sprintf("%d:%s", utcTime, lastPrice.String()) if _, err := helper.DefaultRedis.RemoveBeforeScore(lastPriceKey, float64(lastUtc)); err != nil { log.Errorf("移除 现货交易对:%s %d之前的最新成交价失败,err:%v", symbolName, lastUtc, err) } if err := helper.DefaultRedis.AddSortSet(lastPriceKey, float64(utcTime), content); err != nil { log.Errorf("添加 现货交易对:%s %d之前的最新成交价失败,err:%v", symbolName, lastUtc, err) } } } //判断触发现货下单 if len(trades) > 0 { for index := range trades { // 主单触发 utility.SafeGoParam(binanceservice.JudgeSpotPrice, trades[index]) // 止损单 utility.SafeGoParam(binanceservice.JudgeSpotStopLoss, trades[index]) //减仓 utility.SafeGoParam(binanceservice.JudgeSpotReduce, trades[index]) //加仓 utility.SafeGoParam(binanceservice.JudgeSpotAddPosition, trades[index]) } } } // // HandleWsAllKline 处理kline推送结果 func HandleWsAllKline(msg []byte, tradeSet models.TradeSet) { } type WskLineData struct { Line string `json:"i"` //"1m",K线间隔 Timestamp int64 `json:"t"` // 这根K线的起始时间 Open string `json:"o"` // 这根K线期间第一笔成交价 Close string `json:"c"` // 这根K线期间末一笔成交价 High string `json:"h"` // 这根K线期间最高成交价 Low string `json:"l"` // 这根K线期间最低成交价 Vol string `json:"v"` // 这根K线期间成交量 QuoteVolume string `json:"q"` // 这根K线期间成交额 } type tickerAllMessage struct { Stream string `json:"stream"` Data []map[string]interface{} `json:"data"` } var ( depthDuration int64 = 1700 //深度推送时间间隔,单位毫秒 tickerDuration int64 = 1300 //24小时推送时间间隔,单位毫秒 //allPushDuration int64 = 2000 //allpus推送时间间隔,单位毫秒 marketDuration int64 = 3000 //首页+行情页推送时间间隔,单位毫秒 dealDuration int64 = 1300 //最新成交推送时间间隔,单位毫秒 klineDuration int64 = 800 //kline推送时间间隔,单位毫秒 usdtCode = "USDT" )