Refactor trading execution tools and enhance risk management

- Updated CryptoTradingExecutionTools to delegate order execution to TradeExecutor, streamlining the trading process.
- Improved MarketMonitor to utilize a Channel for price updates, ensuring no tick is dropped and enhancing performance.
- Enhanced RiskManager to calculate total asset value in USDT asynchronously, improving accuracy in risk assessments.
- Introduced a new method in TradingDataService to compute the average entry price for better PnL estimation.
- Updated various models and services to support new features and improve overall code clarity and maintainability.

Author: MayueCif

src/Agents/Tools/Crypto/CryptoTradingExecutionTools.cs

@@ -10,27 +10,24 @@
 namespace MarketAssistant.Agents.Tools.Crypto;
 
 /// <summary>
-/// 虚拟币交易执行工具实现，供 TradingAgent 使用
+/// 虚拟币交易执行工具实现，供 TradingAgent 使用。下单委托给 TradeExecutor 统一入口。
 /// </summary>
 public class CryptoTradingExecutionTools : ITradingExecutionTools
 {
     private readonly BinanceAccountService _accountService;
     private readonly BinanceMarketDataService _marketDataService;
-    private readonly RiskManager _riskManager;
-    private readonly TradingDataService _dataService;
+    private readonly TradeExecutor _tradeExecutor;
     private readonly ILogger<CryptoTradingExecutionTools> _logger;
 
     public CryptoTradingExecutionTools(
         BinanceAccountService accountService,
         BinanceMarketDataService marketDataService,
-        RiskManager riskManager,
-        TradingDataService dataService,
+        TradeExecutor tradeExecutor,
         ILogger<CryptoTradingExecutionTools> logger)
     {
         _accountService = accountService;
         _marketDataService = marketDataService;
-        _riskManager = riskManager;
-        _dataService = dataService;
+        _tradeExecutor = tradeExecutor;
         _logger = logger;
     }
 
@@ -137,51 +134,13 @@ public async Task<TradeResult> PlaceOrderAsync(
         }
 
         if (effectivePrice <= 0)
-        {
-            return new TradeResult
-            {
-                Success = false,
-                ErrorMessage = $"无法确定 {symbol} 的有效价格，拒绝下单"
-            };
-        }
-
-        var riskCheck = await _riskManager.ValidateOrderAsync(symbol, side, quantity, effectivePrice);
-        if (!riskCheck.Passed)
-        {
-            _logger.LogWarning("风控拒绝: {Reason}", riskCheck.Reason);
-            return new TradeResult { Success = false, ErrorMessage = $"风控拒绝: {riskCheck.Reason}" };
-        }
-
-        try
-        {
-            var response = await _accountService.PlaceOrderAsync(
-                symbol, side.ToString().ToUpper(), type.ToString().ToUpper(), quantity, price);
+            return new TradeResult { Success = false, ErrorMessage = $"无法确定 {symbol} 的有效价格，拒绝下单" };
 
-            var record = new TradeRecord
-            {
-                StrategyId = "manual",
-                Symbol = symbol,
-                Side = side,
-                OrderType = type,
-                RequestedQty = quantity,
-                ExecutedQty = decimal.TryParse(response.ExecutedQty, out var eq) ? eq : 0,
-                RequestedPrice = price,
-                ExecutedPrice = decimal.TryParse(response.Price, out var ep) ? ep : effectivePrice,
-                Status = MapOrderStatus(response.Status),
-                BinanceOrderId = response.OrderId,
-                CompletedAt = response.Status == "FILLED" ? DateTime.UtcNow : null
-            };
-
-            await _dataService.SaveTradeRecordAsync(record);
-            await _dataService.UpdateDailyStatsAsync(0, 0);
-
-            return new TradeResult { Success = true, Record = record };
-        }
-        catch (Exception ex)
-        {
-            _logger.LogError(ex, "下单失败: {Symbol} {Side}", symbol, side);
-            return new TradeResult { Success = false, ErrorMessage = ex.Message };
-        }
+        var strategyId = TradingContext.CurrentStrategyId ?? "manual";
+        return await _tradeExecutor.ExecuteOrderAsync(
+            symbol, side, type, quantity, effectivePrice,
+            type == OrderType.Limit ? price : null,
+            strategyId: strategyId);
     }
 
     [Description("查询指定订单的状态")]
