vladzvx
/
klhztrader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

обновление кеша данных об активах
test / deploy_trader_prod (push) Successful in 3m0s Details

dev
vlad zverzhkhovskiy 2025-09-22 12:17:20 +03:00
parent b6d7cd88c3
commit 214eb591bc
17 changed files with 225 additions and 1124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
KLHZ.Trader.Core.Contracts/Messaging/Interfaces/IDataBus.cs
View File

@ -6,12 +6,18 @@ namespace KLHZ.Trader.Core.Contracts.Messaging.Interfaces
public interface IDataBus
{
public bool AddChannel(string key, Channel<IOrderbook> channel);
public bool AddChannel(string key, Channel<IProcessedPrice> channel);
public bool AddChannel(string key, Channel<INewPrice> channel);
public bool AddChannel(string key, Channel<ITradeCommand> channel);
public bool AddChannel(string key, Channel<IMessage> channel);
public bool AddChannel(string key, Channel<INewCandle> channel);
public Task Broadcast(INewPrice newPriceMessage);
public bool AddChannel(string key, Channel<IProcessedPrice> channel);
public bool AddChannel(string key, Channel<ITradeCommand> channel);
public bool AddChannel(string key, Channel<INewCandle> channel);
public Task Broadcast(ITradeCommand command);
public Task Broadcast(IProcessedPrice command);
public Task Broadcast(IOrderbook orderbook);

26
KLHZ.Trader.Core.Tests/AssetTests.cs
View File

@ -1,26 +0,0 @@
using KLHZ.Trader.Core.Exchange.Models.AssetsAccounting;
namespace KLHZ.Trader.Core.Tests
{
public class AssetTests
{
[Test]
public void Test1()
{
var asset = new Asset() { AccountId = "", Figi = "", Ticker = "" };
var dur = TimeSpan.FromSeconds(5);
Assert.IsTrue(asset.Lock(dur).Result);
Assert.IsFalse(asset.Lock(dur).Result);
}
[Test]
public void Test2()
{
var asset = new Asset() { AccountId = "", Figi = "", Ticker = "" };
var dur = TimeSpan.FromSeconds(5);
Assert.IsTrue(asset.Lock(dur).Result);
Task.Delay(dur + dur).Wait();
Assert.IsTrue(asset.Lock(dur).Result);
}
}
}

89
KLHZ.Trader.Core.Tests/TraderTests.cs
View File

@ -1,96 +1,9 @@
using KLHZ.Trader.Core.Common;
using KLHZ.Trader.Core.Exchange.Models.AssetsAccounting;
using KLHZ.Trader.Core.Exchange.Utils;
using KLHZ.Trader.Core.Exchange.Utils;
namespace KLHZ.Trader.Core.Tests
{
public class TraderTests
{
[Test]
public void IsBuyAllowedTest1()
{
BotModeSwitcher.StartPurchase();
var account = new ManagedAccount("111");
account.Total = 10000;
account.Balance = 9000;
account.Assets["123"] = new Asset()
{
AccountId = account.AccountId,
Figi = "123",
Ticker = "123",
Type = AssetType.Futures,
};
Assert.IsTrue(KLHZ.Trader.Core.Exchange.Services.Trader.IsBuyAllowed(account, 3000, 1, 0.5m, 0.3m));
}
[Test]
public void IsBuyAllowedTest2()
{
BotModeSwitcher.StartPurchase();
var account = new ManagedAccount("111");
account.Total = 10000;
account.Balance = 5000;
account.Assets["123"] = new Asset()
{
AccountId = account.AccountId,
Figi = "123",
Ticker = "123",
Type = AssetType.Futures,
};
Assert.IsFalse(KLHZ.Trader.Core.Exchange.Services.Trader.IsBuyAllowed(account, 3000, 1, 0.5m, 0.3m));
}
[Test]
public void IsBuyAllowedTest3()
{
BotModeSwitcher.StartPurchase();
var account = new ManagedAccount("111");
account.Total = 10000;
account.Balance = 5000;
account.Assets["123"] = new Asset()
{
AccountId = account.AccountId,
Figi = "123",
Ticker = "123",
Type = AssetType.Futures,
};
Assert.IsFalse(KLHZ.Trader.Core.Exchange.Services.Trader.IsBuyAllowed(account, 1500, 2, 0.5m, 0.3m));
}
[Test]
public void IsBuyAllowedTest4()
{
BotModeSwitcher.StartPurchase();
var account = new ManagedAccount("111");
account.Total = 10000;
account.Balance = 3000;
account.Assets["123"] = new Asset()
{
AccountId = account.AccountId,
Figi = "123",
Ticker = "123",
Type = AssetType.Futures,
};
Assert.IsFalse(KLHZ.Trader.Core.Exchange.Services.Trader.IsBuyAllowed(account, 1500, 1, 0.5m, 0.3m));
}
[Test]
public void IsBuyAllowedTest5()
{
BotModeSwitcher.StartPurchase();
var account = new ManagedAccount("111");
account.Total = 10000;
account.Balance = 5000;
account.Assets["123"] = new Asset()
{
AccountId = account.AccountId,
Figi = "123",
Ticker = "123",
Type = AssetType.Common,
};
Assert.IsTrue(KLHZ.Trader.Core.Exchange.Services.Trader.IsBuyAllowed(account, 3000, 1, 0.5m, 0.1m));
}
[Test]
public void CalcProfitTest()
{

1
KLHZ.Trader.Core/Exchange/Interfaces/IManagedAccount.cs
View File

@ -10,6 +10,7 @@ namespace KLHZ.Trader.Core.Exchange.Interfaces
bool Initialized { get; }
string AccountId { get; }
Task Init(string accountId);
Task LoadPortfolio();
ImmutableDictionary<string, Asset> Assets { get; }
public Task OpenPosition(string figi, PositionType positionType, decimal stopLossShift, decimal takeProfitShift, long count = 1);
public Task ClosePosition(string figi);

4
KLHZ.Trader.Core/Exchange/Models/AssetsAccounting/Asset.cs
View File

@ -1,9 +1,8 @@
namespace KLHZ.Trader.Core.Exchange.Models.AssetsAccounting
{
public class Asset : LockableExchangeObject
public class Asset
{
public Guid AssetId { get; init; } = Guid.NewGuid();
public long? TradeId { get; init; }
public decimal BlockedItems { get; init; }
public AssetType Type { get; init; }
public PositionType Position { get; init; }
@ -13,6 +12,5 @@
public required string Ticker { get; init; }
public decimal BoughtPrice { get; init; }
public decimal Count { get; init; }
public decimal CountLots { get; init; }
}
}

35
KLHZ.Trader.Core/Exchange/Models/AssetsAccounting/LockedExchangeObject.cs
View File

@ -1,35 +0,0 @@
using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Interfaces;
namespace KLHZ.Trader.Core.Exchange.Models.AssetsAccounting
{
public abstract class LockableExchangeObject : ILockableObject
{
private readonly SemaphoreSlim _sem = new SemaphoreSlim(1, 1);
public Task<bool> Lock(TimeSpan duration)
{
var lockerTask = _sem.WaitAsync(0);
_ = lockerTask.ContinueWith(async (t) =>
{
if (t.Result)
{
await Task.Delay(duration);
_sem.Release();
}
});
return lockerTask;
}
public void Unlock()
{
try
{
_sem.Release();
}
catch
{
}
}
}
}

47
KLHZ.Trader.Core/Exchange/Models/AssetsAccounting/ManagedAccount.cs
View File

@ -1,47 +0,0 @@
using System.Collections.Concurrent;
namespace KLHZ.Trader.Core.Exchange.Models.AssetsAccounting
{
public class ManagedAccount : LockableExchangeObject
{
public readonly string AccountId;
private readonly object _locker = new();
private decimal _balance = 0;
private decimal _total = 0;
internal decimal Balance
{
get
{
lock (_locker)
return _balance;
}
set
{
lock (_locker)
_balance = value;
}
}
internal decimal Total
{
get
{
lock (_locker)
return _total;
}
set
{
lock (_locker)
_total = value;
}
}
internal readonly ConcurrentDictionary<string, Asset> Assets = new();
internal readonly ConcurrentDictionary<string, Order> Orders = new();
public ManagedAccount(string accountId)
{
AccountId = accountId;
}
}
}

68
KLHZ.Trader.Core/Exchange/Services/ExchangeDataReader.cs
View File

@ -5,8 +5,8 @@ using KLHZ.Trader.Core.DataLayer;
using KLHZ.Trader.Core.DataLayer.Entities.Orders;
using KLHZ.Trader.Core.DataLayer.Entities.Prices;
using KLHZ.Trader.Core.Exchange.Extentions;
using KLHZ.Trader.Core.Exchange.Models.AssetsAccounting;
using KLHZ.Trader.Core.Exchange.Models.Configs;
using KLHZ.Trader.Core.Exchange.Utils;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
@ -19,6 +19,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
public class ExchangeDataReader : IHostedService
{
private readonly PortfolioWrapper _portfolioWrapper;
private readonly TraderDataProvider _tradeDataProvider;
private readonly InvestApiClient _investApiClient;
private readonly string[] _instrumentsFigis = [];
@ -30,7 +31,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
private readonly bool _exchangeDataRecievingEnabled;
private readonly ConcurrentDictionary<string, DateTime> _usedOrderIds = new();
public ExchangeDataReader(InvestApiClient investApiClient, IDataBus eventBus, TraderDataProvider tradeDataProvider,
IOptions<ExchangeConfig> options, IDbContextFactory<TraderDbContext> dbContextFactory,
IOptions<ExchangeConfig> options, IDbContextFactory<TraderDbContext> dbContextFactory, PortfolioWrapper portfolioWrapper,
ILogger<ExchangeDataReader> logger)
{
_exchangeDataRecievingEnabled = options.Value.ExchangeDataRecievingEnabled;
@ -41,6 +42,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
_logger = logger;
_managedAccountNamePatterns = options.Value.ManagingAccountNamePatterns.ToArray();
_tradeDataProvider = tradeDataProvider;
_portfolioWrapper = portfolioWrapper;
}
public async Task StartAsync(CancellationToken cancellationToken)
@ -48,6 +50,10 @@ namespace KLHZ.Trader.Core.Exchange.Services
await _tradeDataProvider.Init();
_logger.LogInformation("Инициализация приемника данных с биржи");
var accounts = await _investApiClient.GetAccounts(_managedAccountNamePatterns);
foreach (var acc in accounts)
{
await _portfolioWrapper.AddAccount(acc);
}
_ = CycleSubscribtion(accounts);
}
@ -59,7 +65,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (_exchangeDataRecievingEnabled)
{
var t1 = SubscribeTrades();
var t1 = SubscribeTrades(accounts);
var t2 = SubscribeExchangeData();
await Task.WhenAll(t1, t2);
}
@ -71,13 +77,12 @@ namespace KLHZ.Trader.Core.Exchange.Services
}
}
}
private async Task SubscribeTrades()
private async Task SubscribeTrades(string[] accounts)
{
var req = new TradesStreamRequest();
foreach (var a in _tradeDataProvider.Accounts)
foreach (var a in accounts)
{
req.Accounts.Add(a.Key);
req.Accounts.Add(a);
}
using var stream = _investApiClient.OrdersStream.TradesStream(req);
@ -87,22 +92,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (_usedOrderIds.TryAdd(response.OrderTrades.OrderId, DateTime.UtcNow))
{
var deals = response.OrderTrades.Trades
.Select(t => new DealResult()
{
AccountId = response.OrderTrades.AccountId,
Figi = response.OrderTrades.Figi,
Count = t.Quantity,
Direction = response.OrderTrades.Direction == OrderDirection.Sell ? DealDirection.Sell : DealDirection.Buy,
Price = t.Price,
Success = true
})
.ToArray();
foreach (var d in deals)
{
await _tradeDataProvider.LogDeal(d);
}
await _tradeDataProvider.SyncPortfolio(response.OrderTrades.AccountId);
_ = _portfolioWrapper.Accounts[response.OrderTrades.AccountId].LoadPortfolio();
}
}
}
@ -147,6 +137,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
SubscribeOrderBookRequest = bookRequest
});
var lastUpdateDict = new Dictionary<string, PriceChange>();
var pricesBuffer = new List<PriceChange>();
var orderbookItemsBuffer = new List<OrderbookItem>();
var lastWrite = DateTime.UtcNow;
@ -164,9 +155,40 @@ namespace KLHZ.Trader.Core.Exchange.Services
Direction = (int)response.Trade.Direction,
Count = response.Trade.Quantity,
};
await _tradeDataProvider.AddData(message, TimeSpan.FromHours(7));
//await _eventBus.Broadcast(message);
var exchangeState = ExchangeScheduler.GetCurrentState();
if (exchangeState == Models.Trading.ExchangeState.ClearingTime
&& lastUpdateDict.TryGetValue(message.Figi, out var pri)
&& (DateTime.UtcNow - pri.Time).Minutes > 3)
{
var assets = _portfolioWrapper.Accounts.Values.SelectMany(a => a.Assets.Values).Where(a => a.Figi == message.Figi).ToArray();
foreach (var a in assets)
{
using var context = await _dbContextFactory.CreateDbContextAsync();
context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
await context.Trades.AddAsync(new DataLayer.Entities.Trades.Trade()
{
AssetId = a.AssetId,
AccountId = string.Empty,
Figi = message.Figi,
Ticker = string.Empty,
ArchiveStatus = 0,
Asset = (KLHZ.Trader.Core.DataLayer.Entities.Trades.Enums.AssetType)(int)a.Type,
BoughtAt = DateTime.UtcNow,
Count = 0,
Direction = a.Count > 0 ? DataLayer.Entities.Trades.Enums.TradeDirection.Buy : DataLayer.Entities.Trades.Enums.TradeDirection.Sell,
Position = a.Count > 0 ? DataLayer.Entities.Trades.Enums.PositionType.Long : DataLayer.Entities.Trades.Enums.PositionType.Short,
Price = message.Value,
});
await context.SaveChangesAsync();
}
}
lastUpdateDict[message.Figi] = message;
pricesBuffer.Add(message);
}
if (response.Orderbook != null)

