using Predictalytics.Domain.Entities; using Predictalytics.Domain.Enums; using Predictalytics.Domain.Interfaces; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; namespace Predictalytics.Infrastructure.Data.Repositories; public class TradeRepository : ITradeRepository { private readonly AppDbContext _db; private readonly Microsoft.Extensions.Logging.ILogger _logger; public TradeRepository(AppDbContext db, Microsoft.Extensions.Logging.ILogger logger) { _db = db; _logger = logger; } public async Task GetByPlatformTradeIdAsync(PlatformType platform, string platformTradeId, CancellationToken ct = default) => await _db.Trades.FirstOrDefaultAsync(t => t.Platform == platform && t.PlatformTradeId == platformTradeId, ct); public async Task> GetByTraderIdAsync(int traderId, int skip = 0, int take = 50, CancellationToken ct = default) => await _db.Trades.Include(t => t.Trader).Include(t => t.DbMarket).Where(t => t.TraderId == traderId && t.DbMarketId != null) .OrderByDescending(t => t.ExecutedAt).Skip(skip).Take(take).ToListAsync(ct); public async Task> GetByDbMarketIdAsync(int dbMarketId, int skip = 0, int take = 50, CancellationToken ct = default) => await _db.Trades.Include(t => t.Trader).Include(t => t.DbMarket).Where(t => t.DbMarketId == dbMarketId) .OrderByDescending(t => t.ExecutedAt).Skip(skip).Take(take).ToListAsync(ct); public async Task> GetByMarketIdAsync(string platformMarketId, int skip = 0, int take = 50, CancellationToken ct = default) => await _db.Trades.Include(t => t.Trader).Include(t => t.DbMarket).Where(t => t.MarketId == platformMarketId && t.DbMarketId != null) .OrderByDescending(t => t.ExecutedAt).Skip(skip).Take(take).ToListAsync(ct); public async Task> GetRecentAsync(int count = 50, PlatformType? platform = null, CancellationToken ct = default) { var q = _db.Trades.Include(t => t.Trader).Include(t => t.DbMarket).Where(t => t.DbMarketId != null).AsQueryable(); if (platform.HasValue) q = q.Where(t => t.Platform == platform.Value); return await q.OrderByDescending(t => t.ExecutedAt).Take(count).ToListAsync(ct); } public async Task> GetLargestAsync(int count = 5, DateTime? since = null, CancellationToken ct = default) { var q = _db.Trades.Include(t => t.Trader).Include(t => t.DbMarket).Where(t => t.DbMarketId != null).AsQueryable(); if (since.HasValue) q = q.Where(t => t.ExecutedAt >= since.Value); return await q.OrderByDescending(t => t.Amount).Take(count).ToListAsync(ct); } public async Task GetCountAsync(int? traderId = null, CancellationToken ct = default) { var q = _db.Trades.AsQueryable(); if (traderId.HasValue) q = q.Where(t => t.TraderId == traderId.Value); return await q.CountAsync(ct); } public async Task AddRangeAsync(IEnumerable trades, CancellationToken ct = default) { var tradeList = trades.ToList(); if (tradeList.Count == 0) return; foreach (var t in tradeList) { t.Outcome = StringHelper.Truncate(t.Outcome, 128) ?? ""; t.PlatformTradeId = StringHelper.Truncate(t.PlatformTradeId, 256) ?? ""; t.MarketId = StringHelper.Truncate(t.MarketId, 66) ?? ""; t.AssetId = StringHelper.Truncate(t.AssetId, 80) ?? ""; if (t.TransactionHash != null) t.TransactionHash = StringHelper.Truncate(t.TransactionHash, 66); } foreach (var chunk in tradeList.Chunk(500)) { var sb = new System.Text.StringBuilder("INSERT INTO Trades (PlatformTradeId, MarketId, AssetId, Outcome, Side, Price, Size, Amount, ExecutedAt, TransactionHash, TraderId, MarketOutcomeId, DbMarketId, Platform, IsContextEnriched, AggregatedCount, UsdcSize, OutcomeIndex) VALUES "); var parameters = new List(); for (int i = 0; i < chunk.Length; i++) { var t = chunk[i]; int pIdx = i * 18; sb.Append($"({{{pIdx}}}, {{{pIdx + 1}}}, {{{pIdx + 2}}}, {{{pIdx + 3}}}, {{{pIdx + 4}}}, {{{pIdx + 5}}}, {{{pIdx + 6}}}, {{{pIdx + 7}}}, {{{pIdx + 8}}}, {{{pIdx + 9}}}, {{{pIdx + 10}}}, {{{pIdx + 11}}}, {{{pIdx + 12}}}, {{{pIdx + 13}}}, {{{pIdx + 14}}}, {{{pIdx + 15}}}, {{{pIdx + 16}}}, {{{pIdx + 17}}})"); if (i < chunk.Length - 1) sb.Append(", "); parameters.Add(t.PlatformTradeId); parameters.Add(t.MarketId); parameters.Add(t.AssetId); parameters.Add(t.Outcome); parameters.Add((int)t.Side); parameters.Add(t.Price); parameters.Add(t.Size); parameters.Add(t.Amount); parameters.Add(t.ExecutedAt); parameters.Add(t.TransactionHash ?? (object?)null); parameters.Add(t.TraderId); parameters.Add(t.MarketOutcomeId ?? (object?)null); parameters.Add(t.DbMarketId ?? (object?)null); parameters.Add((int)t.Platform); parameters.Add(t.IsContextEnriched); parameters.Add(t.AggregatedCount ?? (object?)null); parameters.Add(t.UsdcSize ?? (object?)null); parameters.Add(t.OutcomeIndex ?? (object?)null); } // For Aggregated Trades, we want UPSERT logic to update size, amount and VWAP sb.Append(" ON DUPLICATE KEY UPDATE Price=VALUES(Price), Size=VALUES(Size), Amount=VALUES(Amount), AggregatedCount=VALUES(AggregatedCount), UsdcSize=VALUES(UsdcSize), OutcomeIndex=VALUES(OutcomeIndex);"); int maxRetries = 3; var backoffs = new[] { 250, 500, 1000 }; for (int retry = 0; retry <= maxRetries; retry++) { try { var rowsInserted = await _db.Database.ExecuteSqlRawAsync(sb.ToString(), parameters.ToArray(), ct); _logger.LogInformation("Inserted {RowsInserted} trades into the database.", rowsInserted); break; } catch (Exception ex) { var mysqlEx = ex as MySqlConnector.MySqlException ?? ex.InnerException as MySqlConnector.MySqlException; if (mysqlEx != null && (mysqlEx.Number == 1213 || mysqlEx.Number == 1205)) { if (retry == maxRetries) { _logger.LogError(ex, "Failed to insert {Count} trades after {Retries} retries due to deadlocks.", chunk.Length, maxRetries); } else { _logger.LogWarning("Deadlock detected during trade insertion. Retrying in {Delay}ms... (Attempt {Attempt}/{Max})", backoffs[retry], retry + 1, maxRetries); await Task.Delay(backoffs[retry], ct); } } else { throw; } } } } } public async Task GetTotalVolumeAsync(DateTime? since = null, CancellationToken ct = default) { var q = _db.Trades.AsQueryable(); if (since.HasValue) q = q.Where(t => t.ExecutedAt >= since.Value); return await q.SumAsync(t => t.Amount, ct); } public async Task> GetOrphanedTradesAsync(int limit, CancellationToken ct = default) { return await _db.Trades .Where(t => t.MarketOutcomeId == null && !string.IsNullOrEmpty(t.AssetId)) .OrderByDescending(t => t.ExecutedAt) .Take(limit) .ToListAsync(ct); } public async Task> GetKnownPlatformTradeIdsAsync(PlatformType platform, int traderId, IEnumerable platformTradeIds, CancellationToken ct = default) { var idList = platformTradeIds.ToList(); if (idList.Count == 0) return new HashSet(StringComparer.OrdinalIgnoreCase); var ids = await _db.Trades .Where(t => t.Platform == platform && idList.Contains(t.PlatformTradeId)) .Select(t => t.PlatformTradeId) .ToListAsync(ct); return new HashSet(ids, StringComparer.OrdinalIgnoreCase); } public async Task UpdateAsync(Trade trade, CancellationToken ct = default) { trade.Outcome = StringHelper.Truncate(trade.Outcome, 128) ?? ""; trade.PlatformTradeId = StringHelper.Truncate(trade.PlatformTradeId, 256) ?? ""; trade.MarketId = StringHelper.Truncate(trade.MarketId, 66) ?? ""; trade.AssetId = StringHelper.Truncate(trade.AssetId, 80) ?? ""; _db.Trades.Update(trade); await _db.SaveChangesAsync(ct); } public async Task> GetTradesForContextEnrichmentAsync(int limit, CancellationToken ct = default) { return await _db.Trades .Include(t => t.Trader) .Include(t => t.Trader.CurrentScore) .Include(t => t.Trader.WatchlistEntries) .Include(t => t.Trader.Analytics) .Where(t => !t.IsContextEnriched && t.Platform == PlatformType.Polymarket && t.AssetId != "") .Where(t => t.Trader.WatchlistEntries.Any() || (t.Trader.Analytics != null && t.Trader.Analytics.CopytradingScore > 50)) .OrderByDescending(t => t.ExecutedAt) .Take(limit) .ToListAsync(ct); } }