using System; using MongoDB.Driver; using PolyTrader.Core.Persistence; using PolyTraderSharp.Extensions; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Net.WebSockets; using System.Text; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Hosting; using PolyTraderSharp.Models; namespace PolyTraderSharp.Services { public class PolymarketWssClient : BackgroundService { private const string MarketWssUrl = "wss://ws-subscriptions-clob.polymarket.com/ws/market"; private readonly TradingState _state; private readonly ServerSettings _settings; private readonly PolymarketClobClient _clob; private readonly TerminalLogger _logger; private readonly IMongoDatabase _db; private readonly IPositionRepository _positionRepo; private readonly IAccountRepository _accountRepo; // Tracking rate limits for auto redeem: max 2 attempts per position, 5 min apart private readonly ConcurrentDictionary _redeemAttempts = new(); public PolymarketWssClient( TradingState state, ServerSettings settings, PolymarketClobClient clob, TerminalLogger logger, IMongoDatabase db, IPositionRepository positionRepo, IAccountRepository accountRepo) { _state = state; _settings = settings; _clob = clob; _logger = logger; _db = db; _positionRepo = positionRepo; _accountRepo = accountRepo; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { while (!stoppingToken.IsCancellationRequested) { if (!_settings.UsePolymarketWebsockets || _state.GlobalTradingPaused) { await Task.Delay(5000, stoppingToken); continue; } try { await ConnectMarketWssAsync(stoppingToken); } catch (Exception ex) { _logger.Warning($"Polymarket WSS disconnected ({ex.Message}). Retrying in 5s..."); await Task.Delay(5000, stoppingToken); } } } private async Task ConnectMarketWssAsync(CancellationToken stoppingToken) { using var ws = new ClientWebSocket(); _logger.Info("Connecting to Polymarket WSS (Market Stream) for live pricing..."); await ws.ConnectAsync(new Uri(MarketWssUrl), stoppingToken); _logger.Info("βœ… Polymarket Market WSS Connected."); var allSubscriptions = new HashSet(); var subscriptionTask = Task.Run(async () => { while (ws.State == WebSocketState.Open && !stoppingToken.IsCancellationRequested && _settings.UsePolymarketWebsockets) { var neededAssets = new HashSet(); foreach (var acc in _state.Accounts.Values.Where(a => a.IsActive)) foreach (var token in acc.OpenPositions.Keys) neededAssets.Add(token); var missing = neededAssets.Except(allSubscriptions).ToList(); if (missing.Any()) { var req = new { assets_ids = missing, type = "market" }; var json = System.Text.Json.JsonSerializer.Serialize(req); var bytes = Encoding.UTF8.GetBytes(json); await ws.SendAsync(new ArraySegment(bytes), WebSocketMessageType.Text, true, stoppingToken); foreach (var m in missing) allSubscriptions.Add(m); _logger.Info($"πŸ“‘ Polymarket WSS: Subscribed to {missing.Count} new assets. Total: {allSubscriptions.Count}"); } await Task.Delay(5000, stoppingToken); // Check for new positions every 5s } }, stoppingToken); var buffer = new byte[1024 * 64]; // 64kb buffer while (ws.State == WebSocketState.Open && !stoppingToken.IsCancellationRequested && _settings.UsePolymarketWebsockets) { var result = await ws.ReceiveAsync(new ArraySegment(buffer), stoppingToken); if (result.MessageType == WebSocketMessageType.Close) break; var message = Encoding.UTF8.GetString(buffer, 0, result.Count); if (!string.IsNullOrEmpty(message)) { try { ProcessMarketMessage(message); } catch { } } } } private void ProcessMarketMessage(string jsonStr) { try { using var doc = JsonDocument.Parse(jsonStr); var root = doc.RootElement; if (!root.TryGetProperty("event_type", out var evtTypeProp)) return; var eventType = evtTypeProp.GetString(); if (eventType == "price_change") { if (root.TryGetProperty("price_changes", out var changes) && changes.ValueKind == JsonValueKind.Array) { foreach (var change in changes.EnumerateArray()) { if (change.TryGetProperty("asset_id", out var assetIdProp) && change.TryGetProperty("price", out var priceProp)) { string assetId = assetIdProp.GetString()!; decimal.TryParse(priceProp.GetString(), System.Globalization.NumberStyles.Any, System.Globalization.CultureInfo.InvariantCulture, out decimal price); decimal bestBid = price; if (change.TryGetProperty("best_bid", out var bidProp) && decimal.TryParse(bidProp.GetString(), System.Globalization.NumberStyles.Any, System.Globalization.CultureInfo.InvariantCulture, out decimal bBid)) { if (bBid > 0) bestBid = bBid; } UpdateAssetPriceAndCheckAutoRedeem(assetId, bestBid); } } } } else if (eventType == "last_trade_price") { if (root.TryGetProperty("asset_id", out var assetIdProp) && root.TryGetProperty("price", out var priceProp)) { string assetId = assetIdProp.GetString()!; decimal.TryParse(priceProp.GetString(), System.Globalization.NumberStyles.Any, System.Globalization.CultureInfo.InvariantCulture, out decimal price); UpdateAssetPriceAndCheckAutoRedeem(assetId, price); } } } catch { } } private void UpdateAssetPriceAndCheckAutoRedeem(string assetId, decimal price) { if (price <= 0 || string.IsNullOrEmpty(assetId)) return; foreach (var acc in _state.Accounts.Values) { if (acc.OpenPositions.TryGetValue(assetId, out var pos)) { pos.CurrentPrice = price; pos.CurrentValueUsd = pos.Size * price; // Execute Auto-Redeem if config conditions are met if (acc.PreRedeemLimit > 0 && price >= acc.PreRedeemLimit && acc.IsActive) { string redeemKey = $"{acc.AccountId}_{assetId}"; // Spam protection: max 2 attempts per position, 5 minutes apart if (_redeemAttempts.TryGetValue(redeemKey, out var redeemState)) { if (redeemState.Count >= 2) continue; // Permanently ignore after 2 failed attempts if ((DateTime.UtcNow - redeemState.LastAttempt).TotalMinutes < 5) continue; // Wait 5 min between attempts } if (!acc.IsDemo && _state.LiveTradingMode == TradingMode.Active) { _logger.Trade($"🚨 [AUTO REDEEM] {acc.Name} | {pos.MarketQuestion} | Preis >= {acc.PreRedeemLimit}"); // Best effort non-blocking _ = Task.Run(async () => await ExecuteAutoRedeemLive(acc, pos, price)); } else if (acc.IsDemo && _state.DemoTradingMode == TradingMode.Active) { _logger.Trade($"🚨 [AUTO REDEEM DEMO] {acc.Name} | {pos.MarketQuestion} | Preis >= {acc.PreRedeemLimit}"); _ = Task.Run(() => ExecuteAutoRedeemDemo(acc, pos, price)); } } } } } private async Task ExecuteAutoRedeemLive(AccountState acc, Position pos, decimal triggerPrice) { string redeemKey = $"{acc.AccountId}_{pos.TokenId}"; if (pos.Size < 5.0m) { var state = _redeemAttempts.GetOrAdd(redeemKey, _ => (0, DateTime.MinValue)); int newCount = state.Count + 1; _redeemAttempts[redeemKey] = (newCount, DateTime.UtcNow); if (newCount <= 1) // Only log once _logger.Warning($"[AUTO REDEEM] Position {pos.MarketQuestion} zu klein fΓΌr Limit Order (< 5 Shares). Max. 1 Retry in 5 Min."); return; } // Track successful attempt _redeemAttempts.AddOrUpdate(redeemKey, _ => (1, DateTime.UtcNow), (_, old) => (old.Count + 1, DateTime.UtcNow)); try { // The user explicitly requested an exact GTC order using the configured PreRedeemLimit, without slippage decimal expectedFillPrice = acc.PreRedeemLimit; decimal amountUsdc = Math.Max(pos.Size * expectedFillPrice, 0.01m); // Fire and forget SELL via ClobClient var result = await _clob.PlaceOrderAsync(acc, pos.TokenId, "SELL", amountUsdc, expectedFillPrice, "GTC", false, false); if (result == "OK") { _logger.Info($"βœ… Auto-Redeem Sell sent for {acc.Name} at exact Limit {expectedFillPrice:F3} USD (GTC)."); // Assume it's an open matching order. Clob/Market API will sync actual status later. // DO NOT remove from OpenPositions here. Wait for Live Sync to detect the closure // via the API so it can properly fetch the Realized PnL and save the ClosedTrade record! } else { _logger.Error($"❌ Auto-Redeem failed or rejected: {result}."); } } catch (Exception ex) { _logger.Error($"Auto Redeem Exception: {ex.Message}"); } } private void ExecuteAutoRedeemDemo(AccountState acc, Position pos, decimal triggerPrice) { try { if (acc.OpenPositions.TryRemove(pos.TokenId, out _)) { _positionRepo.DeleteDemo(acc.AccountId, pos.TokenId); decimal exactLimitPrice = acc.PreRedeemLimit; decimal exitUsd = pos.Size * exactLimitPrice; decimal realizedPnl = exitUsd - pos.AmountUsd; _state.GlobalPnl += realizedPnl; acc.UpdateBalance(acc.AvailableBalance + exitUsd); var ct = new ClosedTrade { TradeId = _state.GetNextTradeId(), AccountId = acc.AccountId, IsDemo = true, MarketSlug = pos.MarketSlug, MarketQuestion = pos.MarketQuestion, TokenId = pos.TokenId, Outcome = pos.Outcome, Side = "SELL", EntryPrice = pos.EntryPrice, ExitPrice = exactLimitPrice, Size = pos.Size, RealizedPnl = realizedPnl, PnlPercent = pos.AmountUsd > 0 ? (realizedPnl / pos.AmountUsd * 100m) : 0m, OpenedAt = pos.OpenedAt, ClosedAt = DateTime.UtcNow, ExitReason = "Pre Redeem" }; _db.GetCollection("closed_trades").Insert(ct); _accountRepo.Upsert(acc); _logger.Trade($"βœ… [AUTO REDEEM DEMO ERFOLGREICH] {pos.MarketQuestion} | Exit: {pos.Size:F2} @ {exactLimitPrice:F3} | PnL: ${realizedPnl:F2}"); } } catch (Exception ex) { _logger.Error($"Demo Auto Redeem failed: {ex.Message}"); } } } }