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

Modularisierung: WSS-/Alchemy-Listener ins Copytrading-Modul verschoben

Browse Source 
Beide Listener sind copytrading-spezifisch (überwachen die getrackten Master-
Trader bzw. Auto-Redeem nach Copytrading-Settings) und werden jetzt vom Modul
besessen statt von der App. Reiner Move, keine Logikänderung.

- services/AlchemyWebsocketService.cs + services/PolymarketWssClient.cs
  -> src/PolyTrader.Modules.CopyTrading/Services/ (Namespace bleibt transitional).
- CopyTradingModule.RegisterServices registriert nun beide Hosted Services sowie
  IBlockchainWssClientFactory->AlchemyWssClientFactory (Core-WSS-Infra, nur vom
  Copytrading-Blockchain-Listener genutzt).
- Program.cs: die drei Registrierungen entfernt (WSS-Factory + 2 Hosted Services)
  + ungenutztes using PolyTrader.Core.Streaming.
- WSS-Verbindungsinfrastruktur (AlchemyWssClient/BlockchainWss) bleibt in Core.

DI-Auflösung unverändert (gleicher Container). Build grün, --smoke-ui grün.
Laufzeit-Verifikation der WSS-Verbindungen: Server-Test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Richard
2026-07-06 18:44:49 +02:00
co-authored by Claude Opus 4.8
parent 1b51872f00
commit 3456fd9768
4 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/AlchemyWebsocketService.cs
-202
View File
@@ -1,202 +0,0 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
using PolyTrader.Core.Streaming;
using PolyTraderSharp.Models;
namespace PolyTraderSharp.Services
{
public class AlchemyWebsocketService : BackgroundService
{
private const string CtfContractAddress = "0x4D97DCd97eC945f40cF65F87097ACe5EA0476045";
private const string TransferSingleTopic = "0xc3d58168c5ae7397731d063d5bbf3d657854427343f4c083240f7aacaa2d0f62";
private const string TransferBatchTopic = "0x4a39dc06d4c0dbc64b70af90fd698a233a518aa5d07e595d983b8c0526c8f7ce";
private readonly TradingState _state;
private readonly CopyTradingState _copyState;
private readonly ServerSettings _settings;
private readonly TraderMonitorService _traderMonitor;
private readonly TerminalLogger _logger;
private readonly IBlockchainWssClientFactory _wssFactory;
public AlchemyWebsocketService(
TradingState state,
CopyTradingState copyState,
ServerSettings settings,
TraderMonitorService traderMonitor,
TerminalLogger logger,
IBlockchainWssClientFactory wssFactory)
{
_state = state;
_copyState = copyState;
_settings = settings;
_traderMonitor = traderMonitor;
_logger = logger;
_wssFactory = wssFactory;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
if (!_settings.EnableBlockchainListener || string.IsNullOrEmpty(_settings.PolygonRpcUrl))
{
_logger.Info("Blockchain Listener is disabled in settings. Using raw polling.");
_state.IsAlchemyHealthy = false;
return;
}
_logger.Info("Alchemy WSS Service starting up...");
while (!stoppingToken.IsCancellationRequested)
{
if (_state.GlobalTradingPaused ||
(_state.LiveTradingMode == TradingMode.Inactive && _state.DemoTradingMode == TradingMode.Inactive))
{
_state.IsAlchemyHealthy = false;
await Task.Delay(5000, stoppingToken);
continue;
}
try
{
await ConnectAndListenAsync(stoppingToken);
}
catch (WebSocketException ex)
{
// Usually indicates a connection drop or 429
_logger.Warning($"Alchemy WSS drop: {ex.Message}. Falling back to API polling for 5 minutes.");
_state.IsAlchemyHealthy = false;
await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken);
}
catch (Exception ex)
{
if (ex.Message.Contains("429") || ex.Message.Contains("Too Many Requests"))
{
_logger.Error($"Alchemy HTTP 429 Limit reached. Suspending WSS for 5 minutes.");
_state.IsAlchemyHealthy = false;
await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken);
}
else
{
_logger.Error($"Alchemy WSS Error: {ex.Message}. Retrying in 10s...");
_state.IsAlchemyHealthy = false;
await Task.Delay(10000, stoppingToken);
}
}
}
}
/// <summary>
/// Baut EINE WSS-Session über den Core-Client auf. Consumer-Verantwortung: Filter aus den
/// getrackten Master-Tradern bauen, auf Trader-Listen-Änderungen re-subscriben, Health spiegeln.
/// </summary>
private async Task ConnectAndListenAsync(CancellationToken stoppingToken)
{
var activeTraders = _copyState.Traders.Values.Where(t => t.IsActive).ToList();
var activeStateHash = string.Join(",", activeTraders.OrderBy(t => t.Id).Select(t => t.WalletAddress.ToLowerInvariant()));
var subscription = BuildSubscription(activeTraders);
if (subscription.Filters.Count == 0)
{
_logger.Info("Keine aktiven Master-Trader. Socket läuft im Standby...");
_state.IsAlchemyHealthy = false;
await Task.Delay(5000, stoppingToken);
return;
}
var client = _wssFactory.Create();
_logger.Info("Connecting to Alchemy WebSocket (Core WSS client)...");
using var loopCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
// Überwacht Trader-Listen-Änderungen (Re-Subscribe) und spiegelt das Health-Flag.
var monitorTask = Task.Run(async () =>
{
while (!loopCts.IsCancellationRequested)
{
try { await Task.Delay(5000, loopCts.Token); }
catch (OperationCanceledException) { break; }
_state.IsAlchemyHealthy = client.IsHealthy;
var currentTraders = _copyState.Traders.Values.Where(t => t.IsActive).ToList();
var currentHash = string.Join(",", currentTraders.OrderBy(t => t.Id).Select(t => t.WalletAddress.ToLowerInvariant()));
if (currentHash != activeStateHash)
{
_logger.Info("🔄 Master-Trader Liste hat sich geändert. Starte Alchemy WSS mit neuen Filtern neu...");
loopCts.Cancel();
break;
}
}
});
try
{
await client.ConnectAndListenAsync(_settings.PolygonRpcUrl, subscription, OnLog, loopCts.Token);
}
catch (OperationCanceledException)
{
// Erwartet beim Re-Subscribe (Trader-Liste geändert) oder beim Shutdown.
}
finally
{
_state.IsAlchemyHealthy = false;
if (!loopCts.IsCancellationRequested) loopCts.Cancel();
}
}
private BlockchainWssSubscription BuildSubscription(List<TrackedTrader> activeTraders)
{
var subscription = new BlockchainWssSubscription();
var paddedAddresses = activeTraders.Select(t => PadAddress(t.WalletAddress)).ToList();
var topic0 = new List<string> { TransferSingleTopic, TransferBatchTopic };
const int batchSize = 3; // Alchemy begrenzt Topic-Arrays auf max. 3-4 Einträge
for (int i = 0; i < paddedAddresses.Count; i += batchSize)
{
var chunk = paddedAddresses.Skip(i).Take(batchSize).ToList();
// Buys: Master-Trader ist Empfänger (Topic 3)
subscription.Filters.Add(new LogSubscriptionFilter { Address = CtfContractAddress, Topic0 = topic0, Topic3 = chunk });
// Sells: Master-Trader ist Sender (Topic 2)
subscription.Filters.Add(new LogSubscriptionFilter { Address = CtfContractAddress, Topic0 = topic0, Topic2 = chunk });
}
return subscription;
}
private void OnLog(BlockchainLogEvent evt)
{
if (evt.Topics.Count < 4) return;
var fromTopic = evt.Topics[2]?.ToLowerInvariant();
var toTopic = evt.Topics[3]?.ToLowerInvariant();
if (fromTopic == null || toTopic == null) return;
var activeTraders = _copyState.Traders.Values.Where(t => t.IsActive).ToList();
string? triggeredAddress = null;
foreach (var trader in activeTraders)
{
var padded = PadAddress(trader.WalletAddress);
if (fromTopic == padded || toTopic == padded)
{
triggeredAddress = trader.WalletAddress;
break;
}
}
if (!string.IsNullOrEmpty(triggeredAddress))
{
_traderMonitor.TriggerFastBlockchainPoll(evt.TransactionHash, _settings.PolygonRpcUrl, triggeredAddress);
}
}
private string PadAddress(string address)
{
string stripped = address.Replace("0x", "", StringComparison.OrdinalIgnoreCase).ToLowerInvariant();
return "0x" + stripped.PadLeft(64, '0');
}
}
}
services/PolymarketWssClient.cs
-301
View File
@@ -1,301 +0,0 @@
using System;
using PolyTrader.Core.Persistence;
using PolyTrader.Modules.CopyTrading.Persistence;
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 CopyTradingState _copyState;
private readonly ServerSettings _settings;
private readonly PolymarketClobClient _clob;
private readonly TerminalLogger _logger;
private readonly ICopyTradeLogRepository _tradeLog;
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,
CopyTradingState copyState,
ServerSettings settings,
PolymarketClobClient clob,
TerminalLogger logger,
ICopyTradeLogRepository tradeLog,
IPositionRepository positionRepo,
IAccountRepository accountRepo)
{
_state = state;
_copyState = copyState;
_settings = settings;
_clob = clob;
_logger = logger;
_tradeLog = tradeLog;
_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 (_copyState.GetAccountSettings(acc.AccountId).PreRedeemLimit > 0 && price >= _copyState.GetAccountSettings(acc.AccountId).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 >= {_copyState.GetAccountSettings(acc.AccountId).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 >= {_copyState.GetAccountSettings(acc.AccountId).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 = _copyState.GetAccountSettings(acc.AccountId).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 = _copyState.GetAccountSettings(acc.AccountId).PreRedeemLimit;
decimal exitUsd = pos.Size * exactLimitPrice;
decimal realizedPnl = exitUsd - pos.AmountUsd;
_state.GlobalPnl += realizedPnl;
acc.UpdateBalance(acc.AvailableBalance + exitUsd);
var ct = new ClosedTrade
{
TradeId = _copyState.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"
};
_tradeLog.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}");
}
}
}
}