@@ -225,13 +184,4 @@ public IEnumerable<AIFunction> GetFunctions()
         yield return AIFunctionFactory.Create(GetOrderStatusAsync);
         yield return AIFunctionFactory.Create(CancelOrderAsync);
     }
-
-    private static TradeRecordStatus MapOrderStatus(string binanceStatus) => binanceStatus switch
-    {
-        "FILLED" => TradeRecordStatus.Filled,
-        "PARTIALLY_FILLED" => TradeRecordStatus.PartiallyFilled,
-        "CANCELED" or "CANCELLED" => TradeRecordStatus.Cancelled,
-        "REJECTED" or "EXPIRED" => TradeRecordStatus.Failed,
-        _ => TradeRecordStatus.Pending
-    };
 }

src/Trading/MarketMonitor.cs

@@ -1,3 +1,5 @@
+using System.Collections.Concurrent;
+using System.Threading.Channels;
 using MarketAssistant.Agents.Trading;
 using MarketAssistant.Infrastructure.Factories;
 using MarketAssistant.Services.Data;
@@ -9,7 +11,8 @@
 namespace MarketAssistant.Trading;
 
 /// <summary>
-/// 后台市场监控器，订阅实时价格并根据策略触发交易
+/// 后台市场监控器，订阅实时价格并根据策略触发交易。
+/// 使用 Channel 缓冲价格更新，确保不丢弃任何 tick。
 /// </summary>
 public class MarketMonitor : IDisposable
 {
@@ -22,7 +25,17 @@ public class MarketMonitor : IDisposable
 
     private CancellationTokenSource? _cts;
     private bool _isRunning;
-    private readonly SemaphoreSlim _evaluationLock = new(1, 1);
+    private Task? _consumerTask;
+
+    private readonly Channel<(string Symbol, decimal Price)> _priceChannel =
+        Channel.CreateBounded<(string, decimal)>(new BoundedChannelOptions(10000)
+        {
+            FullMode = BoundedChannelFullMode.DropOldest,
+            SingleReader = true,
+            SingleWriter = false
+        });
+
+    private readonly ConcurrentDictionary<string, SemaphoreSlim> _strategyLocks = new();
 
     public bool IsRunning => _isRunning;
 
@@ -70,11 +83,10 @@ public async Task StartAsync()
             .ToList();
 
         if (symbols.Count > 0)
-        {
             await _webSocketService.SubscribeAsync(symbols);
-        }
 
         _webSocketService.PriceUpdated += OnPriceUpdated;
+        _consumerTask = Task.Run(() => ConsumePriceUpdatesAsync(_cts.Token));
 
         _logger.LogInformation("MarketMonitor 已启动，监控 {Count} 个交易对", symbols.Count);
         StatusChanged?.Invoke(true);
@@ -91,6 +103,12 @@ public async Task StopAsync()
         _webSocketService.PriceUpdated -= OnPriceUpdated;
         _cts?.Cancel();
 
+        if (_consumerTask != null)
+        {
+            try { await _consumerTask; }
+            catch (OperationCanceledException) { }
+        }
+
         await _webSocketService.UnsubscribeAllAsync();
 
         _isRunning = false;
@@ -106,56 +124,77 @@ public async Task RefreshSubscriptionsAsync()
         if (!_isRunning)
             return;
 
-        await _webSocketService.UnsubscribeAllAsync();
-
         var activeStrategies = await _dataService.GetStrategiesByStatusAsync(StrategyStatus.Active);
-        var symbols = activeStrategies
+        var newSymbols = activeStrategies
             .Select(s => s.Symbol.ToLowerInvariant())
             .Distinct()
-            .ToList();
+            .ToHashSet();
 
-        if (symbols.Count > 0)
-        {
-            await _webSocketService.SubscribeAsync(symbols);
-        }
+        await _webSocketService.UnsubscribeAllAsync();
 
-        _logger.LogInformation("已刷新监控列表: {Count} 个交易对", symbols.Count);
+        if (newSymbols.Count > 0)
+            await _webSocketService.SubscribeAsync(newSymbols.ToList());
+
+        _logger.LogInformation("已刷新监控列表: {Count} 个交易对", newSymbols.Count);
     }
 
     private void OnPriceUpdated(string symbol, decimal lastPrice, decimal changePercent)
     {
-        if (_cts?.IsCancellationRequested == true)
-            return;
-
-        _ = ProcessPriceUpdateAsync(symbol, lastPrice);
+        _priceChannel.Writer.TryWrite((symbol, lastPrice));
     }
 
