Phase 5.3a: MarketSyncService in den Core
- Nach dem TradingState-Split nutzt MarketSyncService nur noch Core-State (MarketCache) + IMarketRepository + Core-API-Service -> sauber in den Core. - Ungenutzten Shim-Import entfernt. Build 0 Fehler. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Richard
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
43fdc9d181
commit
f5e7eaf2d5
@@ -1,174 +0,0 @@
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MongoDB.Driver;
|
||||
using PolyTrader.Core.Persistence;
|
||||
using PolyTraderSharp.Extensions;
|
||||
using PolyTraderSharp.Models;
|
||||
using PolyTraderSharp.Services;
|
||||
|
||||
namespace PolyTraderSharp.Services
|
||||
{
|
||||
public class MarketSyncService : BackgroundService
|
||||
{
|
||||
private readonly PolymarketApiService _apiService;
|
||||
private readonly IMarketRepository _marketRepo;
|
||||
private readonly TerminalLogger _logger;
|
||||
private readonly JobStatusRow _jobStatus;
|
||||
private readonly TradingState _state;
|
||||
|
||||
public MarketSyncService(PolymarketApiService apiService, IMarketRepository marketRepo, TerminalLogger logger, JobManager jobManager, TradingState state)
|
||||
{
|
||||
_apiService = apiService;
|
||||
_marketRepo = marketRepo;
|
||||
_logger = logger;
|
||||
_state = state;
|
||||
|
||||
_jobStatus = new JobStatusRow
|
||||
{
|
||||
JobName = "Market Data Sync",
|
||||
Description = "Polls Polymarket Gamma API for the 1000 newest markets.",
|
||||
StatusText = "Pending Initial Delay..."
|
||||
};
|
||||
|
||||
_jobStatus.ManualTriggerAction = async () =>
|
||||
{
|
||||
string oldStatus = _jobStatus.StatusText;
|
||||
_jobStatus.StatusText = "Running (Manual)...";
|
||||
await SyncMarketsAsync();
|
||||
_jobStatus.StatusText = "Idle";
|
||||
};
|
||||
|
||||
jobManager.RegisterJob(_jobStatus);
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
_logger.Info("MarketSyncService started. Will sync markets every 1 hour.");
|
||||
|
||||
// Give the app some time to start up before initial sync
|
||||
await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken);
|
||||
_jobStatus.StatusText = "Idle";
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
if (_jobStatus.IsEnabled)
|
||||
{
|
||||
try
|
||||
{
|
||||
_jobStatus.StatusText = "Running (Scheduled)...";
|
||||
await SyncMarketsAsync();
|
||||
_jobStatus.LastRun = DateTime.Now;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.Error($"MarketSyncService loop error: {ex.Message}");
|
||||
_jobStatus.StatusText = "Error!";
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (_jobStatus.StatusText != "Error!")
|
||||
_jobStatus.StatusText = "Idle";
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
_jobStatus.StatusText = "Paused";
|
||||
}
|
||||
|
||||
// Sleep for 3 minutes to keep the Cache extremely fresh against high-frequency listings
|
||||
_jobStatus.NextRun = DateTime.Now.AddMinutes(3);
|
||||
await Task.Delay(TimeSpan.FromMinutes(3), stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SyncMarketsAsync()
|
||||
{
|
||||
_logger.Info("Syncing newest markets from Polymarket API...");
|
||||
var newMarkets = await _apiService.GetRecentMarketsAsync(1000);
|
||||
|
||||
if (newMarkets.Count == 0)
|
||||
{
|
||||
_logger.Warning("No markets returned from Polymarket API during sync.");
|
||||
return;
|
||||
}
|
||||
|
||||
_marketRepo.EnsureIndexes();
|
||||
|
||||
int inserted = 0;
|
||||
int updated = 0;
|
||||
|
||||
foreach (var market in newMarkets)
|
||||
{
|
||||
var existing = _marketRepo.GetById(market.Id);
|
||||
if (existing == null)
|
||||
{
|
||||
_marketRepo.Insert(market);
|
||||
inserted++;
|
||||
|
||||
// NEW: Hot-Load active markets directly into RAM Cache
|
||||
if (!market.Closed && !string.IsNullOrEmpty(market.ClobTokenIds))
|
||||
{
|
||||
try
|
||||
{
|
||||
var tokenIds = System.Text.Json.JsonSerializer.Deserialize<List<string>>(market.ClobTokenIds);
|
||||
if (tokenIds != null)
|
||||
{
|
||||
foreach(var t in tokenIds)
|
||||
{
|
||||
_state.MarketCache[t] = market;
|
||||
}
|
||||
}
|
||||
} catch { } // Ignore JSON parse error if malformed
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Update dynamic fields like EndDate, Active, Closed
|
||||
existing.EndDate = market.EndDate;
|
||||
existing.Active = market.Active;
|
||||
existing.Closed = market.Closed;
|
||||
existing.NegRisk = market.NegRisk;
|
||||
|
||||
// The API can sometimes be slow to assign ClobTokenIds. Update them if we got new ones.
|
||||
if (string.IsNullOrEmpty(existing.ClobTokenIds) && !string.IsNullOrEmpty(market.ClobTokenIds))
|
||||
{
|
||||
existing.ClobTokenIds = market.ClobTokenIds;
|
||||
}
|
||||
|
||||
_marketRepo.Update(existing);
|
||||
updated++;
|
||||
|
||||
// Keep RAM cache synchronized to prevent using stale active/closed flags
|
||||
if (!string.IsNullOrEmpty(existing.ClobTokenIds))
|
||||
{
|
||||
try
|
||||
{
|
||||
var tokenIds = System.Text.Json.JsonSerializer.Deserialize<List<string>>(existing.ClobTokenIds);
|
||||
if (tokenIds != null)
|
||||
{
|
||||
foreach(var t in tokenIds)
|
||||
{
|
||||
if (!existing.Closed)
|
||||
{
|
||||
// Unconditionally keep active markets hot in the cache
|
||||
_state.MarketCache[t] = existing;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Only update if it is already there (e.g. to flag it as closed for running logic)
|
||||
if (_state.MarketCache.ContainsKey(t))
|
||||
_state.MarketCache[t] = existing;
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch { } // Ignore JSON parse error if malformed
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_logger.Info($"Market Sync Complete: {inserted} new markets, {updated} updated.");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user