using KLHZ.Trader.Core.Common; using KLHZ.Trader.Core.Contracts.Declisions.Dtos.Enums; using KLHZ.Trader.Core.Contracts.Declisions.Interfaces; using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Enums; using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Interfaces; using KLHZ.Trader.Core.Contracts.Messaging.Interfaces; using KLHZ.Trader.Core.DataLayer; using KLHZ.Trader.Core.DataLayer.Entities.Declisions; using KLHZ.Trader.Core.DataLayer.Entities.Declisions.Enums; using KLHZ.Trader.Core.DataLayer.Entities.Prices; using KLHZ.Trader.Core.Exchange.Extentions; using KLHZ.Trader.Core.Exchange.Models; using KLHZ.Trader.Core.Math.Declisions.Services.Cache; using KLHZ.Trader.Core.Math.Declisions.Utils; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using System.Collections.Concurrent; using System.Threading.Channels; using Tinkoff.InvestApi; using AssetType = KLHZ.Trader.Core.Exchange.Models.AssetType; namespace KLHZ.Trader.Core.Exchange.Services { public class Trader : IHostedService { private readonly InvestApiClient _investApiClient; private readonly IServiceProvider _provider; private readonly IDataBus _dataBus; private readonly BotModeSwitcher _botModeSwitcher; private readonly IDbContextFactory _dbContextFactory; private readonly ConcurrentDictionary OpeningStops = new(); private readonly ConcurrentDictionary Accounts = new(); private readonly ConcurrentDictionary _historyCash = new(); private readonly ILogger _logger; private readonly double _buyStopLength; private readonly decimal _futureComission; private readonly decimal _shareComission; private readonly decimal _accountCashPart; private readonly decimal _accountCashPartFutures; private readonly decimal _defaultBuyPartOfAccount; private readonly string[] _managedAccountsNamePatterns = []; private readonly string[] _tradingInstrumentsFigis = []; private readonly Channel _pricesChannel = Channel.CreateUnbounded(); private readonly Channel _ordersbookChannel = Channel.CreateUnbounded(); private readonly CancellationTokenSource _cts = new(); public Trader( ILogger logger, BotModeSwitcher botModeSwitcher, IServiceProvider provider, IOptions options, IDataBus dataBus, IDbContextFactory dbContextFactory, InvestApiClient investApiClient) { _logger = logger; _botModeSwitcher = botModeSwitcher; _dataBus = dataBus; _provider = provider; _investApiClient = investApiClient; _managedAccountsNamePatterns = options.Value.ManagingAccountNamePatterns.ToArray(); _dbContextFactory = dbContextFactory; _futureComission = options.Value.FutureComission; _shareComission = options.Value.ShareComission; _accountCashPart = options.Value.AccountCashPart; _accountCashPartFutures = options.Value.AccountCashPartFutures; _defaultBuyPartOfAccount = options.Value.DefaultBuyPartOfAccount; _tradingInstrumentsFigis = options.Value.TradingInstrumentsFigis; _buyStopLength = (double)options.Value.StopBuyLengthMinuts; } public async Task StartAsync(CancellationToken cancellationToken) { //await InitStops(); var accounts = await _investApiClient.GetAccounts(_managedAccountsNamePatterns); var accountsList = new List(); int i = 0; foreach (var accountId in accounts) { var acc = _provider.GetKeyedService(i); if (acc != null) { await acc.Init(accountId); Accounts[accountId] = acc; i++; } else { break; } } _dataBus.AddChannel(nameof(Trader), _pricesChannel); _dataBus.AddChannel(nameof(Trader), _ordersbookChannel); _ = ProcessPrices(); _ = ProcessOrdersbooks(); } private async Task ProcessPrices() { var declisionsForSave = new List(); var processedPrices = new List(); while (await _pricesChannel.Reader.WaitToReadAsync()) { var bigWindowProcessor = nameof(Trader) + "_big"; var smallWindowProcessor = nameof(Trader) + "_small"; var message = await _pricesChannel.Reader.ReadAsync(); if (_tradingInstrumentsFigis.Contains(message.Figi)) { if (_historyCash.TryGetValue(message.Figi, out var unit)) { await unit.AddData(message); } else { unit = new PriceHistoryCacheUnit2(message.Figi, message); _historyCash.TryAdd(message.Figi, unit); } try { if (message.Figi == "FUTIMOEXF000") { var data = await unit.GetData(); if (OpeningStops.TryGetValue(message.Figi, out var dt)) { if (dt < (message.IsHistoricalData ? message.Time : DateTime.UtcNow)) { OpeningStops.TryRemove(message.Figi, out _); } } if ((unit.BidsCount / unit.AsksCount) < 0.5m || (unit.BidsCount / unit.AsksCount) > 2m) { var stopTo = (message.IsHistoricalData ? message.Time : DateTime.UtcNow).AddMinutes(3); OpeningStops.AddOrUpdate(message.Figi, stopTo, (k, v) => stopTo); LogDeclision(declisionsForSave, DeclisionTradeAction.StopBuyShortTime, message); } var resultLongOpen = MovingAverage.CheckByWindowAverageMean(data.timestamps, data.prices, 100, 45, 180, 2.5m); var resultLongClose = MovingAverage.CheckByWindowAverageMean(data.timestamps, data.prices, 100, 15, 120, 2.5m); if (resultLongOpen.bigWindowAv != 0) { LogPrice(processedPrices, message, bigWindowProcessor, resultLongOpen.bigWindowAv); LogPrice(processedPrices, message, smallWindowProcessor, resultLongOpen.smallWindowAv); } if ((resultLongClose.events & TradingEvent.StopBuy) == TradingEvent.StopBuy) { var stopTo = (message.IsHistoricalData ? message.Time : DateTime.UtcNow).AddMinutes(_buyStopLength); OpeningStops.AddOrUpdate(message.Figi, stopTo, (k, v) => stopTo); //LogDeclision(declisionsForSave, DeclisionTradeAction.StopBuy, message); } if ((resultLongOpen.events & TradingEvent.UptrendStart) == TradingEvent.UptrendStart && !OpeningStops.TryGetValue(message.Figi, out _)) { LogDeclision(declisionsForSave, DeclisionTradeAction.OpenLong, message); } if ((resultLongClose.events & TradingEvent.UptrendEnd) == TradingEvent.UptrendEnd) { LogDeclision(declisionsForSave, DeclisionTradeAction.CloseLong, message); } //if ((resultLongOpen.events & TradingEvent.ShortOpen) == TradingEvent.ShortOpen // && !OpeningStops.TryGetValue(message.Figi, out _)) //{ // LogDeclision(declisionsForSave, DeclisionTradeAction.OpenShort, message); //} //if ((resultLongOpen.events & TradingEvent.ShortClose) == TradingEvent.ShortClose) //{ // LogDeclision(declisionsForSave, DeclisionTradeAction.CloseShort, message); //} if ((!message.IsHistoricalData && (processedPrices.Count > 0 || declisionsForSave.Count > 0)) || (message.IsHistoricalData && ((processedPrices.Count + declisionsForSave.Count > 10000) || _pricesChannel.Reader.Count == 0))) { using var context = await _dbContextFactory.CreateDbContextAsync(); context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking; if (processedPrices.Count > 0) { await context.ProcessedPrices.AddRangeAsync(processedPrices); processedPrices.Clear(); } if (declisionsForSave.Count > 0) { await context.Declisions.AddRangeAsync(declisionsForSave); declisionsForSave.Clear(); } await context.SaveChangesAsync(); } } } catch (Exception ex) { _logger.LogError(ex, "Ошибка при боработке новой цены IMOEXF"); } } } } private static void LogPrice(List prices, INewPrice message, string processor, decimal value) { prices.Add(new ProcessedPrice() { Figi = message.Figi, Ticker = message.Ticker, Processor = processor, Time = message.Time, Value = value, }); } private static void LogDeclision(List declisions, DeclisionTradeAction action, INewPrice message) { declisions.Add(new Declision() { AccountId = string.Empty, Figi = message.Figi, Ticker = message.Ticker, Price = message.Value, Time = message.IsHistoricalData ? message.Time : DateTime.UtcNow, Action = action, }); } private async Task ProcessOrdersbooks() { while (await _ordersbookChannel.Reader.WaitToReadAsync()) { var message = await _ordersbookChannel.Reader.ReadAsync(); if (!_historyCash.TryGetValue(message.Figi, out var data)) { data = new PriceHistoryCacheUnit2(message.Figi); _historyCash.TryAdd(message.Figi, data); } await data.AddOrderbook(message); } } public Task StopAsync(CancellationToken cancellationToken) { _cts.Cancel(); return Task.CompletedTask; } private decimal GetComission(AssetType assetType) { if (assetType == AssetType.Common) { return _shareComission; } else if (assetType == AssetType.Futures) { return _futureComission; } else { return 0; } } private decimal GetCount(string accountId, decimal boutPrice) { var balance = Accounts[accountId].Balance; return System.Math.Floor(balance * _defaultBuyPartOfAccount / boutPrice); } private bool IsBuyAllowed(string accountId, decimal boutPrice, decimal count, bool needBigCash) { if (!_botModeSwitcher.CanPurchase()) return false; var balance = Accounts[accountId].Balance; var total = Accounts[accountId].Total; var futures = Accounts[accountId].Assets.Values.FirstOrDefault(v => v.Type == AssetType.Futures); if (futures != null || needBigCash) { if ((balance - boutPrice * count) / total < _accountCashPartFutures) return false; } else { if ((balance - boutPrice * count) / total < _accountCashPart) return false; } return true; } private bool IsSellAllowed(AssetType assetType, PositionType positionType, decimal boutPrice, decimal? requiredPrice, TradeCommandType commandType) { if (commandType >= TradeCommandType.MarketSell && commandType < TradeCommandType.ForceClosePosition && requiredPrice.HasValue) { var comission = GetComission(assetType); if (positionType == PositionType.Long) { return requiredPrice.Value * (1 - comission) > boutPrice * (1 + comission); } else if (positionType == PositionType.Short) { return requiredPrice.Value * (1 + comission) < boutPrice * (1 - comission); } } if (commandType == TradeCommandType.ForceClosePosition) return true; return false; } } }