feat: 完成统计功能实现并修复 WebSocket 并发问题

- 实现全局统计接口（/api/copy-trading/statistics/global） - 实现 Leader 统计接口（/api/copy-trading/statistics/leader） - 实现分类统计接口（/api/copy-trading/statistics/category） - 优化前端统计页面，支持时间范围筛选和响应式布局 - 修复 WebSocket 并发发送消息导致的状态冲突问题 - 为每个会话添加同步锁，确保消息按顺序发送 - 优化错误处理，避免 TEXT_PARTIAL_WRITING 异常
2025-12-03 02:02:18 +08:00
parent dec165cb24
commit 719b01508a
5 changed files with 397 additions and 14 deletions
@@ -49,6 +49,102 @@ class CopyTradingStatisticsController(
            ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_STATISTICS_FETCH_FAILED, e.message))
        }
    }
+    
+    /**
+     * 获取全局统计
+     * POST /api/copy-trading/statistics/global
+     */
+    @PostMapping("/global")
+    fun getGlobalStatistics(@RequestBody request: GlobalStatisticsRequest): ResponseEntity<ApiResponse<StatisticsResponse>> {
+        return try {
+            val result = runBlocking { 
+                statisticsService.getGlobalStatistics(request.startTime, request.endTime) 
+            }
+            result.fold(
+                onSuccess = { response ->
+                    ResponseEntity.ok(ApiResponse.success(response))
+                },
+                onFailure = { e ->
+                    logger.error("获取全局统计失败", e)
+                    when (e) {
+                        is IllegalArgumentException -> ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_ERROR, e.message))
+                        else -> ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_STATISTICS_FETCH_FAILED, e.message))
+                    }
+                }
+            )
+        } catch (e: Exception) {
+            logger.error("获取全局统计异常", e)
+            ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_STATISTICS_FETCH_FAILED, e.message))
+        }
+    }
+    
+    /**
+     * 获取 Leader 统计
+     * POST /api/copy-trading/statistics/leader
+     */
+    @PostMapping("/leader")
+    fun getLeaderStatistics(@RequestBody request: LeaderStatisticsRequest): ResponseEntity<ApiResponse<StatisticsResponse>> {
+        return try {
+            if (request.leaderId <= 0) {
+                return ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_LEADER_ID_INVALID))
+            }
+            
+            val result = runBlocking { 
+                statisticsService.getLeaderStatistics(request.leaderId, request.startTime, request.endTime) 
+            }
+            result.fold(
+                onSuccess = { response ->
+                    ResponseEntity.ok(ApiResponse.success(response))
+                },
+                onFailure = { e ->
+                    logger.error("获取 Leader 统计失败: leaderId=${request.leaderId}", e)
+                    when (e) {
+                        is IllegalArgumentException -> ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_ERROR, e.message))
+                        else -> ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_STATISTICS_FETCH_FAILED, e.message))
+                    }
+                }
+            )
+        } catch (e: Exception) {
+            logger.error("获取 Leader 统计异常: leaderId=${request.leaderId}", e)
+            ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_STATISTICS_FETCH_FAILED, e.message))
+        }
+    }
+    
+    /**
+     * 获取分类统计
+     * POST /api/copy-trading/statistics/category
+     */
+    @PostMapping("/category")
+    fun getCategoryStatistics(@RequestBody request: CategoryStatisticsRequest): ResponseEntity<ApiResponse<StatisticsResponse>> {
+        return try {
+            if (request.category.isBlank()) {
+                return ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_ERROR, "分类不能为空"))
+            }
+            
+            if (request.category != "sports" && request.category != "crypto") {
+                return ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_ERROR, "分类必须是 sports 或 crypto"))
+            }
+            
+            val result = runBlocking { 
+                statisticsService.getCategoryStatistics(request.category, request.startTime, request.endTime) 
+            }
+            result.fold(
+                onSuccess = { response ->
+                    ResponseEntity.ok(ApiResponse.success(response))
+                },
+                onFailure = { e ->
+                    logger.error("获取分类统计失败: category=${request.category}", e)
+                    when (e) {
+                        is IllegalArgumentException -> ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_ERROR, e.message))
+                        else -> ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_STATISTICS_FETCH_FAILED, e.message))
+                    }
+                }
+            )
+        } catch (e: Exception) {
+            logger.error("获取分类统计异常: category=${request.category}", e)
+            ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_STATISTICS_FETCH_FAILED, e.message))
+        }
+    }
 }

 /**
@@ -112,3 +112,41 @@ data class StatisticsDetailRequest(
    val copyTradingId: Long
 )

+/**
+ * 全局统计请求
+ */
+data class GlobalStatisticsRequest(
+    val startTime: Long? = null,
+    val endTime: Long? = null
+)
+
+/**
+ * Leader 统计请求
+ */
+data class LeaderStatisticsRequest(
+    val leaderId: Long,
+    val startTime: Long? = null,
+    val endTime: Long? = null
+)
+
+/**
+ * 分类统计请求
+ */
+data class CategoryStatisticsRequest(
+    val category: String,  // sports 或 crypto
+    val startTime: Long? = null,
+    val endTime: Long? = null
+)
+
+/**
+ * 统计响应（全局/Leader/分类）
+ */
+data class StatisticsResponse(
+    val totalOrders: Long,
+    val totalPnl: String,
+    val winRate: String,
+    val avgPnl: String,
+    val maxProfit: String,
+    val maxLoss: String
+)
+
@@ -393,6 +393,166 @@ class CopyTradingStatisticsService(
        return percent.setScale(2, RoundingMode.HALF_UP).toString()
    }
    