36
KLHZ.Trader.Core/Exchange/Services/ManagedAccount2.cs → KLHZ.Trader.Core/Exchange/Services/ManagedAccount.cs
View File

@ -1,5 +1,4 @@
using Grpc.Core;
using KLHZ.Trader.Core.Common.Extentions;
using KLHZ.Trader.Core.Common.Extentions;
using KLHZ.Trader.Core.DataLayer;
using KLHZ.Trader.Core.Exchange.Extentions;
using KLHZ.Trader.Core.Exchange.Interfaces;
@ -16,7 +15,7 @@ using Asset = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.Asset;
namespace KLHZ.Trader.Core.Exchange.Services
{
public class ManagedAccount2 : IManagedAccount
public class ManagedAccount : IManagedAccount
{
public string AccountId { get; private set; } = string.Empty;
public bool Initialized { get; private set; } = false;
@ -67,7 +66,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1);
private readonly SemaphoreSlim _initSemaphore = new SemaphoreSlim(1, 1);
public ManagedAccount2(InvestApiClient investApiClient, IOptions<ExchangeConfig> options, IDbContextFactory<TraderDbContext> dbContextFactory, ILogger<TraderDataProvider> logger)
public ManagedAccount(InvestApiClient investApiClient, IOptions<ExchangeConfig> options, IDbContextFactory<TraderDbContext> dbContextFactory, ILogger<TraderDataProvider> logger)
{
_investApiClient = investApiClient;
_dbContextFactory = dbContextFactory;
@ -96,7 +95,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
}
}
private async Task LoadPortfolio()
public async Task LoadPortfolio()
{
try
{
@ -140,7 +139,8 @@ namespace KLHZ.Trader.Core.Exchange.Services
context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
var trades = await context.Trades
.Where(t => t.AccountId == AccountId && t.ArchiveStatus == 0)
.Where(t => t.ArchiveStatus == 0)
.OrderByDescending(t => t.BoughtAt)
.ToListAsync();
var oldAssets = _assets.ToDictionary();
@ -153,7 +153,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
var asset = new Asset()
{
AssetId = newAssetId,
TradeId = trade?.Id,
AccountId = AccountId,
Figi = position.Figi,
Ticker = position.Ticker,
@ -266,33 +265,10 @@ namespace KLHZ.Trader.Core.Exchange.Services
_semaphore.Release();
}
private async Task SubscribeTrades()
{
var req = new TradesStreamRequest();
req.Accounts.Add(AccountId);
using var stream = _investApiClient.OrdersStream.TradesStream(req);
await foreach (var response in stream.ResponseStream.ReadAllAsync())
{
if (response.OrderTrades?.Trades != null)
{
if (_usedOrderIds.TryAdd(response.OrderTrades.OrderId, DateTime.UtcNow))
{
await LoadPortfolio();
}
}
}
}
private async Task CyclingOperations()
{
Task? tradesLoadingTas = null;
while (true)
{
if (tradesLoadingTas == null || tradesLoadingTas.Status != TaskStatus.Running)
{
tradesLoadingTas = SubscribeTrades();
}
await Task.Delay(10000);
await LoadPortfolio();
}

32
KLHZ.Trader.Core/Exchange/Services/PortfolioDataProvider.cs
View File

@ -1,32 +0,0 @@
using KLHZ.Trader.Core.Exchange.Interfaces;
using KLHZ.Trader.Core.Exchange.Models.AssetsAccounting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
namespace KLHZ.Trader.Core.Exchange.Services
{
public class PortfolioDataProvider : IHostedService
{
private readonly IServiceProvider _services;
public PortfolioDataProvider(IServiceProvider services)
{
_services = services;
}
public async Task StartAsync(CancellationToken cancellationToken)
{
var acc = _services.GetKeyedService<IManagedAccount>(1);
await acc.Init("2274189208");
await acc.OpenPosition("FUTIMOEXF000", PositionType.Short, 2.5m, 4m);
await acc.ClosePosition("FUTIMOEXF000");
}
public Task StopAsync(CancellationToken cancellationToken)
{
return Task.CompletedTask;
}
}
}

39
KLHZ.Trader.Core/Exchange/Services/PortfolioWrapper.cs Normal file
View File

@ -0,0 +1,39 @@
using KLHZ.Trader.Core.Exchange.Interfaces;
using Microsoft.Extensions.DependencyInjection;
using System.Collections.Concurrent;
namespace KLHZ.Trader.Core.Exchange.Services
{
public class PortfolioWrapper
{
private readonly IServiceProvider _services;
public readonly ConcurrentDictionary<string, IManagedAccount> Accounts = new();
public PortfolioWrapper(IServiceProvider services)
{
_services = services;
}
public async Task AddAccount(string accountId)
{
for (int i = 0; i < 10; i++)
{
var acc = _services.GetKeyedService<IManagedAccount>(i);
if (acc != null)
{
if (acc.Initialized)
{
continue;
}
else
{
await acc.Init(accountId);
Accounts[accountId] = acc;
return;
}
}
}
throw new ArgumentOutOfRangeException("Уже инициализировано максимальное количество счетов.");
}
}
}

520
KLHZ.Trader.Core/Exchange/Services/Trader.cs
View File

@ -1,12 +1,11 @@
using KLHZ.Trader.Core.Common;
using KLHZ.Trader.Core.Contracts.Declisions.Dtos.Enums;
using KLHZ.Trader.Core.Contracts.Messaging.Dtos;
using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Interfaces;
using KLHZ.Trader.Core.Contracts.Messaging.Interfaces;
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.Models.AssetsAccounting;
using KLHZ.Trader.Core.Exchange.Interfaces;
using KLHZ.Trader.Core.Exchange.Models.Configs;
using KLHZ.Trader.Core.Exchange.Models.Trading;
using KLHZ.Trader.Core.Exchange.Utils;
@ -17,10 +16,11 @@ using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Collections.Concurrent;
using System.Security.Cryptography;
using System.Threading.Channels;
using Tinkoff.InvestApi;
using Asset = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.Asset;
using AssetType = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.AssetType;
using PositionType = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.PositionType;
namespace KLHZ.Trader.Core.Exchange.Services
{
@ -28,6 +28,8 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
private readonly IDataBus _dataBus;
private readonly TraderDataProvider _tradeDataProvider;
private readonly PortfolioWrapper _portfolioWrapper;
private readonly ILogger<Trader> _logger;
private readonly ConcurrentDictionary<string, TradingMode> TradingModes = new();
@ -42,6 +44,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
private readonly decimal _accountCashPart;
private readonly decimal _accountCashPartFutures;
private readonly string[] _tradingInstrumentsFigis = [];
private readonly Channel<INewPrice> _pricesChannel = Channel.CreateUnbounded<INewPrice>();
private readonly Channel<IOrderbook> _ordersbookChannel = Channel.CreateUnbounded<IOrderbook>();
@ -49,9 +52,11 @@ namespace KLHZ.Trader.Core.Exchange.Services
ILogger<Trader> logger,
IOptions<ExchangeConfig> options,
IDataBus dataBus,
PortfolioWrapper portfolioWrapper,
TraderDataProvider tradeDataProvider,
InvestApiClient investApiClient)
{
_portfolioWrapper = portfolioWrapper;
_tradeDataProvider = tradeDataProvider;
_logger = logger;
_dataBus = dataBus;
@ -71,13 +76,12 @@ namespace KLHZ.Trader.Core.Exchange.Services
}
}
public async Task StartAsync(CancellationToken cancellationToken)
public Task StartAsync(CancellationToken cancellationToken)
{
await _tradeDataProvider.Init();
_dataBus.AddChannel(nameof(Trader), _pricesChannel);
_dataBus.AddChannel(nameof(Trader), _ordersbookChannel);
_ = ProcessPrices();
_ = ProcessOrders();
return Task.CompletedTask;
}
public async ValueTask<(DateTime[] timestamps, decimal[] prices, bool isFullIntervalExists)> GetData(INewPrice message)
@ -232,7 +236,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
ProcessStops(message, currentTime);
var windowMaxSize = 2000;
await SellAssetsIfNeed(message);
var data = await _tradeDataProvider.GetData(message.Figi, windowMaxSize);
var state = ExchangeScheduler.GetCurrentState(message.Time);
await ProcessClearing(data, state, message);
@ -272,85 +275,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
}
}
private async Task ProcessOrders()
{
while (true)
{
await ProcessOrdersAction();
await Task.Delay(5000);
}
}
private async Task ProcessOrdersAction(bool cancellAll = false, string? figi = null)
{
var accounts = _tradeDataProvider.Accounts.Values.ToArray();
foreach (var account in accounts)
{
foreach (var order in account.Orders)
{
if (!string.IsNullOrEmpty(figi))
{
if (order.Value.Figi != figi)
{
continue;
}
}
if (cancellAll || order.Value.ExpirationTime < DateTime.UtcNow)
{
await _dataBus.Broadcast(new TradeCommand()
{
AccountId = account.AccountId,
Figi = "",
OrderId = order.Key,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.CancelOrder,
});
}
}
}
}
private async Task SellAssetsIfNeed(INewPrice message)
{
if (!BotModeSwitcher.CanSell())
{
_logger.LogWarning("Сброс активов недоступен, т.к. отключены продажи.");
return;
}
var accounts = _tradeDataProvider.Accounts.Values.ToArray();
var assetType = _tradeDataProvider.GetAssetTypeByFigi(message.Figi);
foreach (var acc in accounts)
{
var assets = acc.Assets.Values.Where(a => a.Figi == message.Figi).ToArray();
foreach (var asset in assets)
{
if (await asset.Lock(TimeSpan.FromSeconds(60)))
{
var profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), GetLeverage(message.Figi, asset.Count < 0), asset.Count < 0);
var stoppingKey = message.Figi + asset.AccountId;
if (profit < -100m)
{
var command = new TradeCommand()
{
AccountId = acc.AccountId,
Figi = message.Figi,
CommandType = asset.Count < 0 ? Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketBuy
: Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketSell,
Count = System.Math.Abs((long)asset.Count),
RecomendPrice = null,
EnableMargin = false,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Сброс актива {figi}! id команды {commandId} Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
await LogDeclision(DeclisionTradeAction.CloseLong, message, profit);
await LogDeclision(DeclisionTradeAction.CloseLongReal, message, profit);
}
}
}
}
}
private async Task<TradingEvent> CheckByWindowAverageMean((DateTime[] timestamps, decimal[] prices) data,
INewPrice message, int windowMaxSize, decimal uptrendStartingDetectionMeanfullStep = 0m, decimal uptrendEndingDetectionMeanfullStep = 3m)
{
@ -444,6 +368,61 @@ namespace KLHZ.Trader.Core.Exchange.Services
return null;
}
private async Task ClosePositions(Asset[] assets, INewPrice message, bool withProfitOnly = true)
{
var loggedDeclisions = 0;
var assetType = _tradeDataProvider.GetAssetTypeByFigi(message.Figi);
var assetsForClose = new List<Asset>();
foreach (var asset in assets)
{
if (withProfitOnly)
{
var profit = 0m;
if (assetType == AssetType.Futures)
{
profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), GetLeverage(message.Figi, asset.Count < 0), asset.Count < 0);
}
if (profit > 0)
{
assetsForClose.Add(asset);
if (loggedDeclisions == 0)
{
loggedDeclisions++;
await LogDeclision(asset.Count < 0 ? DeclisionTradeAction.CloseShortReal : DeclisionTradeAction.CloseLongReal, message, profit);
}
}
}
else
{
assetsForClose.Add(asset);
}
}
var tasks = assetsForClose.Select(asset => _portfolioWrapper.Accounts[asset.AccountId].ClosePosition(message.Figi));
await Task.WhenAll(tasks);
}
private async Task OpenPositions(IManagedAccount[] accounts, INewPrice message, PositionType positionType, decimal stopLossShift, decimal takeProfitShift, long count = 1)
{
var loggedDeclisions = 0;
foreach (var acc in accounts)
{
if (IsOperationAllowed(acc, message.Value, 1, _accountCashPartFutures, _accountCashPart))
{
await acc.OpenPosition(message.Figi, positionType, stopLossShift, takeProfitShift, count);
}
if (loggedDeclisions == 0)
{
await LogDeclision(DeclisionTradeAction.OpenLongReal, message);
LongOpeningStops[message.Figi] = message.Time.AddMinutes(1);
loggedDeclisions++;
}
}
}
private async Task ProcessNewPriceIMOEXF2((DateTime[] timestamps, decimal[] prices) data,
ExchangeState state,
INewPrice message, int windowMaxSize)
@ -489,37 +468,12 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase())
{
var accounts = _tradeDataProvider.Accounts
var accounts = _portfolioWrapper.Accounts
.Where(a => !a.Value.Assets.ContainsKey(message.Figi))
.Take(1)
.Select(a => a.Value)
.ToArray();
var loggedDeclisions = 0;
foreach (var acc in accounts)
{
if (IsBuyAllowed(acc.Value, message.Value, 1, _accountCashPartFutures, _accountCashPart))
{
if (RandomNumberGenerator.GetInt32(100) > 50 && await acc.Value.Lock(TimeSpan.FromSeconds(60)))
{
var command = new TradeCommand()
{
AccountId = acc.Value.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketBuy,
Count = 1,
RecomendPrice = null,
ExchangeObject = acc.Value,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Покупка актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
await LogDeclision(DeclisionTradeAction.OpenLongReal, message);
LongOpeningStops[message.Figi] = message.Time.AddMinutes(1);
loggedDeclisions++;
}
}
}
}
await OpenPositions(accounts, message, PositionType.Long, 7, 10, 1);
}
await LogDeclision(DeclisionTradeAction.OpenLong, message);
@ -528,107 +482,26 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (!message.IsHistoricalData && BotModeSwitcher.CanSell())
{
var loggedDeclisions = 0;
var assetsForClose = _tradeDataProvider.Accounts
var assetsForClose = _portfolioWrapper.Accounts
.SelectMany(a => a.Value.Assets.Values)
.Where(a => a.Figi == message.Figi && a.Count > 0)
.ToArray();
foreach (var asset in assetsForClose)
{
if (await asset.Lock(TimeSpan.FromSeconds(60)))
{
var profit = 0m;
if (assetType == AssetType.Common && asset.Count > 0)
{
profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), 1, false);
}
if (assetType == AssetType.Futures)
{
profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), GetLeverage(message.Figi, asset.Count < 0), asset.Count < 0);
}
if (profit > 0)
{
LongClosingStops[message.Figi] = message.Time.AddSeconds(30);
var command = new TradeCommand()
{
AccountId = asset.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketSell,
Count = (long)asset.Count,
RecomendPrice = null,
EnableMargin = false,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Продажа актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
loggedDeclisions++;
await LogDeclision(DeclisionTradeAction.CloseLongReal, message, profit);
}
}
}
}
await ClosePositions(assetsForClose, message);
}
await LogDeclision(DeclisionTradeAction.CloseLong, message);
}
if ((res & TradingEvent.DowntrendEnd) == TradingEvent.DowntrendEnd)
{
if (!ShortClosingStops.ContainsKey(message.Figi))
if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase())
{
if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase())
{
var loggedDeclisions = 0;
var assetsForClose = _tradeDataProvider.Accounts
.SelectMany(a => a.Value.Assets.Values)
.Where(a => a.Figi == message.Figi && a.Count < 0)
.ToArray();
foreach (var asset in assetsForClose)
{
if (await asset.Lock(TimeSpan.FromSeconds(60)))
{
var profit = 0m;
if (assetType == AssetType.Futures)
{
profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), GetLeverage(message.Figi, asset.Count < 0), asset.Count < 0);
}
if (profit > 0)
{
var command = new TradeCommand()
{
AccountId = asset.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketBuy,
Count = System.Math.Abs((long)asset.Count),
RecomendPrice = null,
EnableMargin = false,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Продажа актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
loggedDeclisions++;
await LogDeclision(DeclisionTradeAction.CloseShortReal, message, profit);
}
}
}
}
}
if (message.IsHistoricalData)
{
ShortClosingStops[message.Figi] = message.Time.AddSeconds(30);
}
await LogDeclision(DeclisionTradeAction.CloseShort, message);
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 LogDeclision(DeclisionTradeAction.CloseShort, message);
}
}
@ -652,66 +525,17 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase())
{
var accounts = _tradeDataProvider.Accounts
var accounts = _portfolioWrapper.Accounts
.Where(a => !a.Value.Assets.ContainsKey(message.Figi))
.Take(1)
.Select(a => a.Value)
.ToArray();
var loggedDeclisions = 0;
foreach (var acc in accounts)
{
if (IsBuyAllowed(acc.Value, message.Value, 1, _accountCashPartFutures, _accountCashPart))
{
if (await acc.Value.Lock(TimeSpan.FromSeconds(30)))
{
var command = new TradeCommand()
{
AccountId = acc.Value.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitBuy,
Count = 1,
RecomendPrice = message.Value - 0.5m,
ExchangeObject = acc.Value,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Выставлена заявка на покупку актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
await LogDeclision(DeclisionTradeAction.OpenLongReal, message);
LongOpeningStops[message.Figi] = message.Time.AddMinutes(1);
loggedDeclisions++;
}
}
}
}
await OpenPositions(accounts, message, PositionType.Long, 5, 2, 1);
}
await LogDeclision(DeclisionTradeAction.OpenLong, message);
}
if (!message.IsHistoricalData)
{
foreach (var acc in _tradeDataProvider.Accounts)
{
if (acc.Value.Assets.TryGetValue(message.Figi, out var asset))
{
var order = acc.Value.Orders.Values.FirstOrDefault(o => o.Figi == message.Figi && o.Direction == DealDirection.Sell);
if (order == null && asset.Count > 0 && await asset.Lock(TimeSpan.FromSeconds(60)))
{
var command = new TradeCommand()
{
AccountId = asset.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitSell,
Count = (long)asset.Count,
RecomendPrice = asset.BoughtPrice + 1.5m,
EnableMargin = false,
};
await _dataBus.Broadcast(command);
}
}
}
}
}
private async Task ProcessNewPriceIMOEXF_Growing(
@ -734,38 +558,13 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase())
{
var accounts = _tradeDataProvider.Accounts
var accounts = _portfolioWrapper.Accounts
.Where(a => !a.Value.Assets.ContainsKey(message.Figi))
.Take(1)
.Select(a => a.Value)
.ToArray();
var loggedDeclisions = 0;
foreach (var acc in accounts)
{
if (IsBuyAllowed(acc.Value, message.Value, 1, _accountCashPartFutures, _accountCashPart))
{
if (RandomNumberGenerator.GetInt32(100) > 50 && await acc.Value.Lock(TimeSpan.FromSeconds(12)))
{
var command = new TradeCommand()
{
AccountId = acc.Value.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitBuy,
Count = 1,
RecomendPrice = message.Value - 0.5m,
ExchangeObject = acc.Value,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Выставлена заявка на покупку актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
await LogDeclision(DeclisionTradeAction.OpenLongReal, message);
LongOpeningStops[message.Figi] = message.Time.AddMinutes(1);
loggedDeclisions++;
}
}
}
}
await OpenPositions(accounts, message, PositionType.Long, 10, 20, 1);
}
await LogDeclision(DeclisionTradeAction.OpenLong, message);
@ -775,51 +574,12 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (!message.IsHistoricalData && BotModeSwitcher.CanSell())
{
var loggedDeclisions = 0;
var assetsForClose = _tradeDataProvider.Accounts
var assetsForClose = _portfolioWrapper.Accounts
.SelectMany(a => a.Value.Assets.Values)
.Where(a => a.Figi == message.Figi && a.Count > 0)
.ToArray();
foreach (var asset in assetsForClose)
{
if (await asset.Lock(TimeSpan.FromSeconds(60)))
{
var profit = 0m;
if (assetType == AssetType.Common && asset.Count > 0)
{
profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), 1, false);
}
if (assetType == AssetType.Futures)
{
profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), GetLeverage(message.Figi, asset.Count < 0), asset.Count < 0);
}
if (profit > 0)
{
LongClosingStops[message.Figi] = message.Time.AddSeconds(30);
var command = new TradeCommand()
{
AccountId = asset.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketSell,
Count = (long)asset.Count,
RecomendPrice = null,
EnableMargin = false,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Продажа актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
loggedDeclisions++;
await LogDeclision(DeclisionTradeAction.CloseLongReal, message, profit);
}
}
}
}
await ClosePositions(assetsForClose, message);
}
await LogDeclision(DeclisionTradeAction.CloseLong, message);
}
@ -846,38 +606,14 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (!message.IsHistoricalData && BotModeSwitcher.CanSell())
{
var accounts = _tradeDataProvider.Accounts
var accounts = _portfolioWrapper.Accounts
.Where(a => !a.Value.Assets.ContainsKey(message.Figi))
.Take(1)
.Select(a => a.Value)
.ToArray();
var loggedDeclisions = 0;
foreach (var acc in accounts)
{
if (IsBuyAllowed(acc.Value, message.Value, 1, _accountCashPartFutures, _accountCashPart))
{
if (RandomNumberGenerator.GetInt32(100) > 50 && await acc.Value.Lock(TimeSpan.FromSeconds(12)))
{
var command = new TradeCommand()
{
AccountId = acc.Value.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketSell,
Count = 1,
RecomendPrice = message.Value - 0.5m,
ExchangeObject = acc.Value,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Выставлена заявка на продажу в шорт актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
await LogDeclision(DeclisionTradeAction.OpenLongReal, message);
LongOpeningStops[message.Figi] = message.Time.AddMinutes(1);
loggedDeclisions++;
}
}
}
}
await OpenPositions(accounts, message, PositionType.Short, 10, 20, 1);
}
await LogDeclision(DeclisionTradeAction.OpenShort, message);
@ -889,44 +625,11 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
if (!message.IsHistoricalData && BotModeSwitcher.CanPurchase())
{
var loggedDeclisions = 0;
var assetsForClose = _tradeDataProvider.Accounts
var assetsForClose = _portfolioWrapper.Accounts
.SelectMany(a => a.Value.Assets.Values)
.Where(a => a.Figi == message.Figi && a.Count < 0)
.ToArray();
foreach (var asset in assetsForClose)
{
if (await asset.Lock(TimeSpan.FromSeconds(60)))
{
var profit = 0m;
if (assetType == AssetType.Futures)
{
profit = TradingCalculator.CaclProfit(asset.BoughtPrice, message.Value,
GetComission(assetType), GetLeverage(message.Figi, asset.Count < 0), asset.Count < 0);
}
if (profit > 0)
{
var command = new TradeCommand()
{
AccountId = asset.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.MarketBuy,
Count = System.Math.Abs((long)asset.Count),
RecomendPrice = null,
EnableMargin = false,
};
await _dataBus.Broadcast(command);
_logger.LogWarning("Продажа актива {figi}! id команды {commandId}. Направление сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
message.Figi, command.CommandId, command.CommandType, command.Count, command.EnableMargin);
if (loggedDeclisions == 0)
{
loggedDeclisions++;
await LogDeclision(DeclisionTradeAction.CloseShortReal, message, profit);
}
}
}
}
await ClosePositions(assetsForClose, message);
}
if (message.IsHistoricalData)
@ -936,30 +639,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
await LogDeclision(DeclisionTradeAction.CloseShort, message);
}
}
if (!message.IsHistoricalData)
{
foreach (var acc in _tradeDataProvider.Accounts)
{
if (acc.Value.Assets.TryGetValue(message.Figi, out var asset))
{
var order = acc.Value.Orders.Values.FirstOrDefault(o => o.Figi == message.Figi && o.Direction == DealDirection.Buy);
if (order == null && asset.Count < 0 && await asset.Lock(TimeSpan.FromSeconds(60)))
{
var command = new TradeCommand()
{
AccountId = asset.AccountId,
Figi = message.Figi,
CommandType = Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitBuy,
Count = System.Math.Abs((long)asset.Count),
RecomendPrice = asset.BoughtPrice - step,
EnableMargin = false,
};
await _dataBus.Broadcast(command);
}
}
}
}
}
private async Task ProcessClearing((DateTime[] timestamps, decimal[] prices) data, ExchangeState state, INewPrice message)
@ -969,7 +648,12 @@ namespace KLHZ.Trader.Core.Exchange.Services
&& data.timestamps.Length > 1
&& (data.timestamps[data.timestamps.Length - 1] - data.timestamps[data.timestamps.Length - 2]) > TimeSpan.FromMinutes(3))
{
await _tradeDataProvider.UpdateFuturesPrice(message, data.prices[data.prices.Length - 2]);
var assets = _portfolioWrapper.Accounts.Values.SelectMany(a => a.Assets.Values).Where(a => a.Figi == message.Figi).ToArray();
foreach (var a in assets)
{
}
//await _tradeDataProvider.UpdateFuturesPrice(message, data.prices[data.prices.Length - 2]);
}
}
@ -1094,7 +778,7 @@ namespace KLHZ.Trader.Core.Exchange.Services
return res;
}
internal static bool IsBuyAllowed(ManagedAccount account, decimal boutPrice, decimal count,
internal static bool IsOperationAllowed(IManagedAccount account, decimal boutPrice, decimal count,
decimal accountCashPartFutures, decimal accountCashPart)
{
if (!BotModeSwitcher.CanPurchase()) return false;

230
KLHZ.Trader.Core/Exchange/Services/TraderDataProvider.cs
View File

@ -6,8 +6,6 @@ using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Interfaces;
using KLHZ.Trader.Core.DataLayer;
using KLHZ.Trader.Core.DataLayer.Entities.Declisions;
using KLHZ.Trader.Core.DataLayer.Entities.Prices;
using KLHZ.Trader.Core.Exchange.Extentions;
using KLHZ.Trader.Core.Exchange.Models.AssetsAccounting;
using KLHZ.Trader.Core.Exchange.Models.Configs;
using KLHZ.Trader.Core.Math.Declisions.Dtos.FFT;
using KLHZ.Trader.Core.Math.Declisions.Services.Cache;
@ -18,11 +16,7 @@ using Microsoft.Extensions.Options;
using System.Collections.Concurrent;
using System.Threading.Channels;
using Tinkoff.InvestApi;
using Tinkoff.InvestApi.V1;
using Asset = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.Asset;
using AssetType = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.AssetType;
using Order = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.Order;
using PositionType = KLHZ.Trader.Core.Exchange.Models.AssetsAccounting.PositionType;
namespace KLHZ.Trader.Core.Exchange.Services
{
@ -33,17 +27,14 @@ namespace KLHZ.Trader.Core.Exchange.Services
private readonly InvestApiClient _investApiClient;
private readonly IDbContextFactory<TraderDbContext> _dbContextFactory;
private readonly ILogger<TraderDataProvider> _logger;
private readonly string[] _managedAccountsNamePatterns = [];
private readonly string[] _instrumentsFigis = [];
private readonly ConcurrentDictionary<string, FFTAnalyzeResult> _fftResults = new();
private readonly ConcurrentDictionary<string, InstrumentSettings> _instrumentsSettings = new();
private readonly ConcurrentDictionary<string, string> _tickersCache = new();
private readonly ConcurrentDictionary<string, AssetType> _assetTypesCache = new();
internal readonly ConcurrentDictionary<string, ManagedAccount> Accounts = new();
private readonly bool _isDataRecievingAllowed = false;
private readonly Channel<object> _forSave = Channel.CreateUnbounded<object>();
private readonly SemaphoreSlim _syncSemaphore = new SemaphoreSlim(1, 1);
private readonly SemaphoreSlim _initSemaphore = new SemaphoreSlim(1, 1);
public TraderDataProvider(InvestApiClient investApiClient, IOptions<ExchangeConfig> options, IDbContextFactory<TraderDbContext> dbContextFactory, ILogger<TraderDataProvider> logger)
@ -51,14 +42,8 @@ namespace KLHZ.Trader.Core.Exchange.Services
_investApiClient = investApiClient;
_dbContextFactory = dbContextFactory;
_logger = logger;
_managedAccountsNamePatterns = options.Value.ManagingAccountNamePatterns.ToArray();
_instrumentsFigis = options.Value.DataRecievingInstrumentsFigis.ToArray();
_isDataRecievingAllowed = options.Value.ExchangeDataRecievingEnabled;
foreach (var lev in options.Value.InstrumentsSettings)
{
_instrumentsSettings.TryAdd(lev.Figi, lev);
}
}
public ValueTask<FFTAnalyzeResult> GetFFtResult(string figi)
@ -193,15 +178,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
}
}
var accounts = await _investApiClient.GetAccounts(_managedAccountsNamePatterns);
var accountsList = new List<ManagedAccount>();
foreach (var accountId in accounts)
{
var acc = new ManagedAccount(accountId);
await SyncPortfolio(acc);
Accounts[accountId] = acc;
}
if (_isDataRecievingAllowed)
{
var time = DateTime.UtcNow.AddHours(-1.5);
@ -237,7 +213,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
}
}
_ = SyncPortfolioWorker();
_ = WritePricesTask();
}
catch (Exception ex)
@ -250,186 +225,10 @@ namespace KLHZ.Trader.Core.Exchange.Services
{
return _tickersCache.TryGetValue(figi, out var ticker) ? ticker : string.Empty;
}
public AssetType GetAssetTypeByFigi(string figi)
{
return _assetTypesCache.TryGetValue(figi, out var t) ? t : AssetType.Unknown;
}
private async Task SyncPortfolio(string accountId)
{
if (Accounts.TryGetValue(accountId, out var account))
{
await SyncPortfolio(account);
}
}
private async Task SyncPortfolio(ManagedAccount account)
{
try
{
await _syncSemaphore.WaitAsync();
var portfolio = await _investApiClient.Operations.GetPortfolioAsync(new PortfolioRequest()
{
AccountId = account.AccountId,
});
var oldAssets = account.Assets.Keys.ToHashSet();
using var context = await _dbContextFactory.CreateDbContextAsync();
context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
var trades = await context.Trades
.Where(t => t.AccountId == account.AccountId && t.ArchiveStatus == 0)
.ToListAsync();
foreach (var position in portfolio.Positions)
{
decimal price = 0;
var trade = trades.FirstOrDefault(t => t.Figi == position.Figi);
if (trade != null)
{
trade.Count = position.Quantity;
trade.Position = position.Quantity > 0 ? DataLayer.Entities.Trades.Enums.PositionType.Long : DataLayer.Entities.Trades.Enums.PositionType.Short;
trades.Remove(trade);
price = trade.Price;
context.Trades.Update(trade);
await context.SaveChangesAsync();
}
else
{
price = position.AveragePositionPrice;
}
#pragma warning disable CS0612 // Тип или член устарел
var asset = new Models.AssetsAccounting.Asset()
{
TradeId = trade?.Id,
AccountId = account.AccountId,
Figi = position.Figi,
Ticker = position.Ticker,
BoughtAt = trade?.BoughtAt ?? DateTime.UtcNow,
BoughtPrice = price,
Type = position.InstrumentType.ParseInstrumentType(),
Position = position.Quantity > 0 ? PositionType.Long : PositionType.Short,
BlockedItems = position.BlockedLots,
Count = position.Quantity,
CountLots = position.QuantityLots,
};
#pragma warning restore CS0612 // Тип или член устарел
account.Assets.AddOrUpdate(asset.Figi, asset, (k, v) => asset);
oldAssets.Remove(asset.Figi);
}
account.Total = portfolio.TotalAmountPortfolio;
account.Balance = portfolio.TotalAmountCurrencies;
foreach (var asset in oldAssets)
{
account.Assets.TryRemove(asset, out _);
}
var ids = trades.Select(t => t.Id).ToArray();
await context.Trades
.Where(t => ids.Contains(t.Id))
.ExecuteUpdateAsync(t => t.SetProperty(tr => tr.ArchiveStatus, 1));
var orders = await _investApiClient.Orders.GetOrdersAsync(new GetOrdersRequest() { AccountId = account.AccountId });
var actualOrders = orders.Orders.Select(o => new Order()
{
AccountId = account.AccountId,
Figi = o.Figi,
OrderId = o.OrderId,
Ticker = GetTickerByFigi(o.Figi),
Count = o.LotsRequested,
ExpirationTime = DateTime.UtcNow.AddMinutes(10),
OpenDate = DateTime.UtcNow,
Price = o.AveragePositionPrice,
Direction = (DealDirection)(int)o.Direction
}).ToArray();
foreach (var order in actualOrders)
{
account.Orders[order.OrderId] = order;
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Ошибка при синхранизации портфеля счёта {accountId}", account.AccountId);
}
_syncSemaphore.Release();
}
internal async Task UpdateFuturesPrice(INewPrice newPrice, decimal newPriceValue)
{
using var context = await _dbContextFactory.CreateDbContextAsync();
context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
await context.Trades
.Where(t => t.Figi == newPrice.Figi && t.ArchiveStatus == 0 && t.Asset == DataLayer.Entities.Trades.Enums.AssetType.Future)
.ExecuteUpdateAsync(t => t.SetProperty(tr => tr.Price, newPriceValue).SetProperty(tr => tr.BoughtAt, DateTime.UtcNow));
foreach (var account in Accounts.Values)
{
await SyncPortfolio(account);
}
}
internal async Task LogDeal(DealResult dealResult)
{
using var context = await _dbContextFactory.CreateDbContextAsync();
context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
var priceCoeff = 1m;
var sign = dealResult.Direction == DealDirection.Sell ? -1m : 1;
var dealCount = dealResult.Count * sign;
if (_instrumentsSettings.TryGetValue(dealResult.Figi, out var se))
{
priceCoeff = se.PriceToRubConvertationCoefficient;
}
var trade = await context.Trades.FirstOrDefaultAsync(t => t.ArchiveStatus == 0 && t.Figi == dealResult.Figi && t.AccountId == dealResult.AccountId);
if (trade == null)
{
var newTrade = new DataLayer.Entities.Trades.Trade()
{
AccountId = dealResult.AccountId,
Figi = dealResult.Figi,
Ticker = GetTickerByFigi(dealResult.Figi),
BoughtAt = DateTime.UtcNow,
Count = dealCount,
Price = dealResult.Price * priceCoeff,
Position = dealCount >= 0 ? DataLayer.Entities.Trades.Enums.PositionType.Long : DataLayer.Entities.Trades.Enums.PositionType.Short,
Direction = (DataLayer.Entities.Trades.Enums.TradeDirection)(int)dealResult.Direction,
Asset = (DataLayer.Entities.Trades.Enums.AssetType)(int)GetAssetTypeByFigi(dealResult.Figi)
};
await context.Trades.AddAsync(newTrade);
await context.SaveChangesAsync();
}
else
{
var oldAmount = trade.Price * trade.Count;
var newAmount = dealResult.Price * priceCoeff * dealCount;
var oldCount = trade.Count;
trade.Count = trade.Count + dealCount;
if (trade.Count != 0)// Если суммарное количество элементов позиции сокращается - пересчитывать цену не нужно.
{
if (trade.Count / System.Math.Abs(trade.Count) != oldCount / System.Math.Abs(oldCount))//если сменился знак общего числа активов.
{
trade.Price = dealResult.Price;
trade.Position = trade.Count < 0 ? DataLayer.Entities.Trades.Enums.PositionType.Short : DataLayer.Entities.Trades.Enums.PositionType.Long;
}
else
{
trade.Price = (oldAmount + newAmount) / trade.Count;
}
context.Trades.Update(trade);
await context.SaveChangesAsync();
}
}
}
internal async Task LogPrice(ProcessedPrice price, bool saveImmediately)
{
if (saveImmediately)
@ -444,7 +243,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
await _forSave.Writer.WriteAsync(price);
}
}
internal async Task LogDeclision(Declision declision, bool saveImmediately)
{
if (saveImmediately)
@ -459,26 +257,6 @@ namespace KLHZ.Trader.Core.Exchange.Services
await _forSave.Writer.WriteAsync(declision);
}
}
private async Task SyncPortfolioWorker()
{
while (true)
{
try
{
await Task.Delay(20000);
foreach (var acc in Accounts)
{
await SyncPortfolio(acc.Value);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Ошибка при цикличном обновлении портфеля");
}
}
}
private async Task WritePricesTask()
{
var buffer1 = new List<ProcessedPrice>();
@ -520,11 +298,5 @@ namespace KLHZ.Trader.Core.Exchange.Services
}
}
}
public ValueTask<Asset[]> GetAssetsByFigi(string figi)
{
var assets = Accounts.Values.SelectMany(a => a.Assets.Values.Where(aa => aa.Figi == figi)).ToArray();
return ValueTask.FromResult(assets);
}
}
}

