liuyu/nofx
1
0
Fork 0
mirror of https://github.com/NoFxAiOS/nofx.git synced 2026-06-06 05:51:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge from beta

Browse Source
This commit is contained in:
Icy
2025-11-13 23:05:57 +08:00
parent 9927ede8fd b43315d9fc
commit d787b72d75
210 changed files with 46931 additions and 7541 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
market/api_client.go
View File

@@ -6,6 +6,7 @@ import (
"io"
"log"
"net/http"
"nofx/hook"
"strconv"
"time"
)
@@ -19,10 +20,18 @@ type APIClient struct {
}
func NewAPIClient() *APIClient {
client := &http.Client{
Timeout: 30 * time.Second,
}
hookRes := hook.HookExec[hook.SetHttpClientResult](hook.SET_HTTP_CLIENT, client)
if hookRes != nil && hookRes.Error() == nil {
log.Printf("使用Hook设置的HTTP客户端")
client = hookRes.GetResult()
}
return &APIClient{
client: &http.Client{
Timeout: 30 * time.Second,
},
client: client,
}
}
@@ -74,6 +83,7 @@ func (c *APIClient) GetKlines(symbol, interval string, limit int) ([]Kline, erro
var klineResponses []KlineResponse
err = json.Unmarshal(body, &klineResponses)
if err != nil {
log.Printf("获取K线数据失败,响应内容: %s", string(body))
return nil, err
}

159
market/data.go
View File

@@ -4,10 +4,24 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"math"
"net/http"
"strconv"
"strings"
"sync"
"time"
)
// FundingRateCache 资金费率缓存结构
// Binance Funding Rate 每 8 小时才更新一次，使用 1 小时缓存可显著减少 API 调用
type FundingRateCache struct {
Rate float64
UpdatedAt time.Time
}
var (
fundingRateMap sync.Map // map[string]*FundingRateCache
frCacheTTL = 1 * time.Hour
)
// Get 获取指定代币的市场数据
@@ -22,12 +36,26 @@ func Get(symbol string) (*Data, error) {
return nil, fmt.Errorf("获取3分钟K线失败: %v", err)
}
// Data staleness detection: Prevent DOGEUSDT-style price freeze issues
if isStaleData(klines3m, symbol) {
log.Printf("⚠️ WARNING: %s detected stale data (consecutive price freeze), skipping symbol", symbol)
return nil, fmt.Errorf("%s data is stale, possible cache failure", symbol)
}
// 获取4小时K线数据 (最近10个)
klines4h, err = WSMonitorCli.GetCurrentKlines(symbol, "4h") // 多获取用于计算指标
if err != nil {
return nil, fmt.Errorf("获取4小时K线失败: %v", err)
}
// 检查数据是否为空
if len(klines3m) == 0 {
return nil, fmt.Errorf("3分钟K线数据为空")
}
if len(klines4h) == 0 {
return nil, fmt.Errorf("4小时K线数据为空")
}
// 计算当前指标 (基于3分钟最新数据)
currentPrice := klines3m[len(klines3m)-1].Close
currentEMA20 := calculateEMA(klines3m, 20)
@@ -206,6 +234,7 @@ func calculateIntradaySeries(klines []Kline) *IntradayData {
MACDValues: make([]float64, 0, 10),
RSI7Values: make([]float64, 0, 10),
RSI14Values: make([]float64, 0, 10),
Volume: make([]float64, 0, 10),
}
// 获取最近10个数据点
@@ -216,6 +245,7 @@ func calculateIntradaySeries(klines []Kline) *IntradayData {
for i := start; i < len(klines); i++ {
data.MidPrices = append(data.MidPrices, klines[i].Close)
data.Volume = append(data.Volume, klines[i].Volume)
// 计算每个点的EMA20
if i >= 19 {
@@ -240,6 +270,9 @@ func calculateIntradaySeries(klines []Kline) *IntradayData {
}
}
// 计算3m ATR14
data.ATR14 = calculateATR(klines, 14)
return data
}
@@ -293,7 +326,8 @@ func calculateLongerTermData(klines []Kline) *LongerTermData {
func getOpenInterestData(symbol string) (*OIData, error) {
url := fmt.Sprintf("https://fapi.binance.com/fapi/v1/openInterest?symbol=%s", symbol)
resp, err := http.Get(url)
apiClient := NewAPIClient()
resp, err := apiClient.client.Get(url)
if err != nil {
return nil, err
}
@@ -322,11 +356,23 @@ func getOpenInterestData(symbol string) (*OIData, error) {
}, nil
}
// getFundingRate 获取资金费率
// getFundingRate 获取资金费率（优化：使用 1 小时缓存）
func getFundingRate(symbol string) (float64, error) {
// 检查缓存（有效期 1 小时）
// Funding Rate 每 8 小时才更新，1 小时缓存非常合理
if cached, ok := fundingRateMap.Load(symbol); ok {
cache := cached.(*FundingRateCache)
if time.Since(cache.UpdatedAt) < frCacheTTL {
// 缓存命中，直接返回
return cache.Rate, nil
}
}
// 缓存过期或不存在，调用 API
url := fmt.Sprintf("https://fapi.binance.com/fapi/v1/premiumIndex?symbol=%s", symbol)
resp, err := http.Get(url)
apiClient := NewAPIClient()
resp, err := apiClient.client.Get(url)
if err != nil {
return 0, err
}
@@ -352,6 +398,13 @@ func getFundingRate(symbol string) (float64, error) {
}
rate, _ := strconv.ParseFloat(result.LastFundingRate, 64)
// 更新缓存
fundingRateMap.Store(symbol, &FundingRateCache{
Rate: rate,
UpdatedAt: time.Now(),
})
return rate, nil
}
@@ -359,15 +412,20 @@ func getFundingRate(symbol string) (float64, error) {
func Format(data *Data) string {
var sb strings.Builder
sb.WriteString(fmt.Sprintf("current_price = %.2f, current_ema20 = %.3f, current_macd = %.3f, current_rsi (7 period) = %.3f\n\n",
data.CurrentPrice, data.CurrentEMA20, data.CurrentMACD, data.CurrentRSI7))
// 使用动态精度格式化价格
priceStr := formatPriceWithDynamicPrecision(data.CurrentPrice)
sb.WriteString(fmt.Sprintf("current_price = %s, current_ema20 = %.3f, current_macd = %.3f, current_rsi (7 period) = %.3f\n\n",
priceStr, data.CurrentEMA20, data.CurrentMACD, data.CurrentRSI7))
sb.WriteString(fmt.Sprintf("In addition, here is the latest %s open interest and funding rate for perps:\n\n",
data.Symbol))
if data.OpenInterest != nil {
sb.WriteString(fmt.Sprintf("Open Interest: Latest: %.2f Average: %.2f\n\n",
data.OpenInterest.Latest, data.OpenInterest.Average))
// 使用动态精度格式化 OI 数据
oiLatestStr := formatPriceWithDynamicPrecision(data.OpenInterest.Latest)
oiAverageStr := formatPriceWithDynamicPrecision(data.OpenInterest.Average)
sb.WriteString(fmt.Sprintf("Open Interest: Latest: %s Average: %s\n\n",
oiLatestStr, oiAverageStr))
}
sb.WriteString(fmt.Sprintf("Funding Rate: %.2e\n\n", data.FundingRate))
@@ -394,6 +452,12 @@ func Format(data *Data) string {
if len(data.IntradaySeries.RSI14Values) > 0 {
sb.WriteString(fmt.Sprintf("RSI indicators (14‑Period): %s\n\n", formatFloatSlice(data.IntradaySeries.RSI14Values)))
}
if len(data.IntradaySeries.Volume) > 0 {
sb.WriteString(fmt.Sprintf("Volume: %s\n\n", formatFloatSlice(data.IntradaySeries.Volume)))
}
sb.WriteString(fmt.Sprintf("3m ATR (14‑period): %.3f\n\n", data.IntradaySeries.ATR14))
}
if data.LongerTermContext != nil {
@@ -420,11 +484,42 @@ func Format(data *Data) string {
return sb.String()
}
// formatFloatSlice 格式化float64切片为字符串
// formatPriceWithDynamicPrecision 根据价格区间动态选择精度
// 这样可以完美支持从超低价 meme coin (< 0.0001) 到 BTC/ETH 的所有币种
func formatPriceWithDynamicPrecision(price float64) string {
switch {
case price < 0.0001:
// 超低价 meme coin: 1000SATS, 1000WHY, DOGS
// 0.00002070 → "0.00002070" (8位小数)
return fmt.Sprintf("%.8f", price)
case price < 0.001:
// 低价 meme coin: NEIRO, HMSTR, HOT, NOT
// 0.00015060 → "0.000151" (6位小数)
return fmt.Sprintf("%.6f", price)
case price < 0.01:
// 中低价币: PEPE, SHIB, MEME
// 0.00556800 → "0.005568" (6位小数)
return fmt.Sprintf("%.6f", price)
case price < 1.0:
// 低价币: ASTER, DOGE, ADA, TRX
// 0.9954 → "0.9954" (4位小数)
return fmt.Sprintf("%.4f", price)
case price < 100:
// 中价币: SOL, AVAX, LINK, MATIC
// 23.4567 → "23.4567" (4位小数)
return fmt.Sprintf("%.4f", price)
default:
// 高价币: BTC, ETH (节省 Token)
// 45678.9123 → "45678.91" (2位小数)
return fmt.Sprintf("%.2f", price)
}
}
// formatFloatSlice 格式化float64切片为字符串（使用动态精度）
func formatFloatSlice(values []float64) string {
strValues := make([]string, len(values))
for i, v := range values {
strValues[i] = fmt.Sprintf("%.3f", v)
strValues[i] = formatPriceWithDynamicPrecision(v)
}
return "[" + strings.Join(strValues, ", ") + "]"
}
@@ -453,3 +548,47 @@ func parseFloat(v interface{}) (float64, error) {
return 0, fmt.Errorf("unsupported type: %T", v)
}
}
// isStaleData detects stale data (consecutive price freeze)
// Fix DOGEUSDT-style issue: consecutive N periods with completely unchanged prices indicate data source anomaly
func isStaleData(klines []Kline, symbol string) bool {
if len(klines) < 5 {
return false // Insufficient data to determine
}
// Detection threshold: 5 consecutive 3-minute periods with unchanged price (15 minutes without fluctuation)
const stalePriceThreshold = 5
const priceTolerancePct = 0.0001 // 0.01% fluctuation tolerance (avoid false positives)
// Take the last stalePriceThreshold K-lines
recentKlines := klines[len(klines)-stalePriceThreshold:]
firstPrice := recentKlines[0].Close
// Check if all prices are within tolerance
for i := 1; i < len(recentKlines); i++ {
priceDiff := math.Abs(recentKlines[i].Close-firstPrice) / firstPrice
if priceDiff > priceTolerancePct {
return false // Price fluctuation exists, data is normal
}
}
// Additional check: MACD and volume
// If price is unchanged but MACD/volume shows normal fluctuation, it might be a real market situation (extremely low volatility)
// Check if volume is also 0 (data completely frozen)
allVolumeZero := true
for _, k := range recentKlines {
if k.Volume > 0 {
allVolumeZero = false
break
}
}
if allVolumeZero {
log.Printf("⚠️ %s stale data confirmed: price freeze + zero volume", symbol)
return true
}
// Price frozen but has volume: might be extremely low volatility market, allow but log warning
log.Printf("⚠️ %s detected extreme price stability (no fluctuation for %d consecutive periods), but volume is normal", symbol, stalePriceThreshold)
return false
}

502
market/data_test.go Normal file
View File

@@ -0,0 +1,502 @@
package market
import (
"math"
"testing"
)
// generateTestKlines 生成测试用的 K线数据
func generateTestKlines(count int) []Kline {
klines := make([]Kline, count)
for i := 0; i < count; i++ {
// 生成模拟的价格数据，有一定的波动
basePrice := 100.0
variance := float64(i%10) * 0.5
open := basePrice + variance
high := open + 1.0
low := open - 0.5
close := open + 0.3
volume := 1000.0 + float64(i*100)
klines[i] = Kline{
OpenTime: int64(i * 180000), // 3分钟间隔
Open: open,
High: high,
Low: low,
Close: close,
Volume: volume,
CloseTime: int64((i+1)*180000 - 1),
}
}
return klines
}
// TestCalculateIntradaySeries_VolumeCollection 测试 Volume 数据收集
func TestCalculateIntradaySeries_VolumeCollection(t *testing.T) {
tests := []struct {
name string
klineCount int
expectedVolLen int
}{
{
name: "正常情况 - 20个K线",
klineCount: 20,
expectedVolLen: 10, // 应该收集最近10个
},
{
name: "刚好10个K线",
klineCount: 10,
expectedVolLen: 10,
},
{
name: "少于10个K线",
klineCount: 5,
expectedVolLen: 5, // 应该返回所有5个
},
{
name: "超过10个K线",
klineCount: 30,
expectedVolLen: 10, // 应该只返回最近10个
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
klines := generateTestKlines(tt.klineCount)
data := calculateIntradaySeries(klines)
if data == nil {
t.Fatal("calculateIntradaySeries returned nil")
}
if len(data.Volume) != tt.expectedVolLen {
t.Errorf("Volume length = %d, want %d", len(data.Volume), tt.expectedVolLen)
}
// 验证 Volume 数据正确性
if len(data.Volume) > 0 {
// 计算期望的起始索引
start := tt.klineCount - 10
if start < 0 {
start = 0
}
// 验证第一个 Volume 值
expectedFirstVolume := klines[start].Volume
if data.Volume[0] != expectedFirstVolume {
t.Errorf("First volume = %.2f, want %.2f", data.Volume[0], expectedFirstVolume)
}
// 验证最后一个 Volume 值
expectedLastVolume := klines[tt.klineCount-1].Volume
lastVolume := data.Volume[len(data.Volume)-1]
if lastVolume != expectedLastVolume {
t.Errorf("Last volume = %.2f, want %.2f", lastVolume, expectedLastVolume)
}
}
})
}
}
// TestCalculateIntradaySeries_VolumeValues 测试 Volume 值的正确性
func TestCalculateIntradaySeries_VolumeValues(t *testing.T) {
klines := []Kline{
{Close: 100.0, Volume: 1000.0, High: 101.0, Low: 99.0, Open: 100.0},
{Close: 101.0, Volume: 1100.0, High: 102.0, Low: 100.0, Open: 101.0},
{Close: 102.0, Volume: 1200.0, High: 103.0, Low: 101.0, Open: 102.0},
{Close: 103.0, Volume: 1300.0, High: 104.0, Low: 102.0, Open: 103.0},
{Close: 104.0, Volume: 1400.0, High: 105.0, Low: 103.0, Open: 104.0},
{Close: 105.0, Volume: 1500.0, High: 106.0, Low: 104.0, Open: 105.0},
{Close: 106.0, Volume: 1600.0, High: 107.0, Low: 105.0, Open: 106.0},
{Close: 107.0, Volume: 1700.0, High: 108.0, Low: 106.0, Open: 107.0},
{Close: 108.0, Volume: 1800.0, High: 109.0, Low: 107.0, Open: 108.0},
{Close: 109.0, Volume: 1900.0, High: 110.0, Low: 108.0, Open: 109.0},
}
data := calculateIntradaySeries(klines)
expectedVolumes := []float64{1000.0, 1100.0, 1200.0, 1300.0, 1400.0, 1500.0, 1600.0, 1700.0, 1800.0, 1900.0}
if len(data.Volume) != len(expectedVolumes) {
t.Fatalf("Volume length = %d, want %d", len(data.Volume), len(expectedVolumes))
}
for i, expected := range expectedVolumes {
if data.Volume[i] != expected {
t.Errorf("Volume[%d] = %.2f, want %.2f", i, data.Volume[i], expected)
}
}
}
// TestCalculateIntradaySeries_ATR14 测试 ATR14 计算
func TestCalculateIntradaySeries_ATR14(t *testing.T) {
tests := []struct {
name string
klineCount int
expectZero bool
expectNonZero bool
}{
{
name: "足够数据 - 20个K线",
klineCount: 20,
expectNonZero: true,
},
{
name: "刚好15个K线（ATR14需要至少15个）",
klineCount: 15,
expectNonZero: true,
},
{
name: "数据不足 - 14个K线",
klineCount: 14,
expectZero: true,
},
{
name: "数据不足 - 10个K线",
klineCount: 10,
expectZero: true,
},
{
name: "数据不足 - 5个K线",
klineCount: 5,
expectZero: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
klines := generateTestKlines(tt.klineCount)
data := calculateIntradaySeries(klines)
if data == nil {
t.Fatal("calculateIntradaySeries returned nil")
}
if tt.expectZero && data.ATR14 != 0 {
t.Errorf("ATR14 = %.3f, expected 0 (insufficient data)", data.ATR14)
}
if tt.expectNonZero && data.ATR14 <= 0 {
t.Errorf("ATR14 = %.3f, expected > 0", data.ATR14)
}
})
}
}
// TestCalculateATR 测试 ATR 计算函数
func TestCalculateATR(t *testing.T) {
tests := []struct {
name string
klines []Kline
period int
expectZero bool
}{
{
name: "正常计算 - 足够数据",
klines: []Kline{
{High: 102.0, Low: 100.0, Close: 101.0},
{High: 103.0, Low: 101.0, Close: 102.0},
{High: 104.0, Low: 102.0, Close: 103.0},
{High: 105.0, Low: 103.0, Close: 104.0},
{High: 106.0, Low: 104.0, Close: 105.0},
{High: 107.0, Low: 105.0, Close: 106.0},
{High: 108.0, Low: 106.0, Close: 107.0},
{High: 109.0, Low: 107.0, Close: 108.0},
{High: 110.0, Low: 108.0, Close: 109.0},
{High: 111.0, Low: 109.0, Close: 110.0},
{High: 112.0, Low: 110.0, Close: 111.0},
{High: 113.0, Low: 111.0, Close: 112.0},
{High: 114.0, Low: 112.0, Close: 113.0},
{High: 115.0, Low: 113.0, Close: 114.0},
{High: 116.0, Low: 114.0, Close: 115.0},
},
period: 14,
expectZero: false,
},
{
name: "数据不足 - 等于period",
klines: []Kline{
{High: 102.0, Low: 100.0, Close: 101.0},
{High: 103.0, Low: 101.0, Close: 102.0},
},
period: 2,
expectZero: true,
},
{
name: "数据不足 - 少于period",
klines: []Kline{
{High: 102.0, Low: 100.0, Close: 101.0},
},
period: 14,
expectZero: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
atr := calculateATR(tt.klines, tt.period)
if tt.expectZero {
if atr != 0 {
t.Errorf("calculateATR() = %.3f, expected 0 (insufficient data)", atr)
}
} else {
if atr <= 0 {
t.Errorf("calculateATR() = %.3f, expected > 0", atr)
}
}
})
}
}
// TestCalculateATR_TrueRange 测试 ATR 的 True Range 计算正确性
func TestCalculateATR_TrueRange(t *testing.T) {
// 创建一个简单的测试用例，手动计算期望的 ATR
klines := []Kline{
{High: 50.0, Low: 48.0, Close: 49.0}, // TR = 2.0
{High: 51.0, Low: 49.0, Close: 50.0}, // TR = max(2.0, 2.0, 1.0) = 2.0
{High: 52.0, Low: 50.0, Close: 51.0}, // TR = max(2.0, 2.0, 1.0) = 2.0
{High: 53.0, Low: 51.0, Close: 52.0}, // TR = 2.0
{High: 54.0, Low: 52.0, Close: 53.0}, // TR = 2.0
}
atr := calculateATR(klines, 3)
// 期望的计算：
// TR[1] = max(51-49, |51-49|, |49-49|) = 2.0
// TR[2] = max(52-50, |52-50|, |50-50|) = 2.0
// TR[3] = max(53-51, |53-51|, |51-51|) = 2.0
// 初始 ATR = (2.0 + 2.0 + 2.0) / 3 = 2.0
// TR[4] = max(54-52, |54-52|, |52-52|) = 2.0
// 平滑 ATR = (2.0*2 + 2.0) / 3 = 2.0
expectedATR := 2.0
tolerance := 0.01 // 允许小的浮点误差
if math.Abs(atr-expectedATR) > tolerance {
t.Errorf("calculateATR() = %.3f, want approximately %.3f", atr, expectedATR)
}
}
// TestCalculateIntradaySeries_ConsistencyWithOtherIndicators 测试 Volume 和其他指标的一致性
func TestCalculateIntradaySeries_ConsistencyWithOtherIndicators(t *testing.T) {
klines := generateTestKlines(30)
data := calculateIntradaySeries(klines)
// 所有数组应该存在
if data.MidPrices == nil {
t.Error("MidPrices should not be nil")
}
if data.Volume == nil {
t.Error("Volume should not be nil")
}
// MidPrices 和 Volume 应该有相同的长度（都是最近10个）
if len(data.MidPrices) != len(data.Volume) {
t.Errorf("MidPrices length (%d) should equal Volume length (%d)",
len(data.MidPrices), len(data.Volume))
}
// 所有 Volume 值应该大于 0
for i, vol := range data.Volume {
if vol <= 0 {
t.Errorf("Volume[%d] = %.2f, should be > 0", i, vol)
}
}
}
// TestCalculateIntradaySeries_EmptyKlines 测试空 K线数据
func TestCalculateIntradaySeries_EmptyKlines(t *testing.T) {
klines := []Kline{}
data := calculateIntradaySeries(klines)
if data == nil {
t.Fatal("calculateIntradaySeries should not return nil for empty klines")
}
// 所有切片应该为空
if len(data.MidPrices) != 0 {
t.Errorf("MidPrices length = %d, want 0", len(data.MidPrices))
}
if len(data.Volume) != 0 {
t.Errorf("Volume length = %d, want 0", len(data.Volume))
}
// ATR14 应该为 0（数据不足）
if data.ATR14 != 0 {
t.Errorf("ATR14 = %.3f, want 0", data.ATR14)
}
}
// TestCalculateIntradaySeries_VolumePrecision 测试 Volume 精度保持
func TestCalculateIntradaySeries_VolumePrecision(t *testing.T) {
klines := []Kline{
{Close: 100.0, Volume: 1234.5678, High: 101.0, Low: 99.0},
{Close: 101.0, Volume: 9876.5432, High: 102.0, Low: 100.0},
{Close: 102.0, Volume: 5555.1111, High: 103.0, Low: 101.0},
}
data := calculateIntradaySeries(klines)
expectedVolumes := []float64{1234.5678, 9876.5432, 5555.1111}
for i, expected := range expectedVolumes {
if data.Volume[i] != expected {
t.Errorf("Volume[%d] = %.4f, want %.4f (precision not preserved)",
i, data.Volume[i], expected)
}
}
}
// TestIsStaleData_NormalData tests that normal fluctuating data returns false
func TestIsStaleData_NormalData(t *testing.T) {
klines := []Kline{
{Close: 100.0, Volume: 1000},
{Close: 100.5, Volume: 1200},
{Close: 99.8, Volume: 900},
{Close: 100.2, Volume: 1100},
{Close: 100.1, Volume: 950},
}
result := isStaleData(klines, "BTCUSDT")
if result {
t.Error("Expected false for normal fluctuating data, got true")
}
}
// TestIsStaleData_PriceFreezeWithZeroVolume tests that frozen price + zero volume returns true
func TestIsStaleData_PriceFreezeWithZeroVolume(t *testing.T) {
klines := []Kline{
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
}
result := isStaleData(klines, "DOGEUSDT")
if !result {
t.Error("Expected true for frozen price + zero volume, got false")
}
}
// TestIsStaleData_PriceFreezeWithVolume tests that frozen price but normal volume returns false
func TestIsStaleData_PriceFreezeWithVolume(t *testing.T) {
klines := []Kline{
{Close: 100.0, Volume: 1000},
{Close: 100.0, Volume: 1200},
{Close: 100.0, Volume: 900},
{Close: 100.0, Volume: 1100},
{Close: 100.0, Volume: 950},
}
result := isStaleData(klines, "STABLECOIN")
if result {
t.Error("Expected false for frozen price but normal volume (low volatility market), got true")
}
}
// TestIsStaleData_InsufficientData tests that insufficient data (<5 klines) returns false
func TestIsStaleData_InsufficientData(t *testing.T) {
klines := []Kline{
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
}
result := isStaleData(klines, "BTCUSDT")
if result {
t.Error("Expected false for insufficient data (<5 klines), got true")
}
}
// TestIsStaleData_ExactlyFiveKlines tests edge case with exactly 5 klines
func TestIsStaleData_ExactlyFiveKlines(t *testing.T) {
// Stale case: exactly 5 frozen klines with zero volume
staleKlines := []Kline{
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
{Close: 100.0, Volume: 0},
}
result := isStaleData(staleKlines, "TESTUSDT")
if !result {
t.Error("Expected true for exactly 5 frozen klines with zero volume, got false")
}
// Normal case: exactly 5 klines with fluctuation
normalKlines := []Kline{
{Close: 100.0, Volume: 1000},
{Close: 100.1, Volume: 1100},
{Close: 99.9, Volume: 900},
{Close: 100.0, Volume: 1000},
{Close: 100.05, Volume: 950},
}
result = isStaleData(normalKlines, "TESTUSDT")
if result {
t.Error("Expected false for exactly 5 normal klines, got true")
}
}
// TestIsStaleData_WithinTolerance tests price changes within tolerance (0.01%)
func TestIsStaleData_WithinTolerance(t *testing.T) {
// Price changes within 0.01% tolerance should be treated as frozen
basePrice := 10000.0
tolerance := 0.0001 // 0.01%
smallChange := basePrice * tolerance * 0.5 // Half of tolerance
klines := []Kline{
{Close: basePrice, Volume: 1000},
{Close: basePrice + smallChange, Volume: 1000},
{Close: basePrice - smallChange, Volume: 1000},
{Close: basePrice, Volume: 1000},
{Close: basePrice + smallChange, Volume: 1000},
}
result := isStaleData(klines, "BTCUSDT")
// Should return false because there's normal volume despite tiny price changes
if result {
t.Error("Expected false for price within tolerance but with volume, got true")
}
}
// TestIsStaleData_MixedScenario tests realistic scenario with some history before freeze
func TestIsStaleData_MixedScenario(t *testing.T) {
// Simulate: normal trading → suddenly freezes
klines := []Kline{
{Close: 100.0, Volume: 1000}, // Normal
{Close: 100.5, Volume: 1200}, // Normal
{Close: 100.2, Volume: 1100}, // Normal
{Close: 50.0, Volume: 0}, // Freeze starts
{Close: 50.0, Volume: 0}, // Frozen
{Close: 50.0, Volume: 0}, // Frozen
{Close: 50.0, Volume: 0}, // Frozen
{Close: 50.0, Volume: 0}, // Frozen (last 5 are all frozen)
}
result := isStaleData(klines, "DOGEUSDT")
// Should detect stale data based on last 5 klines
if !result {
t.Error("Expected true for frozen last 5 klines with zero volume, got false")
}
}
// TestIsStaleData_EmptyKlines tests edge case with empty slice
func TestIsStaleData_EmptyKlines(t *testing.T) {
klines := []Kline{}
result := isStaleData(klines, "BTCUSDT")
if result {
t.Error("Expected false for empty klines, got true")
}
}

35
market/monitor.go
View File

@@ -121,19 +121,19 @@ func (m *WSMonitor) Start(coins []string) {
// 初始化交易对
err := m.Initialize(coins)
if err != nil {
log.Fatalf("❌ 初始化币种: %v", err)
log.Printf("❌ 初始化币种失败: %v", err)
return
}
err = m.combinedClient.Connect()
if err != nil {
log.Fatalf("❌ 批量订阅流: %v", err)
log.Printf("❌ 批量订阅流失败: %v", err)
return
}
// 订阅所有交易对
err = m.subscribeAll()
if err != nil {
log.Fatalf("❌ 订阅币种交易对: %v", err)
log.Printf("❌ 订阅币种交易对失败: %v", err)
return
}
}
@@ -159,7 +159,7 @@ func (m *WSMonitor) subscribeAll() error {
for _, st := range subKlineTime {
err := m.combinedClient.BatchSubscribeKlines(m.symbols, st)
if err != nil {
log.Fatalf("❌ 订阅3m K线: %v", err)
log.Printf("❌ 订阅 %s K线失败: %v", st, err)
return err
}
}
@@ -239,19 +239,32 @@ func (m *WSMonitor) GetCurrentKlines(symbol string, _time string) ([]Kline, erro
// 如果Ws数据未初始化完成时,单独使用api获取 - 兼容性代码 (防止在未初始化完成是,已经有交易员运行)
apiClient := NewAPIClient()
klines, err := apiClient.GetKlines(symbol, _time, 100)
m.getKlineDataMap(_time).Store(strings.ToUpper(symbol), klines) //动态缓存进缓存
if err != nil {
return nil, fmt.Errorf("获取%v分钟K线失败: %v", _time, err)
}
// 动态缓存进缓存
m.getKlineDataMap(_time).Store(strings.ToUpper(symbol), klines)
// 订阅 WebSocket 流
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)
log.Printf("警告: 动态订阅%v分钟K线失败: %v (使用API数据)", _time, subErr)
}
if err != nil {
return nil, fmt.Errorf("获取%v分钟K线失败: %v", _time, err)
}
return klines, fmt.Errorf("symbol不存在")
// ✅ FIX: 返回深拷贝而非引用
result := make([]Kline, len(klines))
copy(result, klines)
return result, nil
}
return value.([]Kline), nil
// ✅ FIX: 返回深拷贝而非引用，避免并发竞态条件
klines := value.([]Kline)
result := make([]Kline, len(klines))
copy(result, klines)
return result, nil
}
func (m *WSMonitor) Close() {

2
market/types.go
View File

@@ -30,6 +30,8 @@ type IntradayData struct {
MACDValues []float64
RSI7Values []float64
RSI14Values []float64
Volume []float64
ATR14 float64
}
// LongerTermData 长期数据(4小时时间框架)