bergmandClaude Opus 4.8
a0e367e645
- 8 Positions-Zugriffe (open_positions_{id}) -> IPositionRepository
(FindLive/UpsertLive/DeleteLive).
- 1 Market-Upsert -> IMarketRepository.
- closed_trades bleibt auf _db (ClosedTrade-Repo folgt in Phase 5).
- Verhalten unverändert (Repos bilden Shim-Semantik 1:1 nach); Build 0 Fehler.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1323 lines
71 KiB
C#
1323 lines
71 KiB
C#
using System;
|
||
using MongoDB.Driver;
|
||
using PolyTrader.Core.Persistence;
|
||
using PolyTraderSharp.Extensions;
|
||
using System.Collections.Concurrent;
|
||
using System.Linq;
|
||
using System.Text.Json;
|
||
using System.Threading;
|
||
using System.Threading.Channels;
|
||
using System.Threading.Tasks;
|
||
using Microsoft.Extensions.Hosting;
|
||
using PolyTraderSharp.Models;
|
||
|
||
namespace PolyTraderSharp.Services
|
||
{
|
||
public class TraderMonitorService : BackgroundService
|
||
{
|
||
private readonly TradingState _state;
|
||
private readonly PolymarketApiService _api;
|
||
private readonly PolymarketClobClient _clob;
|
||
private readonly ChannelWriter<CopySignal> _signalWriter;
|
||
private readonly ChannelWriter<ClosedTrade> _closedTradeWriter;
|
||
private readonly TerminalLogger _logger;
|
||
private readonly IMongoDatabase? _db;
|
||
private readonly IPositionRepository _positionRepo;
|
||
private readonly IMarketRepository _marketRepo;
|
||
|
||
// Prevents duplicates. Fast O(1) lookup cache to prevent DB spam.
|
||
private readonly ConcurrentDictionary<string, DateTime> _processedTxHashes = new();
|
||
private DateTime _lastHashCleanup = DateTime.UtcNow;
|
||
private readonly ConcurrentDictionary<string, bool> _processedClosures = new();
|
||
private readonly ConcurrentDictionary<string, DateTime> _lastPolled = new();
|
||
private readonly ConcurrentDictionary<string, bool> _activeWssPolls = new();
|
||
private DateTime _lastLivePoll = DateTime.MinValue;
|
||
private DateTime _lastClosedPoll = DateTime.MinValue;
|
||
private DateTime _lastMasterPositionPoll = DateTime.MinValue;
|
||
|
||
// AutoRedeem Tracker for REST updates
|
||
private static readonly ConcurrentDictionary<string, (int Count, DateTime LastAttempt)> _restRedeemAttempts = new();
|
||
|
||
public TraderMonitorService(
|
||
TradingState state,
|
||
PolymarketApiService api,
|
||
PolymarketClobClient clob,
|
||
ChannelWriter<CopySignal> signalWriter,
|
||
ChannelWriter<ClosedTrade> closedTradeWriter,
|
||
TerminalLogger logger,
|
||
IPositionRepository positionRepo,
|
||
IMarketRepository marketRepo,
|
||
IMongoDatabase? db = null)
|
||
{
|
||
_state = state;
|
||
_api = api;
|
||
_clob = clob;
|
||
_signalWriter = signalWriter;
|
||
_closedTradeWriter = closedTradeWriter;
|
||
_logger = logger;
|
||
_positionRepo = positionRepo;
|
||
_marketRepo = marketRepo;
|
||
_db = db;
|
||
}
|
||
|
||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||
{
|
||
_logger.Info("TraderMonitorService started background API priority polling...");
|
||
|
||
// ===== STARTUP: _processedClosures aus DB vorladen =====
|
||
// Verhindert, dass nach einem Neustart alle historischen geschlossenen Trades
|
||
// erneut als ClosedTrade-Records in die Datenbank geschrieben werden.
|
||
if (_db != null)
|
||
{
|
||
try
|
||
{
|
||
var closedCol = _db.GetCollection<ClosedTrade>("closed_trades");
|
||
var allClosed = closedCol.LiteFind(x => !x.IsDemo);
|
||
int preloaded = 0;
|
||
foreach (var ct in allClosed)
|
||
{
|
||
if (!string.IsNullOrEmpty(ct.TokenId))
|
||
{
|
||
string key = $"{ct.AccountId}_{ct.TokenId}";
|
||
_processedClosures.TryAdd(key, true);
|
||
preloaded++;
|
||
}
|
||
}
|
||
_logger.Info($"_processedClosures vorgeladen: {preloaded} Einträge aus closed_trades geladen (verhindert Duplikate nach Neustart).");
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Warning($"_processedClosures Preload fehlgeschlagen: {ex.Message}");
|
||
}
|
||
}
|
||
|
||
|
||
// ===== STARTUP: Sofortiger Warmup der Live-Positionen und Master-Tracker =====
|
||
// Ohne diesen Warmup ist der Proportionalitätsfilter nach einem Neustart 30s lang blind,
|
||
// und die 2-Min-Karenzzeit erlaubt unberechtigte SELLs.
|
||
try
|
||
{
|
||
_logger.Info("Startup: Lade Live-Positionen und Master-Tracker-Cache...");
|
||
await PollLiveAccountsAsync(stoppingToken);
|
||
_lastLivePoll = DateTime.UtcNow;
|
||
await SyncMasterTraderPositionsAsync(stoppingToken);
|
||
_lastMasterPositionPoll = DateTime.UtcNow;
|
||
_logger.Info($"Startup: MasterTraderPositions warmup abgeschlossen ({_state.MasterTraderPositions.Count} Einträge).");
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Warning($"Startup Warmup fehlgeschlagen (nicht kritisch, wird im Loop nachgeholt): {ex.Message}");
|
||
}
|
||
|
||
while (!stoppingToken.IsCancellationRequested)
|
||
{
|
||
try
|
||
{
|
||
await PollActiveTradersAsync(stoppingToken);
|
||
|
||
// Live Accounts open positions sync (Runs every 30s instead of slamming API constantly)
|
||
if ((DateTime.UtcNow - _lastLivePoll).TotalSeconds > 30)
|
||
{
|
||
await PollLiveAccountsAsync(stoppingToken);
|
||
await PollDemoExpirationsAsync(stoppingToken);
|
||
await CleanupStaleOpenOrdersAsync(stoppingToken);
|
||
_lastLivePoll = DateTime.UtcNow;
|
||
}
|
||
|
||
// Master Trader Position Tracking (Runs every 30s, offset from live poll)
|
||
if ((DateTime.UtcNow - _lastMasterPositionPoll).TotalSeconds > 30)
|
||
{
|
||
await SyncMasterTraderPositionsAsync(stoppingToken);
|
||
_lastMasterPositionPoll = DateTime.UtcNow;
|
||
}
|
||
|
||
// Background Closed Trades Sync (Runs every 2 minutes decoupled from local state)
|
||
if ((DateTime.UtcNow - _lastClosedPoll).TotalMinutes > 2)
|
||
{
|
||
await PollClosedAccountsAsync(stoppingToken);
|
||
_lastClosedPoll = DateTime.UtcNow;
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error($"TraderMonitor polling error: {ex.Message}");
|
||
}
|
||
|
||
// Global Engine Tick (dynamic queue evaluation)
|
||
await Task.Delay(1000, stoppingToken);
|
||
}
|
||
}
|
||
|
||
private async Task PollActiveTradersAsync(CancellationToken ct)
|
||
{
|
||
// Only process ACTIVE trader copies if not paused/inactive
|
||
if (_state.GlobalTradingPaused ||
|
||
(_state.DemoTradingMode == TradingMode.Inactive && _state.LiveTradingMode == TradingMode.Inactive))
|
||
{
|
||
return;
|
||
}
|
||
|
||
var activeTraders = _state.Traders.Values.Where(t => t.IsActive).ToList();
|
||
if (activeTraders.Count == 0) return;
|
||
|
||
var now = DateTime.UtcNow;
|
||
var toPoll = new List<TrackedTrader>();
|
||
|
||
bool isWssHealthy = _state.IsAlchemyHealthy;
|
||
|
||
// Calculate Dynamic Priorities
|
||
// Data API rate limit: 1000 req/10s (general).
|
||
// Worst case: 30 traders × high prio (3s) = ~100 req/10s = 10% capacity.
|
||
// With medium prio at 10s and batches of 10: well within limits.
|
||
foreach (var trader in activeTraders)
|
||
{
|
||
if (!_lastPolled.TryGetValue(trader.WalletAddress, out var lastPoll))
|
||
lastPoll = DateTime.MinValue;
|
||
|
||
double secondsSinceLastPoll = (now - lastPoll).TotalSeconds;
|
||
int requiredInterval = 10; // Medium Prio Default (Data API: 1000/10s headroom)
|
||
|
||
if (isWssHealthy)
|
||
{
|
||
// If WSS is healthy, fall back to safety-net polling
|
||
requiredInterval = 60; // 1 minute (was 2 min)
|
||
}
|
||
else
|
||
{
|
||
if (trader.TotalTrades > 20 || trader.Winrate30t >= 60.0)
|
||
requiredInterval = 3; // High Prio (unchanged — already fast)
|
||
else if (trader.TotalTrades < 5)
|
||
requiredInterval = 30; // Low Prio (was 120s)
|
||
}
|
||
|
||
if (secondsSinceLastPoll >= requiredInterval)
|
||
{
|
||
toPoll.Add(trader);
|
||
}
|
||
}
|
||
|
||
if (toPoll.Count == 0) return;
|
||
|
||
// Batch Execution (Max 10 Concurrent Requests to respect API limits)
|
||
int batchSize = 10;
|
||
for (int i = 0; i < toPoll.Count; i += batchSize)
|
||
{
|
||
if (ct.IsCancellationRequested) break;
|
||
|
||
var batch = toPoll.Skip(i).Take(batchSize);
|
||
var tasks = batch.Select(async trader =>
|
||
{
|
||
_lastPolled[trader.WalletAddress] = DateTime.UtcNow;
|
||
|
||
System.Diagnostics.Stopwatch? sw = null;
|
||
if (_state.DebugPollingLog) sw = System.Diagnostics.Stopwatch.StartNew();
|
||
|
||
var activity = await _api.GetTraderActivityAsync(trader.WalletAddress, limit: 50);
|
||
|
||
|
||
if (_state.DebugPollingLog && sw != null)
|
||
{
|
||
sw.Stop();
|
||
_logger.Debug($"[API-Profiler] Activity-Request für Trader {trader.DisplayName} dauerte {sw.ElapsedMilliseconds} ms.");
|
||
}
|
||
|
||
ProcessActivityItemsMerged(activity, trader);
|
||
});
|
||
|
||
await Task.WhenAll(tasks);
|
||
await Task.Delay(200, ct); // Tiny 200ms breath between batches
|
||
}
|
||
|
||
// Cleanup old hashes periodically (keep for 24 hours to prevent ANY duplicates)
|
||
if ((DateTime.UtcNow - _lastHashCleanup).TotalHours > 1)
|
||
{
|
||
var cutoff = DateTime.UtcNow.AddHours(-24);
|
||
var expired = _processedTxHashes.Where(x => x.Value < cutoff).Select(x => x.Key).ToList();
|
||
foreach (var k in expired) _processedTxHashes.TryRemove(k, out _);
|
||
_lastHashCleanup = DateTime.UtcNow;
|
||
}
|
||
}
|
||
|
||
public void TriggerFastBlockchainPoll(string txHash, string rpcUrl, string walletAddress)
|
||
{
|
||
var trader = _state.Traders.Values.FirstOrDefault(t => t.WalletAddress.Equals(walletAddress, StringComparison.OrdinalIgnoreCase));
|
||
if (trader == null || !trader.IsActive) return;
|
||
|
||
if (_state.EnableBlockchainParser)
|
||
{
|
||
if (!_processedTxHashes.TryAdd(txHash, DateTime.UtcNow)) return; // Debounce txHash duplicate WSS events
|
||
Task.Run(async () =>
|
||
{
|
||
try
|
||
{
|
||
var signals = await _api.ParseBlockchainTransactionAsync(txHash, rpcUrl, trader.WalletAddress);
|
||
if (signals != null && signals.Count > 0)
|
||
{
|
||
foreach (var signal in signals)
|
||
{
|
||
signal.TraderId = trader.Id;
|
||
|
||
if (_state.MarketCache.TryGetValue(signal.TokenId, out var fastCachedData))
|
||
{
|
||
signal.MarketQuestion = fastCachedData.Question;
|
||
signal.MarketSlug = fastCachedData.Slug;
|
||
signal.EndDate = fastCachedData.EndDate;
|
||
|
||
if (!string.IsNullOrEmpty(fastCachedData.ClobTokenIds))
|
||
{
|
||
try {
|
||
var tokenArr = System.Text.Json.JsonSerializer.Deserialize<System.Collections.Generic.List<string>>(fastCachedData.ClobTokenIds);
|
||
if (tokenArr != null) {
|
||
int idx = tokenArr.IndexOf(signal.TokenId);
|
||
if (idx >= 0)
|
||
{
|
||
if (!string.IsNullOrEmpty(fastCachedData.Outcomes))
|
||
{
|
||
var outcomesArr = System.Text.Json.JsonSerializer.Deserialize<System.Collections.Generic.List<string>>(fastCachedData.Outcomes);
|
||
if (outcomesArr != null && idx < outcomesArr.Count)
|
||
{
|
||
signal.Outcome = outcomesArr[idx];
|
||
}
|
||
}
|
||
|
||
// Fallback, if Outcomes array is empty or index out of bounds
|
||
if (string.IsNullOrEmpty(signal.Outcome))
|
||
{
|
||
if (idx == 0) signal.Outcome = "Yes";
|
||
else if (idx == 1) signal.Outcome = "No";
|
||
else if (idx > 1) signal.Outcome = $"Out{idx}";
|
||
}
|
||
}
|
||
}
|
||
} catch {}
|
||
}
|
||
}
|
||
else
|
||
{
|
||
// One-Time Cold Hit from Gamma API (eliminating LiteDB Full-Table-Scan blockage)
|
||
var coldItem = await _api.GetMarketByTokenIdAsync(signal.TokenId);
|
||
if (coldItem != null)
|
||
{
|
||
signal.MarketQuestion = coldItem.Question;
|
||
signal.MarketSlug = coldItem.Slug;
|
||
signal.EndDate = coldItem.EndDate;
|
||
|
||
if (!string.IsNullOrEmpty(coldItem.ClobTokenIds))
|
||
{
|
||
try {
|
||
var tokenArr = System.Text.Json.JsonSerializer.Deserialize<System.Collections.Generic.List<string>>(coldItem.ClobTokenIds);
|
||
if (tokenArr != null) {
|
||
int idx = tokenArr.IndexOf(signal.TokenId);
|
||
if (idx >= 0)
|
||
{
|
||
if (!string.IsNullOrEmpty(coldItem.Outcomes))
|
||
{
|
||
var outcomesArr = System.Text.Json.JsonSerializer.Deserialize<System.Collections.Generic.List<string>>(coldItem.Outcomes);
|
||
if (outcomesArr != null && idx < outcomesArr.Count)
|
||
{
|
||
signal.Outcome = outcomesArr[idx];
|
||
}
|
||
}
|
||
|
||
// Fallback, if Outcomes array is empty or index out of bounds
|
||
if (string.IsNullOrEmpty(signal.Outcome))
|
||
{
|
||
if (idx == 0) signal.Outcome = "Yes";
|
||
else if (idx == 1) signal.Outcome = "No";
|
||
else if (idx > 1) signal.Outcome = $"Out{idx}";
|
||
}
|
||
}
|
||
}
|
||
} catch {}
|
||
}
|
||
|
||
// Update Cache for 0ms next time
|
||
_state.MarketCache[signal.TokenId] = coldItem;
|
||
|
||
// Asynchronously persist to LiteDB snapshot without blocking Hot Path
|
||
if (_db != null)
|
||
{
|
||
_ = Task.Run(() => {
|
||
try {
|
||
_marketRepo.Upsert(coldItem);
|
||
} catch { } // Failsafe
|
||
});
|
||
}
|
||
}
|
||
else
|
||
{
|
||
// Provide fallback display if un-cached and API fails
|
||
signal.MarketQuestion = "Unbekannter Markt (Lädt...)";
|
||
signal.Outcome = signal.TokenId.Substring(0, 6);
|
||
}
|
||
}
|
||
|
||
bool added = _signalWriter.TryWrite(signal);
|
||
if (added)
|
||
{
|
||
string shareType = string.IsNullOrEmpty(signal.Outcome) ? signal.Side : signal.Outcome;
|
||
_logger.Trade($"🚨 [QUELLE: {trader.DisplayName}] WSS\n" +
|
||
$" Markt: {signal.MarketQuestion} \n" +
|
||
$" Aktion: {signal.Side} {shareType} ({signal.Size:F2} Shares @ ${signal.Price:F3})\n" +
|
||
$" Zeit: {signal.Timestamp.ToString("HH:mm:ss")} UTC\n" +
|
||
$" TxHash: {txHash}");
|
||
}
|
||
}
|
||
return; // Fast Track Successful!
|
||
}
|
||
|
||
// If it returns null, parser failed to decode this specific proxy trace. Fall back below.
|
||
_logger.Debug($"FastTrack failed for TX {txHash}. Falling back to Data API loop.");
|
||
_processedTxHashes.TryRemove(txHash, out _); // Revert the lockout so the API can process these!
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error($"FastTrack Parser Error: {ex.Message}");
|
||
_processedTxHashes.TryRemove(txHash, out _); // Revert on exception too
|
||
}
|
||
|
||
// Fallback Trigger: Delegate to normal API indexer with retry loop
|
||
TriggerManualPoll(trader.WalletAddress);
|
||
});
|
||
}
|
||
else
|
||
{
|
||
TriggerManualPoll(walletAddress); // Fallback if Parser is globally disabled
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Triggered instantly by the AlchemyWebsocketService when an EVM TransferSingle is detected.
|
||
/// </summary>
|
||
public void TriggerManualPoll(string walletAddress)
|
||
{
|
||
var trader = _state.Traders.Values.FirstOrDefault(t => t.WalletAddress.Equals(walletAddress, StringComparison.OrdinalIgnoreCase));
|
||
if (trader != null && trader.IsActive)
|
||
{
|
||
if (!_activeWssPolls.TryAdd(trader.WalletAddress, true))
|
||
{
|
||
// Ein WSS-Poll Event läuft bereits für diesen Trader! Vermeidet 429 API-Spam bei Mikro-Trades.
|
||
return;
|
||
}
|
||
|
||
// Force an immediate poll on a separate unblocked thread, completely bypassing the 1s loop delay
|
||
Task.Run(async () =>
|
||
{
|
||
try
|
||
{
|
||
int retries = 15;
|
||
while(retries > 0)
|
||
{
|
||
int hashCountBefore = _processedTxHashes.Count;
|
||
|
||
// Increase limit to 20 for WSS polling to ensure rapid batch waves aren't truncated by the API response length,
|
||
// which had previously caused trades to be completely hidden.
|
||
var activity = await _api.GetTraderActivityAsync(trader.WalletAddress, limit: 20);
|
||
|
||
ProcessActivityItemsMerged(activity, trader);
|
||
|
||
if (_processedTxHashes.Count > hashCountBefore)
|
||
{
|
||
// Hat einen neuen Trade gefunden! Schleife beenden.
|
||
break;
|
||
}
|
||
|
||
// API DB Indexer hat WSS-Event noch nicht verarbeitet, 1s warten...
|
||
await Task.Delay(1000);
|
||
retries--;
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error($"WSS API-Poll Fehler für {trader.DisplayName}: {ex.Message}");
|
||
}
|
||
finally
|
||
{
|
||
_activeWssPolls.TryRemove(trader.WalletAddress, out _);
|
||
}
|
||
});
|
||
}
|
||
}
|
||
|
||
private async Task ExecuteRestAutoRedeemLive(AccountState acc, Position pos)
|
||
{
|
||
string redeemKey = $"{acc.AccountId}_{pos.TokenId}";
|
||
if (pos.Size < 5.0m)
|
||
{
|
||
var state = _restRedeemAttempts.GetOrAdd(redeemKey, _ => (0, DateTime.MinValue));
|
||
int newCount = state.Count + 1;
|
||
_restRedeemAttempts[redeemKey] = (newCount, DateTime.UtcNow);
|
||
if (newCount <= 1) _logger.Warning($"[REST AUTO REDEEM] Position {pos.MarketQuestion} zu klein für Limit Order (< 5 Shares). Max. 1 Retry in 5 Min.");
|
||
return;
|
||
}
|
||
|
||
_restRedeemAttempts.AddOrUpdate(redeemKey, _ => (1, DateTime.UtcNow), (_, old) => (old.Count + 1, DateTime.UtcNow));
|
||
|
||
try
|
||
{
|
||
decimal expectedFillPrice = acc.PreRedeemLimit;
|
||
decimal amountUsdc = Math.Max(pos.Size * expectedFillPrice, 0.01m);
|
||
var result = await _clob.PlaceOrderAsync(acc, pos.TokenId, "SELL", amountUsdc, expectedFillPrice, "GTC", false, false);
|
||
|
||
if (result == "OK")
|
||
{
|
||
_logger.Info($"✅ REST Auto-Redeem Sell sent for {acc.Name} at exact Limit {expectedFillPrice:F3} USD (GTC).");
|
||
}
|
||
else
|
||
{
|
||
_logger.Error($"❌ REST Auto-Redeem failed or rejected: {result}.");
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error($"REST Auto Redeem Exception: {ex.Message}");
|
||
}
|
||
}
|
||
|
||
private async Task PollDemoExpirationsAsync(CancellationToken ct)
|
||
{
|
||
var demoAccounts = _state.Accounts.Values.Where(a => a.IsDemo && a.IsActive).ToList();
|
||
if (demoAccounts.Count == 0) return;
|
||
|
||
foreach (var acc in demoAccounts)
|
||
{
|
||
if (ct.IsCancellationRequested) break;
|
||
|
||
// Check positions that are near expiry, recently expired, or have no expiry but have a slug
|
||
var checkPositions = acc.OpenPositions.Values.Where(p =>
|
||
!string.IsNullOrEmpty(p.MarketSlug) &&
|
||
(
|
||
// Has expiry and is within check window (-1 day to +30 days)
|
||
(p.ExpiryDate.HasValue &&
|
||
(DateTime.UtcNow - p.ExpiryDate.Value).TotalDays > -1 &&
|
||
(DateTime.UtcNow - p.ExpiryDate.Value).TotalDays < 30)
|
||
||
|
||
// No expiry date at all — always check via API
|
||
!p.ExpiryDate.HasValue
|
||
)).ToList();
|
||
|
||
foreach (var pos in checkPositions)
|
||
{
|
||
var (isClosed, isWinner) = await _api.CheckMarketResolutionAsync(pos.MarketSlug, pos.TokenId);
|
||
if (isClosed)
|
||
{
|
||
decimal exitPrice = isWinner ? 1.0m : 0.0m;
|
||
_logger.Info($"🏆 Demo Market {pos.MarketQuestion} aufgelöst! Auszahlung: ${(exitPrice * pos.Size):F2}");
|
||
|
||
var signal = new CopySignal
|
||
{
|
||
TraderId = 0,
|
||
TokenId = pos.TokenId,
|
||
MarketSlug = pos.MarketSlug,
|
||
MarketQuestion = pos.MarketQuestion,
|
||
Outcome = pos.Outcome,
|
||
Side = "SELL",
|
||
Price = exitPrice,
|
||
Size = pos.Size,
|
||
Timestamp = DateTime.UtcNow,
|
||
Reason = "Market Resolved"
|
||
};
|
||
|
||
_signalWriter.TryWrite(signal);
|
||
await Task.Delay(500, ct);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private async Task PollClosedAccountsAsync(CancellationToken ct)
|
||
{
|
||
var liveAccounts = _state.Accounts.Values.Where(a => !a.IsDemo && a.IsActive && !string.IsNullOrEmpty(a.WalletAddress)).ToList();
|
||
if (liveAccounts.Count == 0) return;
|
||
|
||
foreach (var acc in liveAccounts)
|
||
{
|
||
if (ct.IsCancellationRequested) break;
|
||
|
||
var closedPositions = await _api.SyncClosedPositionsAsync(acc.WalletAddress, 50);
|
||
if (closedPositions == null || closedPositions.Count == 0) continue;
|
||
|
||
foreach (var cm in closedPositions)
|
||
{
|
||
string asset = cm.TryGetProperty("asset", out var ap) ? ap.GetString() ?? "" : "";
|
||
if (string.IsNullOrEmpty(asset)) continue;
|
||
|
||
// 1. Immediately extract and clear from OpenPositions cache
|
||
int resolvedSourceId = 0;
|
||
if (acc.OpenPositions.TryRemove(asset, out var removedPos))
|
||
{
|
||
resolvedSourceId = removedPos.SourceTraderId;
|
||
}
|
||
|
||
// 2. Fallback: Wenn TryRemove fehlschlägt (Race Condition mit PollLiveAccountsAsync),
|
||
// SourceTraderId aus der MongoDB open_positions-Tabelle wiederherstellen.
|
||
if (resolvedSourceId <= 0 && _db != null)
|
||
{
|
||
try
|
||
{
|
||
var dbPos = _positionRepo.FindLive(acc.AccountId, asset);
|
||
if (dbPos != null && dbPos.SourceTraderId > 0)
|
||
{
|
||
resolvedSourceId = dbPos.SourceTraderId;
|
||
// Auch removedPos füllen für MarketQuestion/Outcome weiter unten
|
||
if (removedPos == null) removedPos = dbPos;
|
||
}
|
||
}
|
||
catch { }
|
||
}
|
||
|
||
// 3. Fallback: Prüfe ob ein anderer Account dieselbe TokenId mit SourceTraderId hat
|
||
if (resolvedSourceId <= 0)
|
||
{
|
||
foreach (var otherAcc in _state.Accounts.Values)
|
||
{
|
||
if (otherAcc.AccountId == acc.AccountId) continue;
|
||
if (otherAcc.OpenPositions.TryGetValue(asset, out var otherPos) && otherPos.SourceTraderId > 0)
|
||
{
|
||
resolvedSourceId = otherPos.SourceTraderId;
|
||
if (removedPos == null) removedPos = otherPos;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
// 2. Prevent DB Duplicates! Fast check in MongoDB if available.
|
||
bool dbExists = false;
|
||
if (_db != null)
|
||
{
|
||
var col = _db.GetCollection<ClosedTrade>("closed_trades");
|
||
dbExists = col.Find(x => x.AccountId == acc.AccountId && x.TokenId == asset).FirstOrDefault() != null;
|
||
}
|
||
|
||
string duplicateKey = $"{acc.AccountId}_{asset}";
|
||
if (!dbExists && !_processedClosures.ContainsKey(duplicateKey))
|
||
{
|
||
decimal realizedPnl = 0m, entryPrice = 0m, size = 0m, exitPrice = 0m;
|
||
DateTime resolvedTimestamp = DateTime.UtcNow;
|
||
|
||
if (cm.TryGetProperty("timestamp", out var tsProp) ||
|
||
cm.TryGetProperty("updatedAt", out tsProp) ||
|
||
cm.TryGetProperty("closedAt", out tsProp) ||
|
||
cm.TryGetProperty("createdAt", out tsProp))
|
||
{
|
||
try
|
||
{
|
||
if (tsProp.ValueKind == JsonValueKind.Number)
|
||
{
|
||
long ts = tsProp.GetInt64();
|
||
if (ts > 9999999999) ts /= 1000;
|
||
resolvedTimestamp = DateTimeOffset.FromUnixTimeSeconds(ts).UtcDateTime;
|
||
}
|
||
else if (tsProp.ValueKind == JsonValueKind.String && DateTime.TryParse(tsProp.GetString(), out var parsedDate))
|
||
{
|
||
resolvedTimestamp = parsedDate.ToUniversalTime();
|
||
}
|
||
}
|
||
catch { }
|
||
}
|
||
|
||
if (cm.TryGetProperty("realizedPnl", out var rPnlProp)) realizedPnl = ParseDecimal(rPnlProp);
|
||
if (cm.TryGetProperty("avgPrice", out var epProp)) entryPrice = ParseDecimal(epProp);
|
||
if (cm.TryGetProperty("totalBought", out var szProp)) size = ParseDecimal(szProp);
|
||
|
||
// Approximate exit price based on PnL
|
||
decimal investment = size * entryPrice;
|
||
if (size > 0) exitPrice = (investment + realizedPnl) / size;
|
||
|
||
string orderKey = $"{acc.AccountId}_{asset}";
|
||
bool soldByUs = _state.PendingOrderTimestamps.ContainsKey(orderKey);
|
||
string exitReason = soldByUs ? "Master Trader Sold" : "Manuell Geschlossen / System";
|
||
|
||
var ctRecord = new ClosedTrade
|
||
{
|
||
TradeId = _state.GetNextTradeId(),
|
||
AccountId = acc.AccountId,
|
||
SourceTraderId = resolvedSourceId,
|
||
IsDemo = false,
|
||
TokenId = asset,
|
||
MarketSlug = removedPos != null ? removedPos.MarketSlug : (cm.TryGetProperty("slug", out var sp) ? sp.GetString() ?? "" : ""),
|
||
MarketQuestion = removedPos != null ? removedPos.MarketQuestion : (cm.TryGetProperty("title", out var tp) ? tp.GetString() ?? "Unknown Market" : "Unknown Market"),
|
||
Outcome = cm.TryGetProperty("outcome", out var op) ? op.GetString() ?? "" : "",
|
||
Side = "SELL",
|
||
EntryPrice = entryPrice,
|
||
ExitPrice = exitPrice,
|
||
Size = size,
|
||
RealizedPnl = realizedPnl,
|
||
PnlPercent = investment > 0 ? (realizedPnl / investment * 100m) : 0m,
|
||
OpenedAt = removedPos?.OpenedAt ?? resolvedTimestamp,
|
||
ClosedAt = resolvedTimestamp,
|
||
ExitReason = exitReason
|
||
};
|
||
|
||
_processedClosures.TryAdd(duplicateKey, true);
|
||
_closedTradeWriter.TryWrite(ctRecord);
|
||
|
||
_logger.Info($"🏆 Trade {ctRecord.MarketQuestion} synchronisiert (Hintergrund)! PnL: ${realizedPnl:F2}");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private async Task PollLiveAccountsAsync(CancellationToken ct)
|
||
{
|
||
// Always sync live positions so the Dashboard UI accurately reflects open PnL and portfolio balance
|
||
var liveAccounts = _state.Accounts.Values.Where(a => !a.IsDemo && a.IsActive && !string.IsNullOrEmpty(a.WalletAddress)).ToList();
|
||
if (liveAccounts.Count == 0) return;
|
||
|
||
foreach (var acc in liveAccounts)
|
||
{
|
||
if (ct.IsCancellationRequested) break;
|
||
|
||
var posList = await _api.SyncOpenPositionsAsync(acc.WalletAddress);
|
||
if (posList == null) continue; // Skip on API error
|
||
|
||
var currentTokens = new HashSet<string>();
|
||
|
||
foreach (var posJson in posList)
|
||
{
|
||
string asset = posJson.TryGetProperty("asset", out var ap) ? ap.GetString() ?? "" : "";
|
||
if (string.IsNullOrEmpty(asset)) continue;
|
||
|
||
decimal size = 0m, entryPrice = 0m, amountUsd = 0m, curPrice = 0m, curValue = 0m;
|
||
if (posJson.TryGetProperty("size", out var sprop)) size = ParseDecimal(sprop);
|
||
|
||
if (size < 0.001m) continue; // Exclude closed positions from API so Live Sync can process them as closed
|
||
|
||
currentTokens.Add(asset);
|
||
|
||
string slug = posJson.TryGetProperty("slug", out var sp) ? sp.GetString() ?? "" : "";
|
||
string title = posJson.TryGetProperty("title", out var tp) ? tp.GetString() ?? "" : "";
|
||
string opp = posJson.TryGetProperty("oppositeOutcome", out var op) ? op.GetString() ?? "" : "No";
|
||
|
||
if (posJson.TryGetProperty("avgPrice", out var aprop)) entryPrice = ParseDecimal(aprop);
|
||
// Critical Fix: "totalBought" is size. "initialValue" is original USD investment cost.
|
||
if (posJson.TryGetProperty("initialValue", out var tbprop)) amountUsd = ParseDecimal(tbprop);
|
||
if (posJson.TryGetProperty("curPrice", out var cpprop)) curPrice = ParseDecimal(cpprop);
|
||
if (posJson.TryGetProperty("currentValue", out var cvprop)) curValue = ParseDecimal(cvprop);
|
||
|
||
DateTime? expiry = null;
|
||
if (posJson.TryGetProperty("endDate", out var ep))
|
||
{
|
||
if (DateTime.TryParse(ep.GetString(), out var ed)) expiry = DateTime.SpecifyKind(ed.Date, DateTimeKind.Utc);
|
||
}
|
||
|
||
if (acc.OpenPositions.TryGetValue(asset, out var existing))
|
||
{
|
||
existing.Size = size;
|
||
existing.EntryPrice = entryPrice;
|
||
existing.AmountUsd = amountUsd;
|
||
existing.CurrentPrice = curPrice;
|
||
existing.CurrentValueUsd = curValue;
|
||
if (expiry.HasValue) existing.ExpiryDate = expiry;
|
||
|
||
try { _positionRepo.UpsertLive(acc.AccountId, existing); } catch { }
|
||
|
||
// Auto-Redeem Fallback via REST
|
||
if (acc.PreRedeemLimit > 0 && curPrice >= acc.PreRedeemLimit && acc.IsActive)
|
||
{
|
||
string redeemKey = $"{acc.AccountId}_{asset}";
|
||
bool allowAttempt = true;
|
||
|
||
if (_restRedeemAttempts.TryGetValue(redeemKey, out var state))
|
||
{
|
||
if (state.Count >= 2) allowAttempt = false;
|
||
if ((DateTime.UtcNow - state.LastAttempt).TotalMinutes < 5) allowAttempt = false;
|
||
}
|
||
|
||
if (allowAttempt)
|
||
{
|
||
if (!acc.IsDemo && _state.LiveTradingMode == TradingMode.Active)
|
||
{
|
||
_logger.Trade($"🚨 [REST AUTO REDEEM] {acc.Name} | {existing.MarketQuestion} | Preis >= {acc.PreRedeemLimit}");
|
||
_ = Task.Run(async () => await ExecuteRestAutoRedeemLive(acc, existing));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
else
|
||
{
|
||
// --- Smart Master-Trader Zuordnung ---
|
||
// Versuch den Quell-Trader zu identifizieren statt pauschal "Live Sync" zu vergeben
|
||
int resolvedTraderId = 0;
|
||
string resolvedTraderName = "Live Sync";
|
||
string resolvedTraderAddress = "";
|
||
|
||
// 1. Prüfe PendingOrderTimestamps (CopyTradingEngine hat diese Order kürzlich platziert)
|
||
string orderKey = $"{acc.AccountId}_{asset}";
|
||
if (_state.PendingOrderTimestamps.TryGetValue(orderKey, out var pending) && pending.SourceTraderId > 0)
|
||
{
|
||
resolvedTraderId = pending.SourceTraderId;
|
||
if (_state.Traders.TryGetValue(resolvedTraderId, out var pendingTrader))
|
||
{
|
||
resolvedTraderName = pendingTrader.DisplayName;
|
||
resolvedTraderAddress = pendingTrader.WalletAddress;
|
||
}
|
||
}
|
||
|
||
// 2. Prüfe ob ein anderer Account dieselbe TokenId bereits mit echtem SourceTraderId hat
|
||
if (resolvedTraderId == 0)
|
||
{
|
||
foreach (var otherAcc in _state.Accounts.Values)
|
||
{
|
||
if (otherAcc.AccountId == acc.AccountId) continue;
|
||
if (otherAcc.OpenPositions.TryGetValue(asset, out var otherPos) && otherPos.SourceTraderId > 0)
|
||
{
|
||
resolvedTraderId = otherPos.SourceTraderId;
|
||
resolvedTraderName = otherPos.SourceTraderName;
|
||
resolvedTraderAddress = otherPos.SourceTraderAddress;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
// 3. Prüfe ob ein anderer Account denselben Markt (Slug+Outcome) bereits zugeordnet hat
|
||
if (resolvedTraderId == 0 && !string.IsNullOrEmpty(slug))
|
||
{
|
||
string resolvedOutcome = opp == "Yes" ? "No" : "Yes";
|
||
foreach (var otherAcc in _state.Accounts.Values)
|
||
{
|
||
if (otherAcc.AccountId == acc.AccountId) continue;
|
||
var match = otherAcc.OpenPositions.Values.FirstOrDefault(p =>
|
||
p.MarketSlug == slug && p.Outcome == resolvedOutcome && p.SourceTraderId > 0);
|
||
if (match != null)
|
||
{
|
||
resolvedTraderId = match.SourceTraderId;
|
||
resolvedTraderName = match.SourceTraderName;
|
||
resolvedTraderAddress = match.SourceTraderAddress;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
DateTime resolvedOpenedAt = DateTime.UtcNow;
|
||
|
||
// 4. Prüfe lokale DB-Tabelle für den Fall eines Programm-Neustarts / API-Syncs
|
||
if (resolvedTraderId == 0 && _db != null)
|
||
{
|
||
try
|
||
{
|
||
var dbPos = _positionRepo.FindLive(acc.AccountId, asset);
|
||
if (dbPos != null)
|
||
{
|
||
if (dbPos.SourceTraderId > 0)
|
||
{
|
||
resolvedTraderId = dbPos.SourceTraderId;
|
||
resolvedTraderName = dbPos.SourceTraderName;
|
||
resolvedTraderAddress = dbPos.SourceTraderAddress;
|
||
}
|
||
if (dbPos.OpenedAt > DateTime.MinValue)
|
||
{
|
||
resolvedOpenedAt = dbPos.OpenedAt;
|
||
}
|
||
}
|
||
}
|
||
catch { }
|
||
}
|
||
|
||
var newPos = new Position
|
||
{
|
||
TokenId = asset,
|
||
MarketSlug = slug,
|
||
MarketQuestion = title,
|
||
Outcome = opp == "Yes" ? "No" : "Yes",
|
||
SourceTraderId = resolvedTraderId,
|
||
SourceTraderName = resolvedTraderName,
|
||
SourceTraderAddress = resolvedTraderAddress,
|
||
Side = "BUY",
|
||
Size = size,
|
||
EntryPrice = entryPrice,
|
||
AmountUsd = amountUsd,
|
||
CurrentPrice = curPrice,
|
||
CurrentValueUsd = curValue,
|
||
ExpiryDate = expiry,
|
||
OpenedAt = resolvedOpenedAt
|
||
};
|
||
acc.OpenPositions.TryAdd(asset, newPos);
|
||
try { _positionRepo.UpsertLive(acc.AccountId, newPos); } catch { }
|
||
|
||
if (resolvedTraderId > 0)
|
||
_logger.Info($"🌐 Live Position erkannt: {title} ({newPos.Outcome}) - ${amountUsd} - Account: {acc.Name} [Zugeordnet: {resolvedTraderName}]");
|
||
else
|
||
_logger.Info($"🌐 Live Position erkannt: {title} ({newPos.Outcome}) - ${amountUsd} - Account: {acc.Name} [Kein Master-Trader zugeordnet]");
|
||
}
|
||
}
|
||
|
||
var tokensToInvestigate = acc.OpenPositions
|
||
.Where(kvp => !currentTokens.Contains(kvp.Key))
|
||
.Where(kvp => (DateTime.UtcNow - kvp.Value.OpenedAt).TotalMinutes > 5)
|
||
.Select(kvp => kvp.Key)
|
||
.ToList();
|
||
if (tokensToInvestigate.Count > 0)
|
||
{
|
||
var closedPositions = await _api.SyncClosedPositionsAsync(acc.WalletAddress, 50);
|
||
|
||
foreach (var k in tokensToInvestigate)
|
||
{
|
||
if (acc.OpenPositions.TryGetValue(k, out var removedPos))
|
||
{
|
||
// 2. Fallback: Auch im Live Sync: Wenn SourceTraderId verloren ging,
|
||
// aus der MongoDB open_positions-Tabelle wiederherstellen.
|
||
if (removedPos.SourceTraderId <= 0 && _db != null)
|
||
{
|
||
try
|
||
{
|
||
var dbPos = _positionRepo.FindLive(acc.AccountId, k);
|
||
if (dbPos != null && dbPos.SourceTraderId > 0)
|
||
{
|
||
removedPos.SourceTraderId = dbPos.SourceTraderId;
|
||
}
|
||
}
|
||
catch { }
|
||
}
|
||
|
||
JsonElement? matchedClose = null;
|
||
foreach (var cm in closedPositions)
|
||
{
|
||
if (cm.TryGetProperty("asset", out var ap) && ap.GetString() == k)
|
||
{
|
||
matchedClose = cm;
|
||
break;
|
||
}
|
||
}
|
||
|
||
if (matchedClose.HasValue)
|
||
{
|
||
acc.OpenPositions.TryRemove(k, out _); // Safe removal!
|
||
try { _positionRepo.DeleteLive(acc.AccountId, k); } catch { }
|
||
|
||
decimal realizedPnl = 0m;
|
||
DateTime resolvedClosedAt = DateTime.UtcNow;
|
||
|
||
if (matchedClose.Value.TryGetProperty("timestamp", out var tsProp) ||
|
||
matchedClose.Value.TryGetProperty("updatedAt", out tsProp) ||
|
||
matchedClose.Value.TryGetProperty("closedAt", out tsProp) ||
|
||
matchedClose.Value.TryGetProperty("createdAt", out tsProp))
|
||
{
|
||
try
|
||
{
|
||
if (tsProp.ValueKind == JsonValueKind.Number)
|
||
{
|
||
long ts = tsProp.GetInt64();
|
||
if (ts > 9999999999) ts /= 1000;
|
||
resolvedClosedAt = DateTimeOffset.FromUnixTimeSeconds(ts).UtcDateTime;
|
||
}
|
||
else if (tsProp.ValueKind == JsonValueKind.String && DateTime.TryParse(tsProp.GetString(), out var parsedDate))
|
||
{
|
||
resolvedClosedAt = parsedDate.ToUniversalTime();
|
||
}
|
||
}
|
||
catch { }
|
||
}
|
||
|
||
if (matchedClose.Value.TryGetProperty("realizedPnl", out var rPnlProp)) realizedPnl = ParseDecimal(rPnlProp);
|
||
|
||
_state.GlobalPnl += realizedPnl;
|
||
decimal exitPrice = removedPos.Size > 0 ? (removedPos.AmountUsd + realizedPnl) / removedPos.Size : 0m;
|
||
|
||
string duplicateKey = $"{acc.AccountId}_{removedPos.TokenId}";
|
||
if (!_processedClosures.ContainsKey(duplicateKey))
|
||
{
|
||
_logger.Info($"🏆 Live Market {removedPos.MarketQuestion} geschlossen! PnL: ${(realizedPnl):F2}");
|
||
|
||
string orderKey = $"{acc.AccountId}_{removedPos.TokenId}";
|
||
bool soldByUs = _state.PendingOrderTimestamps.ContainsKey(orderKey);
|
||
string exitReason = soldByUs ? "Master Trader Sold" : "Market Resolved";
|
||
|
||
var ctRecord = new ClosedTrade
|
||
{
|
||
TradeId = _state.GetNextTradeId(),
|
||
AccountId = acc.AccountId,
|
||
SourceTraderId = removedPos.SourceTraderId,
|
||
IsDemo = false,
|
||
MarketSlug = removedPos.MarketSlug,
|
||
MarketQuestion = removedPos.MarketQuestion,
|
||
Outcome = removedPos.Outcome,
|
||
Side = "SELL",
|
||
EntryPrice = removedPos.EntryPrice,
|
||
ExitPrice = exitPrice,
|
||
Size = removedPos.Size,
|
||
RealizedPnl = realizedPnl,
|
||
PnlPercent = removedPos.AmountUsd > 0 ? (realizedPnl / removedPos.AmountUsd * 100m) : 0m,
|
||
OpenedAt = removedPos.OpenedAt,
|
||
ClosedAt = resolvedClosedAt,
|
||
ExitReason = exitReason
|
||
};
|
||
|
||
if (soldByUs) _state.PendingOrderTimestamps.TryRemove(orderKey, out _);
|
||
|
||
_processedClosures.TryAdd(duplicateKey, true);
|
||
_closedTradeWriter.TryWrite(ctRecord);
|
||
}
|
||
}
|
||
else
|
||
{
|
||
var (isClosed, isWinner) = await _api.CheckMarketResolutionAsync(removedPos.MarketSlug, removedPos.TokenId);
|
||
|
||
if (isClosed)
|
||
{
|
||
acc.OpenPositions.TryRemove(k, out _); // Safe removal!
|
||
try { _positionRepo.DeleteLive(acc.AccountId, k); } catch { }
|
||
|
||
decimal exitPrice = isWinner ? 1.0m : 0.0m;
|
||
decimal exitUsd = removedPos.Size * exitPrice;
|
||
decimal realizedPnl = exitUsd - removedPos.AmountUsd;
|
||
|
||
_state.GlobalPnl += realizedPnl;
|
||
|
||
string duplicateKey = $"{acc.AccountId}_{removedPos.TokenId}";
|
||
if (!_processedClosures.ContainsKey(duplicateKey))
|
||
{
|
||
_logger.Info($"🏆 Live Market {removedPos.MarketQuestion} aufgelöst (Fallback)! Auszahlung: ${(exitPrice * removedPos.Size):F2}");
|
||
|
||
string orderKey = $"{acc.AccountId}_{removedPos.TokenId}";
|
||
bool soldByUs = _state.PendingOrderTimestamps.ContainsKey(orderKey);
|
||
string exitReason = soldByUs ? "Master Trader Sold" : "Market Resolved";
|
||
|
||
var ctRecord = new ClosedTrade
|
||
{
|
||
TradeId = _state.GetNextTradeId(),
|
||
AccountId = acc.AccountId,
|
||
SourceTraderId = removedPos.SourceTraderId,
|
||
IsDemo = false,
|
||
MarketSlug = removedPos.MarketSlug,
|
||
MarketQuestion = removedPos.MarketQuestion,
|
||
Outcome = removedPos.Outcome,
|
||
Side = "SELL",
|
||
EntryPrice = removedPos.EntryPrice,
|
||
ExitPrice = exitPrice,
|
||
Size = removedPos.Size,
|
||
RealizedPnl = realizedPnl,
|
||
PnlPercent = removedPos.AmountUsd > 0 ? (realizedPnl / removedPos.AmountUsd * 100m) : 0m,
|
||
OpenedAt = removedPos.OpenedAt,
|
||
ClosedAt = DateTime.UtcNow,
|
||
ExitReason = exitReason
|
||
};
|
||
|
||
if (soldByUs) _state.PendingOrderTimestamps.TryRemove(orderKey, out _);
|
||
|
||
_processedClosures.TryAdd(duplicateKey, true);
|
||
_closedTradeWriter.TryWrite(ctRecord);
|
||
}
|
||
|
||
if (isWinner)
|
||
{
|
||
/*
|
||
* DEATIVIERT: Automatischer Redeem via Python Script ist vorerst pausiert.
|
||
* User kann die gewonnenen Shares per Klick im Polymarket Web-Interface redeemen.
|
||
* Die Datenbank hat die PnL trotzdem bereits korrekt aufgezeichnet!
|
||
*
|
||
try
|
||
{
|
||
System.Diagnostics.Process.Start(new System.Diagnostics.ProcessStartInfo
|
||
{
|
||
FileName = "python",
|
||
Arguments = $"redeem_markets.py {removedPos.TokenId} {acc.ApiKey} {acc.PrivateKey} {acc.ApiPassphrase}",
|
||
UseShellExecute = false,
|
||
CreateNoWindow = true
|
||
});
|
||
_logger.Info($"Python Redeem Script für Token {removedPos.TokenId} asynchron ausgeführt.");
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error($"Fehler beim Starten von redeem_markets.py: {ex.Message}");
|
||
}
|
||
*/
|
||
_logger.Info($"🏆 Token {removedPos.TokenId} bereit für manuellen Redeem via Polymarket-Webseite. (P&L wurde bereits gebucht).");
|
||
}
|
||
}
|
||
else
|
||
{
|
||
var ageMinutes = (DateTime.UtcNow - removedPos.OpenedAt).TotalMinutes;
|
||
if (ageMinutes >= 60)
|
||
{
|
||
if (acc.OpenPositions.TryRemove(k, out _))
|
||
{
|
||
try { _positionRepo.DeleteLive(acc.AccountId, k); } catch { }
|
||
_logger.Info($"🌐 Live Position {removedPos.MarketQuestion} final entfernt (Ext. Verkauft/Wartend nach {ageMinutes:F0} Min.)");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
await Task.Delay(500, ct);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Periodically syncs how many shares each master trader holds for tokens we've copied.
|
||
/// This data is used by CopyTradingEngine to determine if a SELL signal is a partial sell (ignore) or a full exit (copy).
|
||
/// </summary>
|
||
private async Task SyncMasterTraderPositionsAsync(CancellationToken ct)
|
||
{
|
||
try
|
||
{
|
||
// Step 1: Collect all (TraderId -> Set<TokenId>) from our open positions across all accounts
|
||
var traderTokenMap = new Dictionary<int, HashSet<string>>();
|
||
|
||
foreach (var acc in _state.Accounts.Values.Where(a => a.IsActive))
|
||
{
|
||
foreach (var pos in acc.OpenPositions.Values)
|
||
{
|
||
if (pos.SourceTraderId <= 0 || string.IsNullOrEmpty(pos.TokenId)) continue;
|
||
|
||
if (!traderTokenMap.TryGetValue(pos.SourceTraderId, out var tokens))
|
||
{
|
||
tokens = new HashSet<string>();
|
||
traderTokenMap[pos.SourceTraderId] = tokens;
|
||
}
|
||
tokens.Add(pos.TokenId);
|
||
}
|
||
}
|
||
|
||
if (traderTokenMap.Count == 0) return;
|
||
|
||
// Step 2: For each trader, fetch their current positions and update the cache
|
||
foreach (var (traderId, tokenIds) in traderTokenMap)
|
||
{
|
||
if (ct.IsCancellationRequested) break;
|
||
|
||
if (!_state.Traders.TryGetValue(traderId, out var trader) || string.IsNullOrEmpty(trader.WalletAddress))
|
||
continue;
|
||
|
||
var positionSizes = await _api.GetTraderPositionSizesAsync(trader.WalletAddress, tokenIds);
|
||
|
||
// Update cache for all tokens this trader is supposed to hold
|
||
foreach (var tokenId in tokenIds)
|
||
{
|
||
string key = $"{traderId}_{tokenId}";
|
||
decimal shares = positionSizes.ContainsKey(tokenId) ? positionSizes[tokenId] : 0m;
|
||
_state.MasterTraderPositions[key] = (shares, DateTime.UtcNow);
|
||
}
|
||
|
||
await Task.Delay(200, ct); // Brief delay between traders to avoid rate limits
|
||
}
|
||
}
|
||
catch (OperationCanceledException) { }
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error($"SyncMasterTraderPositions Error: {ex.Message}");
|
||
}
|
||
}
|
||
|
||
private async Task CleanupStaleOpenOrdersAsync(CancellationToken ct)
|
||
{
|
||
var keysToProcess = _state.PendingOrderTimestamps.ToArray();
|
||
if (keysToProcess.Length == 0) return;
|
||
|
||
foreach (var kvp in keysToProcess)
|
||
{
|
||
if (ct.IsCancellationRequested) break;
|
||
|
||
var parts = kvp.Key.Split('_', 2);
|
||
if (parts.Length != 2 || !int.TryParse(parts[0], out int accountId)) continue;
|
||
string tokenId = parts[1];
|
||
|
||
if (!_state.Accounts.TryGetValue(accountId, out var account) || account.IsDemo) continue;
|
||
|
||
// Determine timeout based on trader category
|
||
int timeoutMinutes = 30; // Default: 30 min
|
||
if (_state.Traders.TryGetValue(kvp.Value.SourceTraderId, out var trader) && trader.Category == "HF")
|
||
{
|
||
timeoutMinutes = 3; // HF Trader: 3 min
|
||
}
|
||
|
||
double ageMinutes = (DateTime.UtcNow - kvp.Value.PlacedAt).TotalMinutes;
|
||
if (ageMinutes < timeoutMinutes) continue;
|
||
|
||
// Order is stale — cancel it
|
||
try
|
||
{
|
||
var openOrders = await _clob.GetOpenOrdersAsync(account, tokenId);
|
||
if (openOrders.Count > 0)
|
||
{
|
||
foreach (var order in openOrders)
|
||
{
|
||
_logger.Warning($"⏰ [{account.Name}] Stale Order Timeout ({ageMinutes:F0} min > {timeoutMinutes} min). Storniere Order {order.Id} für {tokenId.Substring(0, Math.Min(10, tokenId.Length))}...");
|
||
await _clob.CancelOrderAsync(account, order.Id);
|
||
}
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error($"Stale Order Cleanup Error: {ex.Message}");
|
||
}
|
||
|
||
// Remove from tracking regardless (even if cancel failed, we don't want to spam retries)
|
||
_state.PendingOrderTimestamps.TryRemove(kvp.Key, out _);
|
||
}
|
||
}
|
||
|
||
private decimal ParseDecimal(JsonElement prop)
|
||
{
|
||
if (prop.ValueKind == JsonValueKind.Number) return prop.GetDecimal();
|
||
if (prop.ValueKind == JsonValueKind.String && decimal.TryParse(prop.GetString(), System.Globalization.NumberStyles.Any, System.Globalization.CultureInfo.InvariantCulture, out var parsed)) return parsed;
|
||
return 0m;
|
||
}
|
||
|
||
private void ProcessActivityItemsMerged(List<JsonElement> activity, TrackedTrader trader)
|
||
{
|
||
try
|
||
{
|
||
// Group fragmented trades by txHash + asset + side so dust-fills do not block major fills
|
||
var validItems = activity.Where(act => {
|
||
string type = act.GetProperty("type").GetString()?.ToUpper() ?? "";
|
||
return type == "TRADE" || type == "BUY" || type == "SELL";
|
||
}).ToList();
|
||
|
||
var grouped = validItems.GroupBy(act => {
|
||
string tx = act.GetProperty("transactionHash").GetString() ?? "";
|
||
string sideStr = act.GetProperty("type").GetString() ?? "";
|
||
if (act.TryGetProperty("side", out var sProp) && sProp.ValueKind == JsonValueKind.String) sideStr = sProp.GetString() ?? sideStr;
|
||
else if (act.TryGetProperty("action", out var acProp) && acProp.ValueKind == JsonValueKind.String) sideStr = acProp.GetString() ?? sideStr;
|
||
else if (act.TryGetProperty("tradeType", out var ttProp) && ttProp.ValueKind == JsonValueKind.String) sideStr = ttProp.GetString() ?? sideStr;
|
||
|
||
string asset = "";
|
||
if (act.TryGetProperty("asset", out var ap) && ap.ValueKind == JsonValueKind.String) asset = ap.GetString() ?? "";
|
||
if (string.IsNullOrEmpty(asset) && act.TryGetProperty("tokenId", out var tidProp) && tidProp.ValueKind == JsonValueKind.String) asset = tidProp.GetString() ?? "";
|
||
|
||
string parsedSide = sideStr.ToUpper().Contains("SELL") ? "SELL" : "BUY";
|
||
return $"{tx}_{asset}_{parsedSide}";
|
||
}).ToList();
|
||
|
||
foreach (var group in grouped)
|
||
{
|
||
string uniqueTradeKey = group.Key;
|
||
if (string.IsNullOrEmpty(uniqueTradeKey) || uniqueTradeKey.StartsWith("_")) continue;
|
||
|
||
var parts = uniqueTradeKey.Split('_');
|
||
if (parts.Length < 3) continue;
|
||
|
||
string txHash = parts[0];
|
||
string asset = parts[1];
|
||
string parsedSide = parts.Last(); // "BUY" or "SELL"
|
||
|
||
var elements = group.ToList();
|
||
var firstAct = elements.First(); // we take metadata like timestamps/titles from the first item
|
||
|
||
// Accumulate size and define weighted price
|
||
decimal totalSize = 0m;
|
||
decimal weightedPriceSum = 0m;
|
||
|
||
foreach (var act in elements)
|
||
{
|
||
decimal price = 0m;
|
||
if (act.TryGetProperty("price", out var priceProp))
|
||
{
|
||
if (priceProp.ValueKind == JsonValueKind.Number) price = priceProp.GetDecimal();
|
||
else if (priceProp.ValueKind == JsonValueKind.String) decimal.TryParse(priceProp.GetString(), out price);
|
||
}
|
||
|
||
decimal size = 0m;
|
||
if (act.TryGetProperty("size", out var sizeProp))
|
||
{
|
||
if (sizeProp.ValueKind == JsonValueKind.Number) size = sizeProp.GetDecimal();
|
||
else if (sizeProp.ValueKind == JsonValueKind.String) decimal.TryParse(sizeProp.GetString(), out size);
|
||
}
|
||
|
||
totalSize += size;
|
||
weightedPriceSum += (price * size);
|
||
}
|
||
|
||
if (totalSize <= 0m) continue;
|
||
decimal avgPrice = weightedPriceSum / totalSize;
|
||
|
||
// Parse timestamp to prevent old trades
|
||
DateTime tradeTs = DateTime.UtcNow;
|
||
if (firstAct.TryGetProperty("timestamp", out var tsProp))
|
||
{
|
||
if (tsProp.ValueKind == JsonValueKind.Number) // Unix
|
||
{
|
||
long rawTs = tsProp.GetInt64();
|
||
if (rawTs > 1000000000000)
|
||
tradeTs = DateTimeOffset.FromUnixTimeMilliseconds(rawTs).UtcDateTime;
|
||
else
|
||
tradeTs = DateTimeOffset.FromUnixTimeSeconds(rawTs).UtcDateTime;
|
||
}
|
||
else if (tsProp.ValueKind == JsonValueKind.String && DateTime.TryParse(tsProp.GetString(), null, System.Globalization.DateTimeStyles.AssumeUniversal | System.Globalization.DateTimeStyles.AdjustToUniversal, out var dt))
|
||
tradeTs = dt;
|
||
}
|
||
|
||
// If trade is older than 120 seconds or has an impossible future date (clock drift / timezone bug), skip
|
||
double ageSeconds = (DateTime.UtcNow - tradeTs).TotalSeconds;
|
||
if (ageSeconds > 120 || ageSeconds < -120)
|
||
{
|
||
if (ageSeconds < 86400)
|
||
{
|
||
if (_state.DebugPollingLog) _logger.Debug($"Activity skipped due to age ({ageSeconds}s / Date: {tradeTs:O}): {txHash}");
|
||
}
|
||
continue;
|
||
}
|
||
|
||
// CRITICAL: Check if the bare txHash was already processed by the Fast-Track parser.
|
||
// Fast-Track stores just "txHash", but here we use "txHash_asset_side".
|
||
// Without this check, a SELL processed by Fast-Track would be re-ingested as a BUY
|
||
// by the API (since the API sees both sides of the orderbook) — causing phantom purchases!
|
||
if (_processedTxHashes.ContainsKey(txHash))
|
||
continue; // Already handled by Fast-Track blockchain parser
|
||
|
||
// Only place the hash lock AFTER all filtering is successful (preventing dust fragments locking out main batches)!
|
||
if (!_processedTxHashes.TryAdd(uniqueTradeKey, DateTime.UtcNow))
|
||
continue; // Duplicate trade or redundant polling request
|
||
|
||
if (avgPrice <= 0.005m || totalSize <= 1.0m) continue; // Prevent absolute dust trades spanning
|
||
if (avgPrice > 0.99m) continue;
|
||
|
||
string displayQuestion = "Unknown Market";
|
||
if (firstAct.TryGetProperty("title", out var titleProp)) displayQuestion = titleProp.GetString() ?? "Unknown Market";
|
||
if (string.IsNullOrEmpty(displayQuestion) || displayQuestion == "Unknown Market")
|
||
{
|
||
// Fallback title evaluation
|
||
if (firstAct.TryGetProperty("marketQuestion", out var mqProp)) displayQuestion = mqProp.GetString() ?? "Unknown Market";
|
||
}
|
||
|
||
var signal = new CopySignal
|
||
{
|
||
TraderId = trader.Id,
|
||
TokenId = asset,
|
||
ConditionId = "",
|
||
MarketSlug = firstAct.TryGetProperty("slug", out var sp) ? sp.GetString() ?? "" : (firstAct.TryGetProperty("marketSlug", out var msp) ? msp.GetString() ?? "" : ""),
|
||
Side = parsedSide,
|
||
Price = avgPrice,
|
||
Size = totalSize,
|
||
Timestamp = tradeTs,
|
||
MarketQuestion = displayQuestion,
|
||
Outcome = firstAct.TryGetProperty("outcome", out var outProp) ? outProp.GetString() ?? "" : "",
|
||
Reason = parsedSide.ToUpper().Contains("SELL") ? "Master Trader Sold" : ""
|
||
};
|
||
|
||
// Parse endDate from activity JSON for market expiry
|
||
if (firstAct.TryGetProperty("endDate", out var endDateProp))
|
||
{
|
||
if (endDateProp.ValueKind == JsonValueKind.String && DateTime.TryParse(endDateProp.GetString(), null, System.Globalization.DateTimeStyles.AssumeUniversal | System.Globalization.DateTimeStyles.AdjustToUniversal, out var endDt))
|
||
signal.EndDate = endDt;
|
||
else if (endDateProp.ValueKind == JsonValueKind.Number)
|
||
signal.EndDate = DateTimeOffset.FromUnixTimeSeconds(endDateProp.GetInt64()).UtcDateTime;
|
||
}
|
||
else if (firstAct.TryGetProperty("end_date_iso", out var endIso) && endIso.ValueKind == JsonValueKind.String)
|
||
{
|
||
if (DateTime.TryParse(endIso.GetString(), null, System.Globalization.DateTimeStyles.AssumeUniversal | System.Globalization.DateTimeStyles.AdjustToUniversal, out var endDt2))
|
||
signal.EndDate = endDt2;
|
||
}
|
||
|
||
string shareType = string.IsNullOrEmpty(signal.Outcome) ? signal.Side : signal.Outcome;
|
||
bool wAdded = _signalWriter.TryWrite(signal);
|
||
if (wAdded)
|
||
{
|
||
_logger.Trade($"🚨 [QUELLE: {trader.DisplayName}] API\n" +
|
||
$" Markt: {signal.MarketQuestion}\n" +
|
||
$" Aktion: {signal.Side} {shareType} ({signal.Size:F2} Shares @ ${signal.Price:F3})\n" +
|
||
$" Zeit: {signal.Timestamp:HH:mm:ss} UTC");
|
||
}
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Warning($"Fehler beim Parsen einer Activity JSON (Merged): {ex.Message}");
|
||
}
|
||
}
|
||
}
|
||
}
|