131
KLHZ.Trader.Core/Exchange/Services/TradingCommandsExecutor.cs
View File

@ -1,131 +0,0 @@
using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Interfaces;
using KLHZ.Trader.Core.Contracts.Messaging.Interfaces;
using KLHZ.Trader.Core.Exchange.Models.AssetsAccounting;
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
{
if (tradeCommand.CommandType == Contracts.Messaging.Dtos.Enums.TradeCommandType.CancelOrder && !string.IsNullOrEmpty(tradeCommand.OrderId))
{
var res = await _investApiClient.Orders.CancelOrderAsync(new CancelOrderRequest() { AccountId = tradeCommand.AccountId, OrderId = tradeCommand.OrderId });
}
else
{
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;
}
else if (tradeCommand.CommandType == Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitSell && tradeCommand.RecomendPrice.HasValue)
{
dir = OrderDirection.Sell;
orderType = OrderType.Limit;
}
if (orderType == OrderType.Unspecified)
{
return;
}
var req = new PostOrderRequest()
{
AccountId = tradeCommand.AccountId,
InstrumentId = tradeCommand.Figi,
Direction = dir,
Price = tradeCommand.RecomendPrice ?? 0,
OrderType = orderType,
Quantity = tradeCommand.Count,
ConfirmMarginTrade = tradeCommand.EnableMargin,
};
_logger.LogWarning("Получена команда c id {commandId} на операцию с активом {figi}! Тип заявки сделки: {dir}; Количество активов: {count}; Разрешена ли маржиналка: {margin}",
tradeCommand.CommandId, req.InstrumentId, req.OrderType, req.Quantity, req.ConfirmMarginTrade);
var res = await _investApiClient.Orders.PostOrderAsync(req);
if ((tradeCommand.CommandType == Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitBuy
|| tradeCommand.CommandType == Contracts.Messaging.Dtos.Enums.TradeCommandType.LimitSell)
&& tradeCommand.RecomendPrice.HasValue)
{
_tradeDataProvider.Accounts[tradeCommand.AccountId].Orders[res.OrderId] = new Models.AssetsAccounting.Order()
{
AccountId = tradeCommand.AccountId,
Figi = tradeCommand.Figi,
OrderId = res.OrderId,
Ticker = _tradeDataProvider.GetTickerByFigi(tradeCommand.Figi),
Count = res.LotsRequested,
Direction = (DealDirection)(int)dir,
ExpirationTime = DateTime.UtcNow.AddMinutes(10),
OpenDate = DateTime.UtcNow,
Price = tradeCommand.RecomendPrice.Value,
};
}
_logger.LogWarning("Исполнена команда c id {commandId} на операцию с активом {figi}! Направление: {dir}; Число лотов: {lots};", tradeCommand.CommandId, res.Figi,
res.Direction, res.LotsExecuted);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Ошибка при покупке актива на счёт {acc}. figi: {figi}", tradeCommand.AccountId, tradeCommand.Figi);
}
tradeCommand.ExchangeObject?.Unlock();
}
private async Task ProcessCommands()
{
while (await _channel.Reader.WaitToReadAsync())
{
var command = await _channel.Reader.ReadAsync();
await ExecuteCommand(command);
}
}
}
}

