using KLHZ.Trader.Core.Common; using KLHZ.Trader.Core.Contracts.Common.Enums; using KLHZ.Trader.Core.Contracts.Declisions.Dtos.Enums; using KLHZ.Trader.Core.Contracts.Messaging.Dtos; 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.Entities.Declisions.Enums; using KLHZ.Trader.Core.Exchange.Interfaces; using KLHZ.Trader.Core.Exchange.Models.AssetsAccounting; using KLHZ.Trader.Core.Exchange.Models.Configs; using KLHZ.Trader.Core.Exchange.Models.Trading; using KLHZ.Trader.Core.Exchange.Utils; using KLHZ.Trader.Core.Math.Declisions.Utils; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using System.Collections.Concurrent; using System.Collections.Immutable; using System.Security.Cryptography; using System.Threading.Channels; using Tinkoff.InvestApi; namespace KLHZ.Trader.Core.Exchange.Services { public class Trader : IHostedService { private readonly IDataBus _dataBus; private readonly TraderDataProvider _tradeDataProvider; private readonly PortfolioWrapper _portfolioWrapper; private readonly ExchangeConfig _exchangeConfig; private readonly ILogger _logger; private readonly ConcurrentDictionary TradingModes = new(); private readonly ConcurrentDictionary SupportLevels = new(); private readonly ConcurrentDictionary DPirsonValues = new(); private readonly ConcurrentDictionary _supportLevelsCalculationTimes = new(); private readonly Channel _pricesChannel = Channel.CreateUnbounded(); private readonly Channel _commands = Channel.CreateUnbounded(); private readonly Channel _orderbooks = Channel.CreateUnbounded(); public Trader( ILogger logger, IOptions options, IDataBus dataBus, PortfolioWrapper portfolioWrapper, TraderDataProvider tradeDataProvider, InvestApiClient investApiClient) { _portfolioWrapper = portfolioWrapper; _tradeDataProvider = tradeDataProvider; _logger = logger; _dataBus = dataBus; _exchangeConfig = options.Value; foreach (var f in _exchangeConfig.TradingInstrumentsFigis) { TradingModes[f] = TradingMode.None; } } public Task StartAsync(CancellationToken cancellationToken) { _dataBus.AddChannel(nameof(Trader), _pricesChannel); _dataBus.AddChannel(nameof(Trader), _orderbooks); _dataBus.AddChannel(nameof(Trader), _commands); _ = ProcessPrices(); _ = ProcessOrderbooks(); _ = ProcessCommands(); return Task.CompletedTask; } private async Task ProcessCommands() { while (await _commands.Reader.WaitToReadAsync()) { var command = await _commands.Reader.ReadAsync(); try { if (command.CommandType == TradeCommandType.OpenLong || command.CommandType == TradeCommandType.OpenShort) { var fakeMessage = new TradeDataItem() { Figi = command.Figi, Ticker = "", Count = command.Count, Direction = 1, IsHistoricalData = false, Time = DateTime.UtcNow, Price = command.RecomendPrice ?? 0m }; var positionType = command.CommandType == TradeCommandType.OpenLong ? PositionType.Long : PositionType.Short; var st = GetStops(fakeMessage); var stops = st.GetStops(positionType); var accounts = _portfolioWrapper.Accounts .Where(a => !a.Value.Assets.ContainsKey(command.Figi)) .Take(1) .Select(a => a.Value) .ToArray(); await OpenPositions(accounts, fakeMessage, positionType, stops.stopLoss, stops.takeProfit, System.Math.Abs(command.Count)); } else { var fakeMessage = new TradeDataItem() { Figi = command.Figi, Ticker = "", Count = command.Count, Direction = 1, IsHistoricalData = false, Time = DateTime.UtcNow, Price = command.RecomendPrice ?? 0m }; var assetsForClose = _portfolioWrapper.Accounts .SelectMany(a => a.Value.Assets.Values) .Where(a => a.Figi == fakeMessage.Figi) .ToArray(); await ClosePositions(assetsForClose, fakeMessage, false); } } catch (Exception ex) { _logger.LogError(ex, "Ошибка при выполнении команды."); } } } private async Task ProcessOrderbooks() { while (await _orderbooks.Reader.WaitToReadAsync()) { var message = await _orderbooks.Reader.ReadAsync(); await _tradeDataProvider.AddOrderbook(message); } } private async Task ProcessPrices() { var pricesCache1 = new Dictionary>(); var pricesCache2 = new Dictionary>(); while (await _pricesChannel.Reader.WaitToReadAsync()) { var message = await _pricesChannel.Reader.ReadAsync(); if (!message.IsHistoricalData && DateTime.UtcNow - message.Time > TimeSpan.FromMinutes(1)) { continue; } await CloseMarginPositionsIfNeed(message); try { if (message.IsHistoricalData) { message = TraderUtils.FilterHighFreqValues(message, message.Direction == 1 ? pricesCache1 : pricesCache2); } if (_exchangeConfig.TradingInstrumentsFigis.Contains(message.Figi) && message.Direction == 1) { await _tradeDataProvider.AddData(message); if (message.Figi == "FUTIMOEXF000") { await ProcessIMOEXF(message); } } } catch (Exception ex) { _logger.LogError(ex, "Ошибка при боработке новой цены."); } } } private async Task ProcessIMOEXF(ITradeDataItem message) { if (message.Figi == "FUTIMOEXF000") { await CalcSupportLevels(message, 3, 5); var stops = GetStops(message); var pirson = await CalcPirson(message); var declisionPirson = await ProcessPirson(pirson, message); var declisionsSupportLevels = await ProcessSupportLevels(message); var declisionsStops = ProcessStops(stops, 1.5m); var res = TraderUtils.MergeResultsMult(declisionPirson, declisionsSupportLevels); res = TraderUtils.MergeResultsMult(res, declisionsStops); await ExecuteDeclisions(res.ToImmutableDictionary(), message, stops, 1); } } private async Task> ProcessPirson(PirsonCalculatingResult pirson, ITradeDataItem message) { var res = TraderUtils.GetInitDict(0); if (pirson.Success && DPirsonValues.TryGetValue(message.Figi, out var olddpirs)) { if (olddpirs < -0.3m && pirson.Pirson > -0.3m && pirson.PriceDiff > 0 && (pirson.TradesDiff > 0)) { res[TradingEvent.OpenLong] = Constants.PowerUppingCoefficient; await _tradeDataProvider.LogPrice(message, "diffs_pirson_diff_point_long_in", message.Price); } if (olddpirs > 0.3m && pirson.Pirson < 0.3m && pirson.PriceDiff < 0 && (pirson.TradesDiff > 0)) { res[TradingEvent.OpenShort] = Constants.PowerUppingCoefficient; await _tradeDataProvider.LogPrice(message, "diffs_pirson_diff_point_short_in", message.Price); } if (olddpirs > 0.7m && pirson.Pirson < 0.7m) { // await _tradeDataProvider.LogPrice(message, "diffs_pirson_diff_point_long_out", message.Price); } if (olddpirs < -0.7m && pirson.Pirson > -0.7m) { // await _tradeDataProvider.LogPrice(message, "diffs_pirson_diff_point_short_out", message.Price); } } DPirsonValues[message.Figi] = pirson.Pirson; return res.ToImmutableDictionary(); } private async Task CalcPirson(ITradeDataItem message) { var cacheSize = TimeSpan.FromSeconds(400); var smallWindow = TimeSpan.FromSeconds(180); var bigWindow = TimeSpan.FromSeconds(360); var meanWindowForCottelation = TimeSpan.FromSeconds(360); var currentTime = message.IsHistoricalData ? message.Time : DateTime.UtcNow; var buys = await _tradeDataProvider.GetDataForTimeWindow(message.Figi, cacheSize, selector: (i) => i.Direction == 1); var trades = await _tradeDataProvider.GetDataForTimeWindow(message.Figi, cacheSize); if (trades.TryCalcTimeWindowsDiff(bigWindow, smallWindow, v => v.Count, false, out var tradesDiff) && buys.TryCalcTimeDiff(bigWindow, smallWindow, v => v.Price, true, out var pricesDiff)) { await _tradeDataProvider.LogPrice(message, "privcesDiff", pricesDiff); await _tradeDataProvider.LogPrice(message, "tradevolume_diff", tradesDiff); await _tradeDataProvider.AddData(message.Figi, "5min_diff", new Contracts.Declisions.Dtos.CachedValue() { Time = message.Time, Value2 = tradesDiff, Value = pricesDiff, Figi = message.Figi, Ticker = message.Ticker, }); var diffs = await _tradeDataProvider.GetDataForTimeWindow(message.Figi, cacheSize, "5min_diff"); if (diffs.TryCalcPirsonCorrelation(meanWindowForCottelation, out var pirson)) { var res = pirson; await _tradeDataProvider.LogPrice(message, "diffs_pirson", (decimal)pirson); //await _tradeDataProvider.AddData(message.Figi, "diffs_pirson", new Contracts.Declisions.Dtos.CachedValue() //{ // Time = message.Time, // Value = (decimal)pirson, // Figi = message.Figi, // Ticker = message.Ticker, //}); return new PirsonCalculatingResult() { Pirson = res, PriceDiff = pricesDiff, TradesDiff = tradesDiff, Success = true, }; } } return new PirsonCalculatingResult() { Success = false, }; } private async Task CloseMarginPositionsIfNeed(ITradeDataItem message) { var state = ExchangeScheduler.GetCurrentState(message.Time); if (!message.IsHistoricalData && state == ExchangeState.ClearingTime) { var futuresFigis = _portfolioWrapper.Accounts.Values.SelectMany(v => v.Assets.Values.Where(a => a.Type == AssetType.Futures)).ToArray(); await ClosePositions(futuresFigis, message, false); } } private async Task CalcSupportLevels(ITradeDataItem message, int leverage, decimal supportLevelWidth, int depthHours = 3) { if (_supportLevelsCalculationTimes.TryGetValue(message.Figi, out var lastTime)) { if ((message.Time - lastTime).TotalMinutes < 30) { return; } } var data = await _tradeDataProvider.GetDataForTimeWindow(message.Figi, TimeSpan.FromHours(depthHours)); if (data.Length > 0) { if (data[^1].Time - data[0].Time < TimeSpan.FromHours(depthHours - 1)) { data = await _tradeDataProvider.GetDataForTimeWindow(message.Figi, TimeSpan.FromHours(depthHours + 12)); if (data[^1].Time - data[0].Time < TimeSpan.FromHours(depthHours - 1)) { return; } } var hist = Statistics.CalcHistogram(data); var convs = Statistics.CalcConvolution(hist, leverage).ToList(); var orderedConvs = convs.OrderByDescending(c => c.Sum).Take(5).ToList(); orderedConvs = [.. orderedConvs.OrderBy(c => c.Value)]; var levelsForAdd = new List(); foreach (var c in orderedConvs) { var low = c.Value - supportLevelWidth; var high = c.Value + supportLevelWidth; if (levelsForAdd.Count > 0) { var last = levelsForAdd.Last(); if (last.HighValue < low) { levelsForAdd.Add(new SupportLevel() { HighValue = high, LowValue = low, Value = c.Value, }); } else if (last.HighValue >= low && last.HighValue < high) { levelsForAdd[^1] = new SupportLevel() { LowValue = last.LowValue, HighValue = high, Value = last.LowValue + (high - last.LowValue) / 2 }; } } else { levelsForAdd.Add(new SupportLevel() { HighValue = high, LowValue = low, Value = c.Value, }); } } var finalLevels = new SupportLevel[levelsForAdd.Count]; var i = 0; foreach (var level in levelsForAdd) { DateTime? time = null; foreach (var item in data) { if (item.Price >= level.LowValue && item.Price < level.HighValue) { time = item.Time; } } finalLevels[i] = new SupportLevel() { HighValue = level.HighValue, LowValue = level.LowValue, Value = level.Value, LastLevelTime = time, }; i++; } SupportLevels[message.Figi] = finalLevels; await _tradeDataProvider.LogPrice(message, "support_level_calc", message.Price); } _supportLevelsCalculationTimes[message.Figi] = message.Time; } private async Task ClosePositions(Asset[] assets, ITradeDataItem message, bool withProfitOnly = true) { var loggedDeclisions = 0; var assetType = _tradeDataProvider.GetAssetTypeByFigi(message.Figi); var assetsForClose = new List(); var price = message.Price; if (price == 0) { price = await _tradeDataProvider.GetLastPrice(message.Figi); } price = System.Math.Round(price, 2); var messages = new List(); foreach (var asset in assets) { Asset? assetForClose = null; string? mess = null; if (withProfitOnly) { var profit = 0m; if (assetType == AssetType.Futures) { if (_tradeDataProvider.Orderbooks.TryGetValue(message.Figi, out var orderbook)) { if (asset.Count < 0 && orderbook.Asks.Length > 0) { price = orderbook.Asks[0].Price; } else if (orderbook.Bids.Length > 0) { price = orderbook.Bids[0].Price; } } profit = TradingCalculator.CaclProfit(asset.BoughtPrice, price, GetComission(assetType), GetLeverage(message.Figi, asset.Count < 0), asset.Count < 0); } if (profit > 0) { profit = System.Math.Round(profit, 2); assetForClose = asset; mess = $"Закрываю позицию {asset.Figi} ({(asset.Count > 0 ? "лонг" : "шорт")}) на счёте {_portfolioWrapper.Accounts[asset.AccountId].AccountName}. Количество {(long)asset.Count}, цена ~{price}, профит {profit}"; if (loggedDeclisions == 0) { loggedDeclisions++; await _tradeDataProvider.LogDeclision(asset.Count < 0 ? DeclisionTradeAction.CloseShortReal : DeclisionTradeAction.CloseLongReal, message, profit); } } } else { mess = $"Закрываю позицию {asset.Figi} ({(asset.Count > 0 ? "лонг" : "шорт")}) на счёте {_portfolioWrapper.Accounts[asset.AccountId].AccountName}. Количество {(long)asset.Count}, цена ~{price}"; assetForClose = asset; } if (assetForClose != null && mess != null) { await _portfolioWrapper.Accounts[asset.AccountId].ClosePosition(message.Figi); await _dataBus.Broadcast(new MessageForAdmin() { Text = mess }); } } } private async Task OpenPositions(IManagedAccount[] accounts, ITradeDataItem message, PositionType positionType, decimal stopLossShift, decimal takeProfitShift, long count = 1) { var loggedDeclisions = 0; var sign = positionType == PositionType.Long ? 1 : 1; foreach (var acc in accounts) { if (TraderUtils.IsOperationAllowed(acc, message.Price, 1, _exchangeConfig.AccountCashPartFutures, _exchangeConfig.AccountCashPart)) { await acc.OpenPosition(message.Figi, positionType, stopLossShift, takeProfitShift, count); await _dataBus.Broadcast(new MessageForAdmin() { Text = $"Открываю позицию {message.Figi} ({(positionType == PositionType.Long ? "лонг" : "шорт")}) " + $"на счёте {acc.AccountName}. Количество {(positionType == PositionType.Long ? "" : "-")}{count}, " + $"цена ~{System.Math.Round(message.Price, 2)}. Стоп лосс: {(positionType == PositionType.Long ? "-" : "+")}{stopLossShift}. " + $"Тейк профит: {(positionType == PositionType.Long ? "+" : "-")}{takeProfitShift}" }); } if (loggedDeclisions == 0) { await _tradeDataProvider.LogDeclision(DeclisionTradeAction.OpenLongReal, message); loggedDeclisions++; } } } private async Task ExecuteDeclisions(ImmutableDictionary result, ITradeDataItem message, Stops st, int accountsForOpening = 1) { var state = ExchangeScheduler.GetCurrentState(); if (result[TradingEvent.OpenLong] >= Constants.UppingCoefficient && state == ExchangeState.Open ) { var stops = st.GetStops(PositionType.Long); if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase()) { var accounts = _portfolioWrapper.Accounts .Where(a => !a.Value.Assets.ContainsKey(message.Figi)) .Take(accountsForOpening) .Select(a => a.Value) .ToArray(); await OpenPositions(accounts, message, PositionType.Long, stops.stopLoss, stops.takeProfit, 1); } await _tradeDataProvider.LogDeclision(DeclisionTradeAction.OpenLong, message.Price, message.Time.AddMilliseconds(RandomNumberGenerator.GetInt32(-100, 100)), message); await _tradeDataProvider.LogDeclision(DeclisionTradeAction.ResetStopsLong, message.Price + stops.takeProfit, message.Time.AddMilliseconds(-RandomNumberGenerator.GetInt32(300, 1000)), message); await _tradeDataProvider.LogDeclision(DeclisionTradeAction.ResetStopsLong, message.Price - stops.stopLoss, message.Time.AddMilliseconds(RandomNumberGenerator.GetInt32(300, 1000)), message); } if (result[TradingEvent.OpenShort] >= Constants.UppingCoefficient && state == ExchangeState.Open ) { var stops = st.GetStops(PositionType.Long); if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase()) { var accounts = _portfolioWrapper.Accounts .Where(a => !a.Value.Assets.ContainsKey(message.Figi)) .Take(1) .Select(a => a.Value) .ToArray(); await OpenPositions(accounts, message, PositionType.Short, stops.stopLoss, stops.takeProfit, 1); } await _tradeDataProvider.LogDeclision(DeclisionTradeAction.OpenShort, message.Price, message.Time.AddMilliseconds(RandomNumberGenerator.GetInt32(-100, 100)), message); await _tradeDataProvider.LogDeclision(DeclisionTradeAction.ResetStopsShort, message.Price - stops.takeProfit, message.Time.AddMilliseconds(-RandomNumberGenerator.GetInt32(300, 1000)), message); await _tradeDataProvider.LogDeclision(DeclisionTradeAction.ResetStopsShort, message.Price + stops.stopLoss, message.Time.AddMilliseconds(RandomNumberGenerator.GetInt32(300, 1000)), message); } if (result[TradingEvent.CloseLong] >= Constants.UppingCoefficient * 10) { if (!message.IsHistoricalData && BotModeSwitcher.CanSell()) { var assetsForClose = _portfolioWrapper.Accounts .SelectMany(a => a.Value.Assets.Values) .Where(a => a.Figi == message.Figi && a.Count > 0) .ToArray(); await ClosePositions(assetsForClose, message); } await _tradeDataProvider.LogDeclision(DeclisionTradeAction.CloseLong, message.Price, message.Time.AddMilliseconds(RandomNumberGenerator.GetInt32(-100, 100)), message); } if (result[TradingEvent.CloseShort] >= Constants.UppingCoefficient * 10) { if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase()) { var assetsForClose = _portfolioWrapper.Accounts .SelectMany(a => a.Value.Assets.Values) .Where(a => a.Figi == message.Figi && a.Count < 0) .ToArray(); await ClosePositions(assetsForClose, message); } await _tradeDataProvider.LogDeclision(DeclisionTradeAction.CloseShort, message.Price, message.Time.AddMilliseconds(RandomNumberGenerator.GetInt32(-100, 100)), message); } } public Task StopAsync(CancellationToken cancellationToken) { return Task.CompletedTask; } private decimal GetComission(AssetType assetType) { if (assetType == AssetType.Common) { return _exchangeConfig.ShareComission; } else if (assetType == AssetType.Futures) { return _exchangeConfig.FutureComission; } else { return 0; } } private decimal GetLeverage(string figi, bool isShort) { var res = 1m; var leverage = _exchangeConfig.InstrumentsSettings.FirstOrDefault(l => l.Figi == figi); if (leverage != null) { res = isShort ? leverage.ShortLeverage : leverage.LongLeverage; } return res; } private Stops GetStops(ITradeDataItem message) { decimal longStopLossShift = message.Value * 0.99m; decimal longTakeProfitShift = message.Value * 1.03m; decimal shortStopLossShift = message.Value * 0.99m; decimal shortTakeProfitShift = message.Value * 1.03m; if (SupportLevels.TryGetValue(message.Figi, out var levels)) { if (levels.Length > 0) { var levelsByTime = levels.Where(l => l.LastLevelTime.HasValue) .OrderByDescending(l => l.LastLevelTime) .ToArray(); if (message.Price >= levelsByTime[0].LowValue && message.Price < levelsByTime[0].HighValue) { longStopLossShift = message.Price - levelsByTime[0].LowValue; shortStopLossShift = message.Price + levelsByTime[0].HighValue; } else { var levelsByDiff = levels.Where(l => l.LastLevelTime.HasValue) .OrderBy(l => System.Math.Abs(l.Value - message.Price)) .ToArray(); var nearestLevel = levelsByDiff[0]; longStopLossShift = message.Price - nearestLevel.HighValue; shortStopLossShift = nearestLevel.LowValue - message.Price; } } } return new Stops(longStopLossShift, longTakeProfitShift, shortStopLossShift, shortTakeProfitShift); } private static ImmutableDictionary ProcessStops(Stops stops, decimal meanfullLevel) { var res = TraderUtils.GetInitDict(1); if (stops.LongTakeProfitShift < meanfullLevel || stops.LongStopLossShift < meanfullLevel) { res[TradingEvent.OpenLong] = Constants.BlockingCoefficient; } if (stops.ShortTakeProfitShift < meanfullLevel || stops.ShortStopLossShift < meanfullLevel) { res[TradingEvent.OpenShort] = Constants.BlockingCoefficient; } return res.ToImmutableDictionary(); } private async Task> ProcessSupportLevels(ITradeDataItem message) { var res = TraderUtils.GetInitDict(1); if (SupportLevels.TryGetValue(message.Figi, out var levels)) { foreach (var lev in levels) { if (message.Price >= lev.LowValue && message.Price < lev.HighValue) { await _tradeDataProvider.LogPrice(message, "support_level", message.Price); } } if (levels.Length > 0) { var levelsByTime = levels.Where(l => l.LastLevelTime.HasValue) .OrderByDescending(l => l.LastLevelTime) .ToArray(); if (message.Price >= levelsByTime[0].LowValue && message.Price < levelsByTime[0].HighValue) { if (message.Price > levelsByTime[0].Value) { res[TradingEvent.OpenLong] = Constants.BlockingCoefficient; } if (message.Price < levelsByTime[0].Value) { res[TradingEvent.OpenShort] = Constants.BlockingCoefficient; } } } } return res.ToImmutableDictionary(); } } }