Richard/Predictalytics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix WinRate 0% bug: Optimize TradeReconciliationWorker with bulk SQL and reset Trader analytics flag

Browse Source
This commit is contained in:
Richard
2026-07-07 10:56:52 +02:00
parent 910d25159a
commit d24f0ecc2e
2 changed files with 35 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
query.csx
BIN
View File
Binary file not shown.
src/Predictalytics.Worker/Services/TradeReconciliationWorker.cs
+32 -25
View File
@@ -16,7 +16,7 @@ public class TradeReconciliationWorker : BackgroundService
private readonly ILogger<TradeReconciliationWorker> _logger;
private const int BatchSize = 250;
private const int IntervalMinutes = 15;
private const int IntervalSeconds = 30;
public TradeReconciliationWorker(IServiceProvider services, ILogger<TradeReconciliationWorker> logger)
{
@@ -26,7 +26,7 @@ public class TradeReconciliationWorker : BackgroundService
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_logger.LogInformation("🛠️ TradeReconciliationWorker started (batch: {Batch}, interval: {Min}m)", BatchSize, IntervalMinutes);
_logger.LogInformation("🛠️ TradeReconciliationWorker started (batch: {Batch}, interval: {Sec}s)", BatchSize, IntervalSeconds);
// Initial delay to let other workers settle
await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
@@ -52,7 +52,7 @@ public class TradeReconciliationWorker : BackgroundService
_logger.LogError(ex, "Error in TradeReconciliationWorker cycle");
}
await Task.Delay(TimeSpan.FromMinutes(IntervalMinutes), stoppingToken);
await Task.Delay(TimeSpan.FromSeconds(IntervalSeconds), stoppingToken);
}
_logger.LogInformation("🛠️ TradeReconciliationWorker stopped");
@@ -61,37 +61,44 @@ public class TradeReconciliationWorker : BackgroundService
private async Task<int> ReconcileBatchAsync(CancellationToken ct)
{
using var scope = _services.CreateScope();
var tradeRepo = scope.ServiceProvider.GetRequiredService<ITradeRepository>();
var marketRepo = scope.ServiceProvider.GetRequiredService<IMarketRepository>();
var orphanedTrades = await tradeRepo.GetOrphanedTradesAsync(BatchSize, ct);
if (orphanedTrades.Count == 0) return 0;
int count = 0;
foreach (var trade in orphanedTrades)
{
if (ct.IsCancellationRequested) break;
var db = scope.ServiceProvider.GetRequiredService<Predictalytics.Infrastructure.Data.AppDbContext>();
// Bulk update orphaned trades
int reconciledCount = 0;
try
{
var outcome = await marketRepo.GetOutcomeByTokenIdAsync(trade.AssetId, ct);
if (outcome != null)
{
trade.MarketOutcomeId = outcome.Id;
trade.Outcome = outcome.Label;
if (outcome.Market != null)
trade.DbMarketId = outcome.Market.Id;
reconciledCount = await Microsoft.EntityFrameworkCore.RelationalDatabaseFacadeExtensions.ExecuteSqlRawAsync(db.Database, @"
UPDATE Trades t
INNER JOIN MarketOutcomes o ON t.AssetId = o.TokenId
INNER JOIN Markets m ON o.MarketId = m.Id
SET t.MarketOutcomeId = o.Id,
t.Outcome = o.Label,
t.DbMarketId = m.Id
WHERE t.MarketOutcomeId IS NULL AND t.AssetId != '';
", ct);
await tradeRepo.UpdateAsync(trade, ct);
count++;
if (reconciledCount > 0)
{
// Force re-analysis of traders who now have new linked trades that haven't been applied
await Microsoft.EntityFrameworkCore.RelationalDatabaseFacadeExtensions.ExecuteSqlRawAsync(db.Database, @"
UPDATE Traders t
SET LastAnalyzedAt = NULL
WHERE LastAnalyzedAt IS NOT NULL
AND EXISTS (
SELECT 1 FROM Trades tr
LEFT JOIN TraderPositions tp ON tr.TraderId = tp.TraderId AND tr.MarketOutcomeId = tp.MarketOutcomeId
WHERE tr.TraderId = t.Id
AND tr.MarketOutcomeId IS NOT NULL
AND (tp.Id IS NULL OR tr.Id > tp.LastAppliedTradeId)
);
", ct);
}
}
catch (Exception ex)
{
_logger.LogWarning("Failed to reconcile trade {TradeId}: {Msg}", trade.Id, ex.Message);
}
_logger.LogError(ex, "Error executing bulk reconciliation SQL");
}
return count;
return reconciledCount;
}
}