+    /**
+     * 获取全局统计
+     */
+    suspend fun getGlobalStatistics(startTime: Long? = null, endTime: Long? = null): Result<StatisticsResponse> {
+        return try {
+            // 获取所有跟单关系
+            val allCopyTradings = copyTradingRepository.findAll()
+            
+            // 计算统计信息
+            val statistics = calculateAggregateStatistics(allCopyTradings.map { it.id!! }, startTime, endTime)
+            
+            Result.success(statistics)
+        } catch (e: Exception) {
+            logger.error("获取全局统计失败", e)
+            Result.failure(e)
+        }
+    }
+    
+    /**
+     * 获取 Leader 统计
+     */
+    suspend fun getLeaderStatistics(leaderId: Long, startTime: Long? = null, endTime: Long? = null): Result<StatisticsResponse> {
+        return try {
+            // 获取该 Leader 的所有跟单关系
+            val copyTradings = copyTradingRepository.findByLeaderId(leaderId)
+            
+            if (copyTradings.isEmpty()) {
+                return Result.failure(IllegalArgumentException("Leader $leaderId 没有跟单关系"))
+            }
+            
+            // 计算统计信息
+            val statistics = calculateAggregateStatistics(copyTradings.map { it.id!! }, startTime, endTime)
+            
+            Result.success(statistics)
+        } catch (e: Exception) {
+            logger.error("获取 Leader 统计失败: leaderId=$leaderId", e)
+            Result.failure(e)
+        }
+    }
+    
+    /**
+     * 获取分类统计
+     */
+    suspend fun getCategoryStatistics(category: String, startTime: Long? = null, endTime: Long? = null): Result<StatisticsResponse> {
+        return try {
+            // 验证分类
+            if (category != "sports" && category != "crypto") {
+                return Result.failure(IllegalArgumentException("分类必须是 sports 或 crypto"))
+            }
+            
+            // 获取该分类的所有 Leader
+            val leaders = leaderRepository.findAll().filter { it.category == category }
+            
+            if (leaders.isEmpty()) {
+                return Result.failure(IllegalArgumentException("分类 $category 没有 Leader"))
+            }
+            
+            // 获取这些 Leader 的所有跟单关系
+            val leaderIds = leaders.mapNotNull { it.id }
+            val copyTradings = copyTradingRepository.findAll().filter { it.leaderId in leaderIds }
+            
+            if (copyTradings.isEmpty()) {
+                return Result.failure(IllegalArgumentException("分类 $category 没有跟单关系"))
+            }
+            
+            // 计算统计信息
+            val statistics = calculateAggregateStatistics(copyTradings.map { it.id!! }, startTime, endTime)
+            
+            Result.success(statistics)
+        } catch (e: Exception) {
+            logger.error("获取分类统计失败: category=$category", e)
+            Result.failure(e)
+        }
+    }
+    
+    /**
+     * 计算聚合统计信息（多个跟单关系的汇总）
+     */
+    private suspend fun calculateAggregateStatistics(
+        copyTradingIds: List<Long>,
+        startTime: Long?,
+        endTime: Long?
+    ): StatisticsResponse {
+        // 获取所有买入订单
+        val allBuyOrders = copyTradingIds.flatMap { copyOrderTrackingRepository.findByCopyTradingId(it) }
+            .filter { order ->
+                // 时间筛选
+                when {
+                    startTime != null && endTime != null -> order.createdAt >= startTime && order.createdAt <= endTime
+                    startTime != null -> order.createdAt >= startTime
+                    endTime != null -> order.createdAt <= endTime
+                    else -> true
+                }
+            }
+        
+        // 获取所有匹配明细（已实现盈亏）
+        val allMatchDetails = copyTradingIds.flatMap { sellMatchDetailRepository.findByCopyTradingId(it) }
+            .filter { detail ->
+                // 时间筛选
+                when {
+                    startTime != null && endTime != null -> detail.createdAt >= startTime && detail.createdAt <= endTime
+                    startTime != null -> detail.createdAt >= startTime
+                    endTime != null -> detail.createdAt <= endTime
+                    else -> true
+                }
+            }
+        
+        // 计算统计指标
+        val totalOrders = allBuyOrders.size.toLong()
+        val totalPnl = allMatchDetails.sumOf { it.realizedPnl.toSafeBigDecimal() }
+        
+        // 计算胜率：盈利订单数 / 总订单数
+        // 盈利订单：该订单的所有匹配明细的盈亏总和 > 0
+        val profitableOrders = allBuyOrders.count { buyOrder ->
+            val orderPnl = allMatchDetails
+                .filter { it.buyOrderId == buyOrder.buyOrderId }
+                .sumOf { it.realizedPnl.toSafeBigDecimal() }
+            orderPnl.gt(BigDecimal.ZERO)
+        }
+        val winRate = if (totalOrders > 0) {
+            (BigDecimal(profitableOrders).divide(BigDecimal(totalOrders), 4, RoundingMode.HALF_UP) * BigDecimal(100))
+                .setScale(2, RoundingMode.HALF_UP)
+        } else {
+            BigDecimal.ZERO
+        }
+        
+        // 平均盈亏
+        val avgPnl = if (totalOrders > 0) {
+            totalPnl.divide(BigDecimal(totalOrders), 8, RoundingMode.HALF_UP)
+        } else {
+            BigDecimal.ZERO
+        }
+        
+        // 最大盈利和最大亏损（按订单计算）
+        var maxProfit = BigDecimal.ZERO
+        var maxLoss = BigDecimal.ZERO
+        
+        allBuyOrders.forEach { buyOrder ->
+            val orderPnl = allMatchDetails
+                .filter { it.buyOrderId == buyOrder.buyOrderId }
+                .sumOf { it.realizedPnl.toSafeBigDecimal() }
+            
+            if (orderPnl.gt(maxProfit)) {
+                maxProfit = orderPnl
+            }
+            if (orderPnl < maxLoss) {
+                maxLoss = orderPnl
+            }
+        }
+        
+        return StatisticsResponse(
+            totalOrders = totalOrders,
+            totalPnl = totalPnl.toString(),
+            winRate = winRate.toString(),
+            avgPnl = avgPnl.toString(),
+            maxProfit = maxProfit.toString(),
+            maxLoss = maxLoss.toString()
+        )
+    }
+    
    /**
     * 统计数据结构
     */
