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

D3: Implement IngestMode classification, weekly biopsy, SnapshotOnly bypass, and Aggregated import grouping

Browse Source
This commit is contained in:
Richard
2026-07-19 17:23:19 +02:00
parent 6fd7e563f4
commit 3ed0b4df27
7 changed files with 401 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Predictalytics.Worker/Services/PollingWorker.cs
+32 -1
View File
@@ -53,7 +53,7 @@ public class PollingWorker : BackgroundService
var rateLimiter = scope.ServiceProvider.GetRequiredService<IRateLimiter>();
var trader = await traderRepo.GetByIdAsync(t.Id, stoppingToken);
if (trader == null) continue;
if (trader == null || trader.IngestMode == IngestMode.SnapshotOnly) continue;
var provider = providers.FirstOrDefault(p => p.Platform == trader.Platform && p.IsImplemented);
if (provider == null) continue;
@@ -122,6 +122,37 @@ public class PollingWorker : BackgroundService
}
}
if (trader.IngestMode == IngestMode.Aggregated && newTrades.Count > 0)
{
var aggregated = new List<Domain.Entities.Trade>();
foreach (var grp in newTrades.GroupBy(t => new { t.MarketOutcomeId, t.Side, Hour = t.ExecutedAt.ToString("yyyyMMddHH") }))
{
var first = grp.First();
var totalAmount = grp.Sum(t => t.Amount);
var totalSize = grp.Sum(t => t.Size);
var vwap = totalSize > 0 ? totalAmount / totalSize : first.Price;
var aggTrade = new Domain.Entities.Trade
{
PlatformTradeId = $"AGG_{trader.Id}_{grp.Key.MarketOutcomeId}_{grp.Key.Side}_{grp.Key.Hour}",
TraderId = trader.Id,
MarketOutcomeId = first.MarketOutcomeId,
DbMarketId = first.DbMarketId,
MarketId = first.MarketId,
AssetId = first.AssetId,
Outcome = first.Outcome,
Side = first.Side,
Price = vwap,
Amount = totalAmount,
Size = totalSize,
ExecutedAt = first.ExecutedAt,
AggregatedCount = grp.Count()
};
aggregated.Add(aggTrade);
}
newTrades = aggregated;
}
// ── Persist new trades ──
if (newTrades.Count > 0)
{