-    private async Task ProcessPriceUpdateAsync(string symbol, decimal lastPrice)
+    /// <summary>
+    /// Channel 消费者：顺序评估策略，异步执行触发的交易（每策略独立锁）
+    /// </summary>
+    private async Task ConsumePriceUpdatesAsync(CancellationToken ct)
     {
-        if (!await _evaluationLock.WaitAsync(0))
-            return;
-
         try
         {
-            var triggered = await _strategyEngine.EvaluateStrategiesAsync(
-                symbol, lastPrice, _cts?.Token ?? CancellationToken.None);
-
-            foreach (var strategy in triggered)
+            await foreach (var (symbol, price) in _priceChannel.Reader.ReadAllAsync(ct))
             {
-                await HandleTriggeredStrategyAsync(strategy, lastPrice);
+                try
+                {
+                    var triggered = await _strategyEngine.EvaluateStrategiesAsync(symbol, price, ct);
+                    foreach (var strategy in triggered)
+                        _ = ExecuteWithStrategyLockAsync(strategy, price, ct);
+                }
+                catch (OperationCanceledException) { throw; }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, "价格更新处理异常: {Symbol}", symbol);
+                }
             }
         }
         catch (OperationCanceledException)
         {
-            // 正常取消，忽略
+            _logger.LogDebug("价格消费者已取消");
+        }
+    }
+
+    /// <summary>
+    /// 带策略级锁的异步执行，防止同一策略并发触发
+    /// </summary>
+    private async Task ExecuteWithStrategyLockAsync(
+        TradingStrategy strategy, decimal price, CancellationToken ct)
+    {
+        var strategyLock = _strategyLocks.GetOrAdd(strategy.Id, _ => new SemaphoreSlim(1, 1));
+        if (!await strategyLock.WaitAsync(0, ct))
+        {
+            _logger.LogDebug("策略 {Id} 正在执行中，跳过本次触发", strategy.Id);
+            return;
+        }
+
+        try
+        {
+            await HandleTriggeredStrategyAsync(strategy, price);
         }
         catch (Exception ex)
         {
-            _logger.LogError(ex, "价格更新处理异常: {Symbol}", symbol);
+            _logger.LogError(ex, "策略执行异常: {StrategyId}", strategy.Id);
         }
         finally
         {
-            _evaluationLock.Release();
+            strategyLock.Release();
         }
     }
 