@@ -34,6 +34,9 @@ class UnifiedWebSocketHandler(
    // 存储每个连接的最后活动时间
    private val lastActivityTime = ConcurrentHashMap<String, Long>()
    
+    // 存储每个会话的同步锁，确保同一会话的消息按顺序发送
+    private val sessionLocks = ConcurrentHashMap<String, Any>()
+    
    // 协程作用域
    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
    private var cleanupJob: Job? = null
@@ -52,6 +55,8 @@ class UnifiedWebSocketHandler(
    override fun afterConnectionEstablished(session: WebSocketSession) {
        clientSessions[session.id] = session
        lastActivityTime[session.id] = System.currentTimeMillis()
+        // 为每个会话创建同步锁
+        sessionLocks[session.id] = Any()
        
        // 注册会话到订阅服务
        subscriptionService.registerSession(session.id) { wsMessage ->
@@ -65,10 +70,20 @@ class UnifiedWebSocketHandler(
        // 处理心跳
        if (payload == "PING" || payload == "ping") {
            lastActivityTime[session.id] = System.currentTimeMillis()
-            try {
-                session.sendMessage(TextMessage("PONG"))
-            } catch (e: Exception) {
-                logger.error("发送心跳响应失败: ${session.id}, ${e.message}", e)
+            // 心跳响应也使用同步锁，避免与数据消息冲突
+            val lock = sessionLocks[session.id]
+            if (lock != null && session.isOpen) {
+                synchronized(lock) {
+                    try {
+                        if (session.isOpen) {
+                            session.sendMessage(TextMessage("PONG"))
+                        }
+                    } catch (e: IllegalStateException) {
+                        logger.warn("发送心跳响应时 WebSocket 状态异常: ${session.id}, ${e.message}")
+                    } catch (e: Exception) {
+                        logger.error("发送心跳响应失败: ${session.id}, ${e.message}", e)
+                    }
+                }
            }
            return
        }
@@ -130,21 +145,42 @@ class UnifiedWebSocketHandler(
    
    /**
     * 发送消息给客户端
+     * 使用同步锁确保同一会话的消息按顺序发送，避免并发冲突
     */
    private fun sendMessageToClient(sessionId: String, message: WsMessage) {
        val session = clientSessions[sessionId]
-        if (session != null && session.isOpen) {
+        if (session == null || !session.isOpen) {
+            logger.warn("客户端会话不存在或已关闭: $sessionId")
+            cleanup(sessionId)
+            return
+        }
+        
+        // 获取该会话的同步锁
+        val lock = sessionLocks[sessionId] ?: return
+        
+        // 使用同步块确保同一会话的消息按顺序发送
+        synchronized(lock) {
+            // 再次检查会话状态（可能在等待锁的过程中会话已关闭）
+            val currentSession = clientSessions[sessionId]
+            if (currentSession == null || !currentSession.isOpen) {
+                logger.warn("客户端会话在发送消息前已关闭: $sessionId")
+                cleanup(sessionId)
+                return
+            }
+            
            try {
                val json = objectMapper.writeValueAsString(message)
-                session.sendMessage(TextMessage(json))
+                currentSession.sendMessage(TextMessage(json))
                lastActivityTime[sessionId] = System.currentTimeMillis()
+            } catch (e: IllegalStateException) {
+                // WebSocket 状态异常（如 TEXT_PARTIAL_WRITING），记录但不清理会话
+                // 这可能是暂时的状态问题，让后续消息有机会重试
+                logger.warn("发送消息时 WebSocket 状态异常: $sessionId, ${e.message}")
+                // 不立即清理，等待连接自然关闭或下次发送时再处理
            } catch (e: Exception) {
                logger.error("发送消息失败: $sessionId, ${e.message}", e)
                cleanup(sessionId)
            }
-        } else {
-            logger.warn("客户端会话不存在或已关闭: $sessionId")
-            cleanup(sessionId)
        }
    }
    
@@ -155,12 +191,14 @@ class UnifiedWebSocketHandler(
        try {
            val session = clientSessions.remove(sessionId)
            lastActivityTime.remove(sessionId)
+            sessionLocks.remove(sessionId)  // 清理同步锁
            subscriptionService.unregisterSession(sessionId)
            
            if (session != null && session.isOpen) {
                try {
                    session.close(CloseStatus.NORMAL)
                } catch (e: Exception) {
+                    // 忽略关闭时的异常
                }
            }
            
@@ -1,13 +1,20 @@
 import { useEffect, useState } from 'react'
-import { Card, Row, Col, Statistic, message } from 'antd'
-import { ArrowUpOutlined, ArrowDownOutlined } from '@ant-design/icons'
+import { Card, Row, Col, Statistic, message, DatePicker, Space, Button, Typography } from 'antd'
+import { ArrowUpOutlined, ArrowDownOutlined, ReloadOutlined } from '@ant-design/icons'
+import type { Dayjs } from 'dayjs'
 import { apiService } from '../services/api'
 import type { Statistics as StatisticsType } from '../types'
 import { formatUSDC } from '../utils'
+import { useMediaQuery } from 'react-responsive'
+
+const { RangePicker } = DatePicker
+const { Title } = Typography

 const Statistics: React.FC = () => {
+  const isMobile = useMediaQuery({ maxWidth: 768 })
  const [stats, setStats] = useState<StatisticsType | null>(null)
  const [loading, setLoading] = useState(false)
+  const [dateRange, setDateRange] = useState<[Dayjs | null, Dayjs | null]>([null, null])
  
  useEffect(() => {
    fetchStatistics()
@@ -16,7 +23,10 @@ const Statistics: React.FC = () => {
  const fetchStatistics = async () => {
    setLoading(true)
    try {
-      const response = await apiService.statistics.global()
+      const startTime = dateRange[0] ? dateRange[0].valueOf() : undefined
+      const endTime = dateRange[1] ? dateRange[1].valueOf() : undefined
+      
+      const response = await apiService.statistics.global({ startTime, endTime })
      if (response.data.code === 0 && response.data.data) {
        setStats(response.data.data)
      } else {
@@ -29,10 +39,49 @@ const Statistics: React.FC = () => {
    }
  }
  
+  const handleDateRangeChange = (dates: [Dayjs | null, Dayjs | null] | null) => {
+    setDateRange(dates || [null, null])
+  }
+  
+  const handleReset = () => {
+    setDateRange([null, null])
+    // 重置后自动刷新
+    setTimeout(() => {
+      fetchStatistics()
+    }, 100)
+  }
+  
  return (
    <div>
-      <div style={{ marginBottom: '16px' }}>
-        <h2>统计信息</h2>
+      <div style={{ marginBottom: '16px', display: 'flex', justifyContent: 'space-between', alignItems: 'center', flexWrap: 'wrap', gap: '12px' }}>
+        <Title level={2} style={{ margin: 0 }}>统计信息</Title>
+        <Space size="middle" wrap>
+          <RangePicker
+            value={dateRange}
+            onChange={handleDateRangeChange}
+            format="YYYY-MM-DD"
+            placeholder={['开始日期', '结束日期']}
+            size={isMobile ? 'middle' : 'large'}
+            allowClear
+          />
+          <Button
+            type="primary"
+            icon={<ReloadOutlined />}
+            onClick={fetchStatistics}
+            loading={loading}
+            size={isMobile ? 'middle' : 'large'}
+          >
+            刷新
+          </Button>
+          {(dateRange[0] || dateRange[1]) && (
+            <Button
+              onClick={handleReset}
+              size={isMobile ? 'middle' : 'large'}
+            >
+              重置
+            </Button>
+          )}
+        </Space>
      </div>
      
      <Row gutter={[16, 16]}>
@@ -73,6 +122,8 @@ const Statistics: React.FC = () => {
            <Statistic
              title="平均盈亏"
              value={formatUSDC(stats?.avgPnl || '0')}
+              prefix={stats?.avgPnl && parseFloat(stats.avgPnl || '0') >= 0 ? <ArrowUpOutlined /> : <ArrowDownOutlined />}
+              valueStyle={{ color: stats?.avgPnl && parseFloat(stats.avgPnl || '0') >= 0 ? '#3f8600' : '#cf1322' }}
              suffix="USDC"
              loading={loading}
            />