67
KLHZ.Trader.Core/TG/Services/BotMessagesHandler.cs
View File

@ -1,6 +1,4 @@
using KLHZ.Trader.Core.Common;
using KLHZ.Trader.Core.Contracts.Messaging.Dtos;
using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Enums;
using KLHZ.Trader.Core.Contracts.Messaging.Interfaces;
using KLHZ.Trader.Core.Exchange.Services;
using Microsoft.Extensions.Logging;
@ -18,9 +16,11 @@ namespace KLHZ.Trader.Core.TG.Services
private readonly ImmutableArray<long> _admins = [];
private readonly IDataBus _eventBus;
private readonly ILogger<BotMessagesHandler> _logger;
private readonly PortfolioWrapper _portfolioWrapper;
private readonly TraderDataProvider _traderDataProvider;
public BotMessagesHandler(IDataBus eventBus, IOptions<TgBotConfig> options, ILogger<BotMessagesHandler> logger, TraderDataProvider traderDataProvider)
public BotMessagesHandler(IDataBus eventBus, PortfolioWrapper portfolioWrapper, IOptions<TgBotConfig> options, ILogger<BotMessagesHandler> logger, TraderDataProvider traderDataProvider)
{
_portfolioWrapper = portfolioWrapper;
_traderDataProvider = traderDataProvider;
_logger = logger;
_eventBus = eventBus;
@ -76,65 +76,32 @@ namespace KLHZ.Trader.Core.TG.Services
break;
}
case "скинуть IMOEXF":
case "сбросить IMOEXF":
{
var assets = await _traderDataProvider.GetAssetsByFigi("FUTIMOEXF000");
var assets = _portfolioWrapper.Accounts
.SelectMany(a => a.Value.Assets)
.Select(a => a.Value)
.Where(a => a.Figi == "FUTIMOEXF000");
foreach (var asset in assets)
{
if (asset.Count > 0)
{
var command = new TradeCommand()
{
AccountId = asset.AccountId,
CommandType = TradeCommandType.MarketSell,
RecomendPrice = null,
Figi = asset.Figi,
Count = System.Math.Abs((long)asset.Count),
EnableMargin = false,
};
await _eventBus.Broadcast(command);
}
if (asset.Count < 0)
{
var command = new TradeCommand()
{
AccountId = asset.AccountId,
CommandType = TradeCommandType.MarketBuy,
RecomendPrice = null,
Figi = asset.Figi,
Count = System.Math.Abs((long)asset.Count),
EnableMargin = false,
};
await _eventBus.Broadcast(command);
}
await _portfolioWrapper.Accounts[asset.AccountId].ClosePosition("FUTIMOEXF000");
}
break;
}
case "продать IMOEXF":
case "лонг IMOEXF":
{
var command = new TradeCommand()
{
AccountId = "2274189208",
CommandType = TradeCommandType.MarketSell,
RecomendPrice = null,
Figi = "FUTIMOEXF000",
Count = 1
};
await _eventBus.Broadcast(command);
var acc = _portfolioWrapper.Accounts.Values.FirstOrDefault(a => !a.Assets.ContainsKey("FUTIMOEXF000"));
if (acc != null)
await acc.OpenPosition("FUTIMOEXF000", Exchange.Models.AssetsAccounting.PositionType.Long, 10, 10, 1);
break;
}
case "купить IMOEXF":
case "шорт IMOEXF":
{
var command = new TradeCommand()
{
AccountId = "2274189208",
CommandType = TradeCommandType.MarketBuy,
RecomendPrice = null,
Figi = "FUTIMOEXF000",
Count = 1
};
await _eventBus.Broadcast(command);
var acc = _portfolioWrapper.Accounts.Values.FirstOrDefault(a => !a.Assets.ContainsKey("FUTIMOEXF000"));
if (acc != null)
await acc.OpenPosition("FUTIMOEXF000", Exchange.Models.AssetsAccounting.PositionType.Short, 10, 10, 1);
break;
}
case "ребут":

