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.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 BuyStops = 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(); _ = BackgroundWorker(); } private async Task InitStops() { using var context = await _dbContextFactory.CreateDbContextAsync(); context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking; var dt = DateTime.UtcNow.AddMinutes(-_buyStopLength); var stops = await context.Declisions.Where(d => d.Time > dt && d.Action == DeclisionTradeAction.StopBuy).ToArrayAsync(); foreach (var stop in stops) { var time = stop.Time.AddMinutes(_buyStopLength); BuyStops.TryAdd(stop.Figi, time); } } private async Task ProcessPrices() { while (await _pricesChannel.Reader.WaitToReadAsync()) { 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); // } // var data = await unit.GetData(); // var declisionsForSave = new List(); // if (message.Figi == "FUTIMOEXF000") // { // var result = MovingAverage.CheckByWindowAverageMean(data.timestamps, data.prices, 100, 3f); // if ((result & TradingEvent.StopBuy) == TradingEvent.StopBuy) // { // var stopTo = DateTime.UtcNow.AddMinutes(_buyStopLength); // BuyStops.AddOrUpdate(message.Figi, stopTo, (k, v) => stopTo); // declisionsForSave.Add(new Declision() // { // AccountId = string.Empty, // Figi = message.Figi, // Ticker = message.Ticker, // Price = message.Value, // Time = message.IsHistoricalData ? message.Time : DateTime.UtcNow, // Action = DeclisionTradeAction.StopBuy, // }); // } // if ((result & TradingEvent.LongOpen) == TradingEvent.LongOpen // && !BuyStops.TryGetValue(message.Figi, out _)) // { // var stopTo = DateTime.UtcNow.AddMinutes(_buyStopLength); // BuyStops.AddOrUpdate(message.Figi, stopTo, (k, v) => stopTo); // declisionsForSave.Add(new Declision() // { // AccountId = string.Empty, // Figi = message.Figi, // Ticker = message.Ticker, // Price = message.Value, // Time = message.IsHistoricalData ? message.Time : DateTime.UtcNow, // Action = DeclisionTradeAction.OpenLong, // }); // } // if ((result & TradingEvent.LongClose) == TradingEvent.LongClose) // { // declisionsForSave.Add(new Declision() // { // AccountId = string.Empty, // Figi = message.Figi, // Ticker = message.Ticker, // Price = message.Value, // Time = message.IsHistoricalData ? message.Time : DateTime.UtcNow, // Action = DeclisionTradeAction.CloseLong, // }); // } // if ((result & TradingEvent.ShortOpen) == TradingEvent.ShortOpen && (unit.AsksCount/ unit.BidsCount>2 )) // { // declisionsForSave.Add(new Declision() // { // AccountId = string.Empty, // Figi = message.Figi, // Ticker = message.Ticker, // Price = message.Value, // Time = message.IsHistoricalData ? message.Time : DateTime.UtcNow, // Action = DeclisionTradeAction.OpenShort, // }); // } // using var context = await _dbContextFactory.CreateDbContextAsync(); // context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking; // await context.AddRangeAsync(declisionsForSave); // await context.SaveChangesAsync(); // } //} } } 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); } } private async Task BackgroundWorker() { var keysForRemove = new List(); while (!_cts.IsCancellationRequested) { var time = DateTime.UtcNow; foreach (var kvp in BuyStops) { if (kvp.Value > time) { keysForRemove.Add(kvp.Key); } } foreach (var key in keysForRemove) { BuyStops.TryRemove(key, out _); } await Task.Delay(10000); } } 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; } } }