@@ -167,54 +206,83 @@ private async Task HandleTriggeredStrategyAsync(TradingStrategy strategy, decima
         }
         else
         {
-            var result = await _tradeExecutor.ExecuteTradeAsync(strategy, currentPrice, ct: _cts?.Token ?? default);
+            var result = await _tradeExecutor.ExecuteTradeAsync(
+                strategy, currentPrice, ct: _cts?.Token ?? default);
+
             if (result.Success && result.Record != null)
-            {
                 TradeExecuted?.Invoke(result.Record);
-            }
 
-            if (strategy.MaxExecutions.HasValue &&
-                strategy.ExecutionCount + 1 >= strategy.MaxExecutions.Value)
-            {
-                await _dataService.UpdateStrategyStatusAsync(strategy.Id, StrategyStatus.Completed);
-            }
+            await CheckStrategyCompletionAsync(strategy);
         }
     }
 
     private async Task HandleAISignalAsync(TradingStrategy strategy, decimal currentPrice)
     {
         try
         {
+            TradingContext.CurrentStrategyId = strategy.Id;
+
             var agent = _agentFactory.CreateAgent();
             var prompt = $"""
                 分析交易对 {strategy.Symbol}，当前价格 {currentPrice}。
                 策略配置: {strategy.CustomParams ?? "无"}
                 请评估是否应该执行 {strategy.Side} 操作，数量 {strategy.Quantity}。
                 如果决定交易，请调用 PlaceOrder 工具执行。
+                如果决定不交易，请说明理由。
                 """;
 
             var messages = new List<ChatMessage>
             {
                 new(ChatRole.User, prompt)
             };
 
-            await _dataService.UpdateStrategyTriggeredAsync(strategy.Id);
-
             var response = await agent.RunAsync(messages, session: null, options: null,
                 cancellationToken: _cts?.Token ?? default);
             _logger.LogDebug("TradingAgent 响应: {Content}", response.Text);
+
+            // 只在 Agent 实际执行了交易后才更新触发计数
+            var recentRecords = await _dataService.GetRecordsByStrategyAsync(strategy.Id);
+            var hasNewTrade = recentRecords.Any(r =>
+                r.CreatedAt > (strategy.LastTriggeredAt ?? DateTime.MinValue));
+
+            if (hasNewTrade)
+            {
+                await _dataService.UpdateStrategyTriggeredAsync(strategy.Id);
+                await CheckStrategyCompletionAsync(strategy);
+            }
         }
         catch (Exception ex)
         {
             _logger.LogError(ex, "AI 信号策略执行失败: {StrategyId}", strategy.Id);
         }
+        finally
+        {
+            TradingContext.CurrentStrategyId = null;
+        }
+    }
+
+    private async Task CheckStrategyCompletionAsync(TradingStrategy strategy)
+    {
+        if (!strategy.MaxExecutions.HasValue)
+            return;
+
+        var updated = await _dataService.GetStrategyAsync(strategy.Id);
+        if (updated != null && updated.ExecutionCount >= updated.MaxExecutions!.Value)
+        {
+            await _dataService.UpdateStrategyStatusAsync(strategy.Id, StrategyStatus.Completed);
+            _strategyEngine.ClearPeakPrice(strategy.Id);
+            _strategyLocks.TryRemove(strategy.Id, out _);
+        }
     }
 
     public void Dispose()
     {
         _cts?.Cancel();
         _cts?.Dispose();
-        _evaluationLock.Dispose();
+        _priceChannel.Writer.TryComplete();
+        foreach (var kvp in _strategyLocks)
+            kvp.Value.Dispose();
+        _strategyLocks.Clear();
         GC.SuppressFinalize(this);
     }
 }

src/Trading/Models/TradingModels.cs

@@ -141,10 +141,27 @@ public class OrderStatusInfo
 public class RiskCheckResult
 {
     public bool Passed { get; set; }
+    public bool NeedsConfirmation { get; set; }
     public string? Reason { get; set; }
 
     public static RiskCheckResult Pass() => new() { Passed = true };
     public static RiskCheckResult Reject(string reason) => new() { Passed = false, Reason = reason };
+    public static RiskCheckResult RequireConfirmation(string reason) =>
+        new() { Passed = false, NeedsConfirmation = true, Reason = reason };
+}
+
+/// <summary>
+/// 交易上下文，用于在 Agent 工具调用链中传递当前策略 ID
+/// </summary>
+public static class TradingContext
+{
+    private static readonly AsyncLocal<string?> _strategyId = new();
+
+    public static string? CurrentStrategyId
+    {
+        get => _strategyId.Value;
+        set => _strategyId.Value = value;
+    }
 }
 
 #endregion