PolyTraderSharp/services/PolymarketWssClient.cs

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<string, (int Count, DateTime LastAttempt)> _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<string>();
            var subscriptionTask = Task.Run(async () =>
            {
                while (ws.State == WebSocketState.Open && !stoppingToken.IsCancellationRequested && _settings.UsePolymarketWebsockets)
                {
                    var neededAssets = new HashSet<string>();
                    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<byte>(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<byte>(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<ClosedTrade>("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}");
            }
        }
    }
}