2
KLHZ.Trader.Core/TG/Services/BotStarter.cs
View File

@ -20,7 +20,7 @@ namespace KLHZ.Trader.Core.TG.Services
{
_botClient = new TelegramBotClient(cfg.Value.Token);
_updateHandler = updateHandler;
dataBus.AddChannel("", _messages);
dataBus.AddChannel(string.Empty, _messages);
_admins = ImmutableArray.CreateRange(options.Value.Admins);
_ = ProcessMessages();
}

10
KLHZ.Trader.Service/Program.cs
View File

@ -45,21 +45,15 @@ builder.Services.AddDbContextFactory<TraderDbContext>(options =>
builder.Services.AddHostedService<BotStarter>();
builder.Services.AddHostedService<ExchangeDataReader>();
builder.Services.AddHostedService<Trader>();
builder.Services.AddHostedService<TradingCommandsExecutor>();
builder.Services.AddHostedService<PortfolioDataProvider>();
//builder.Services.AddHostedService<KalmanPredictor>();
builder.Services.AddSingleton<IUpdateHandler, BotMessagesHandler>();
builder.Services.AddSingleton<PortfolioWrapper>();
builder.Services.AddSingleton<TraderDataProvider>();
builder.Services.AddSingleton<IDataBus, DataBus>();
for (int i = 0; i < 10; i++)
{
builder.Services.AddKeyedSingleton<IManagedAccount, ManagedAccount2>(i);
builder.Services.AddKeyedSingleton<IManagedAccount, ManagedAccount>(i);
}
builder.Services.Configure<TgBotConfig>(builder.Configuration.GetSection(nameof(TgBotConfig)));