klhztrader/KLHZ.Trader.Core/Exchange/Services/TradingCommandsExecutor.cs

using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Interfaces;
using KLHZ.Trader.Core.Contracts.Messaging.Interfaces;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Threading.Channels;
using Tinkoff.InvestApi;
using Tinkoff.InvestApi.V1;

namespace KLHZ.Trader.Core.Exchange.Services
{
    public class TradingCommandsExecutor : IHostedService
    {
        private readonly TraderDataProvider _tradeDataProvider;
        private readonly InvestApiClient _investApiClient;
        private readonly IDataBus _dataBus;
        private readonly ILogger<TradingCommandsExecutor> _logger;
        private readonly Channel<ITradeCommand> _channel = Channel.CreateUnbounded<ITradeCommand>();

        public TradingCommandsExecutor(InvestApiClient investApiClient, IDataBus dataBus, ILogger<TradingCommandsExecutor> logger, TraderDataProvider tradeDataProvider)
        {
            _investApiClient = investApiClient;
            _dataBus = dataBus;
            _dataBus.AddChannel(nameof(TradingCommandsExecutor), _channel);
            _logger = logger;
            _tradeDataProvider = tradeDataProvider;
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            _ = ProcessCommands();
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }

        internal async Task ExecuteCommand(ITradeCommand tradeCommand)
        {
            try
            {
                var dir = OrderDirection.Unspecified;
                var orderType = OrderType.Unspecified;
                if (tradeCommand.CommandType == Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketBuy)
                {
                    dir = OrderDirection.Buy;
                    orderType = OrderType.Market;
                }
                else if (tradeCommand.CommandType == Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketSell)
                {
                    dir = OrderDirection.Sell;
                    orderType = OrderType.Market;
                }
                else if (tradeCommand.CommandType == Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitBuy && tradeCommand.RecomendPrice.HasValue)
                {
                    dir = OrderDirection.Buy;
                    orderType = OrderType.Limit;
                }

                var req = new PostOrderRequest()
                {
                    AccountId = tradeCommand.AccountId,
                    InstrumentId = tradeCommand.Figi,
                    Direction = dir,
                    Price = tradeCommand.RecomendPrice ?? 0,
                    OrderType = orderType,
                    Quantity = tradeCommand.Count,
                    ConfirmMarginTrade = tradeCommand.EnableMargin,
                };

                var res = await _investApiClient.Orders.PostOrderAsync(req);

                //var result = new DealResult
                //{
                //    Count = sign * res.LotsExecuted,
                //    Price = res.ExecutedOrderPrice,
                //    Success = true,
                //    Direction = dealDirection,
                //    AccountId = tradeCommand.AccountId,
                //    Figi = tradeCommand.Figi,
                //};
                //await _tradeDataProvider.LogDeal(result);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Ошибка при покупке актива на счёт {acc}. figi: {figi}", tradeCommand.AccountId, tradeCommand.Figi);
            }
        }

        private async Task ProcessCommands()
        {
            while (await _channel.Reader.WaitToReadAsync())
            {
                var command = await _channel.Reader.ReadAsync();
                await ExecuteCommand(command);
            }
        }
    }
}