using System.Collections.Concurrent; using System.Globalization; using IBApi; using IBKRTrader.Core.Logging; using IBKRTrader.Core.Time; namespace IBKRTrader.Core.Trading.Ibkr; /// /// Hält die Socket-Verbindung zur TWS bzw. zum IB Gateway und übersetzt die callback-basierte /// TWS-API in awaitable Anfragen: jede Anfrage bekommt eine reqId, deren Antworten in einem /// Slot gesammelt und über einen aufgelöst werden. /// /// Verbunden wird träge bei der ersten Anfrage und danach bei Bedarf erneut: TWS ist nach einem /// Neustart mehrere Minuten nicht bereit (erst abgelehnte Verbindungen, dann abgebrochene /// Handshakes), ein einmaliger Verbindungsversuch beim Programmstart würde das nicht überleben. /// internal sealed class IbkrConnection : DefaultEWrapper, IDisposable { private const string LogModule = "IBKR"; private readonly LoggingService _logger; private readonly string _host; private readonly int _port; private readonly int _clientId; private readonly int _marketDataType; private readonly TimeSpan _connectTimeout; private readonly EReaderMonitorSignal _signal = new(); private readonly EClientSocket _socket; private readonly SemaphoreSlim _connectLock = new(1, 1); private readonly ConcurrentDictionary _quotes = new(); private readonly ConcurrentDictionary _accounts = new(); private readonly ConcurrentDictionary _contracts = new(); private readonly ConcurrentDictionary _chains = new(); private readonly ConcurrentDictionary _greeks = new(); private readonly ConcurrentDictionary _orders = new(); private readonly ConcurrentDictionary _executions = new(); private readonly ConcurrentDictionary _contractCache = new(StringComparer.OrdinalIgnoreCase); // Portfolio-Abruf ist ein kontoweites Abonnement, keine reqId-Anfrage – daher nur ein Slot. private PortfolioSlot? _portfolio; // Dasselbe gilt fuer das Orderbuch: reqAllOpenOrders traegt keine reqId. private OpenOrderSlot? _openOrders; private TaskCompletionSource _handshake = NewTcs(); private TaskCompletionSource? _serverTime; /// Ab diesem Versatz zur TWS-Serverzeit gilt die Uhr des Hosts als verstellt. private const double MaxClockSkewSeconds = 5; // Der Hinweis auf angenommene Zeitzonen soll einmal je Verbindung kommen, nicht je Abruf. private bool _zoneWarningIssued; private volatile bool _ready; private volatile bool _disposed; private int _nextRequestId = 1000; private int _nextOrderId = -1; private string? _account; public IbkrConnection(LoggingService logger, string host, int port, int clientId, int marketDataType, TimeSpan connectTimeout) { _logger = logger; _host = host; _port = port; _clientId = clientId; _marketDataType = marketDataType; _connectTimeout = connectTimeout; _socket = new EClientSocket(this, _signal); } /// Kontonummer, die TWS beim Verbinden gemeldet hat (z. B. "DUR371528"). public string? Account => _account; // ─── Verbindung ─────────────────────────────────────────────────────────── public async Task EnsureConnectedAsync(CancellationToken ct) { if (_disposed) return false; if (_ready && _socket.IsConnected()) return true; await _connectLock.WaitAsync(ct).ConfigureAwait(false); try { if (_disposed) return false; if (_ready && _socket.IsConnected()) return true; return await ConnectAsync(ct).ConfigureAwait(false); } finally { _connectLock.Release(); } } private async Task ConnectAsync(CancellationToken ct) { _ready = false; _handshake = NewTcs(); SafeDisconnect(); // eConnect blockiert und hängt nach einem TWS-Neustart auch schon mal minutenlang ohne // Antwort – deshalb auf einem Hintergrund-Thread mit Zeitlimit statt direkt. var connect = Task.Run(() => { try { _socket.eConnect(_host, _port, _clientId); return _socket.IsConnected(); } catch { // Die API meldet Socket-Fehler bereits über error(Exception); hier reicht das Ergebnis. return false; } }, ct); if (!await WaitAsync(connect, _connectTimeout, ct).ConfigureAwait(false) || !connect.Result) { _logger.Warn(LogModule, $"Keine Verbindung zu {_host}:{_port} (Client {_clientId}) – läuft TWS/IB Gateway?"); SafeDisconnect(); return false; } StartReader(); // Erst nextValidId bestätigt den Handshake; vorher sind keine Anfragen zulässig. Bleibt es // aus, hat TWS die Verbindung nicht freigegeben (fehlende Trusted IP oder offenes Popup). if (!await WaitAsync(_handshake.Task, _connectTimeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, "TWS hat den Handshake nicht bestätigt (kein nextValidId). Trusted IP 127.0.0.1 prüfen " + "und TWS neu starten – siehe docs/TWS-Setup-Checkliste.md."); SafeDisconnect(); return false; } _ready = true; _zoneWarningIssued = false; _socket.reqMarketDataType(_marketDataType); _logger.Info(LogModule, $"Verbunden mit {_host}:{_port} (Client {_clientId}), Konto {_account ?? "unbekannt"}."); await LogTimeContextAsync(ct).ConfigureAwait(false); return true; } /// /// Schreibt den vollständigen Zeitkontext einer Verbindung ins Log: Betriebszeitzone, /// Systemzeitzone und den Versatz zur Uhr des TWS-Servers. /// /// Wozu: Wir betreiben Instanzen in EU und US, künftig auf Linux-VMs. Weicht die /// Betriebszeitzone von der des Hosts ab oder geht die VM-Uhr nach, verschieben sich /// Buchungszeiten – ohne dass irgendwo ein Fehler auftaucht. Steht der Kontext am Anfang jeder /// Verbindung im Log, lässt sich das im Nachhinein an einer Zeile ablesen statt zu raten. /// private async Task LogTimeContextAsync(CancellationToken ct) { var context = $"Zeitkontext: Betriebszeitzone {AppTimeZone.CurrentId}, " + $"Systemzeitzone {TimeZoneInfo.Local.Id}"; var pending = NewTcs(); _serverTime = pending; try { _socket.reqCurrentTime(); if (!await WaitAsync(pending.Task, TimeSpan.FromSeconds(5), ct).ConfigureAwait(false)) { _logger.Info(LogModule, context + ", TWS-Serverzeit nicht ermittelbar."); return; } var serverUtc = DateTimeOffset.FromUnixTimeSeconds(pending.Task.Result).UtcDateTime; var skew = (serverUtc - DateTime.UtcNow).TotalSeconds; context += $", TWS-Serverzeit {serverUtc:yyyy-MM-dd HH:mm:ss}Z, " + $"Uhrenversatz {skew.ToString("+0.0;-0.0;0", CultureInfo.InvariantCulture)} s"; if (Math.Abs(skew) > MaxClockSkewSeconds) _logger.Warn(LogModule, context + " – die Uhren laufen auseinander. In virtuellen Maschinen ist das ein häufiger " + "Fehler; die Zeitsynchronisation des Hosts prüfen, sonst wandern Buchungszeiten."); else _logger.Info(LogModule, context + "."); } finally { _serverTime = null; } } /// /// Fasst nach jedem Abruf zusammen, wie viele Zeitstempel TWS mit Zonenangabe gemeldet hat und /// wie viele über die Betriebszeitzone angenommen wurden. Die angenommenen sind die /// Stelle, an der ein Wechsel zwischen EU- und US-Host lautlos danebenliegt. /// private void LogTimeProvenance(IReadOnlyList stamps) { if (stamps.Count == 0) return; var reported = stamps.Count(t => t.Source == IbkrMapping.ExecutionTimeSource.ReportedZone); var assumed = stamps.Count(t => t.IsAssumed); var unreadable = stamps.Count(t => t.Source == IbkrMapping.ExecutionTimeSource.Unparsable); var zones = string.Join(", ", stamps.Where(t => t.ReportedZone is not null) .Select(t => t.ReportedZone!) .Distinct()); var text = $"Zeitstempel von {stamps.Count} Ausführung(en): {reported} mit gemeldeter Zone" + (zones.Length > 0 ? $" ({zones})" : "") + $", {assumed} über die Betriebszeitzone {AppTimeZone.CurrentId}" + (unreadable > 0 ? $", {unreadable} unlesbar" : "") + "."; _logger.Info(LogModule, text); // Unlesbare Zeitstempel sind immer ein Defekt – die Ausführung landet sonst auf DateTime.MinValue. if (unreadable > 0) _logger.Warn(LogModule, $"{unreadable} Ausführung(en) mit unlesbarem Zeitstempel – das Format von TWS hat sich " + "vermutlich geändert. IbkrMapping.ResolveExecutionTime prüfen."); // Angenommene Zonen sind nur dann heikel, wenn Betriebs- und Systemzeitzone auseinandergehen: // dann ist nicht mehr offensichtlich, gegen welche Uhr TWS die Zeit gemeldet hat. if (assumed > 0 && !_zoneWarningIssued && !string.Equals(AppTimeZone.CurrentId, TimeZoneInfo.Local.Id, StringComparison.OrdinalIgnoreCase)) { _zoneWarningIssued = true; _logger.Warn(LogModule, $"{assumed} Zeitstempel ohne Zonenangabe wurden gegen die Betriebszeitzone " + $"{AppTimeZone.CurrentId} gerechnet, das System läuft aber auf {TimeZoneInfo.Local.Id}. " + "Stimmt Trading.ApplicationTimeZoneId nicht mit der Zeitzone des TWS-Hosts überein, " + "liegen die Buchungszeiten daneben. Einmal gegen TWS gegenprüfen."); } } private void StartReader() { var reader = new EReader(_socket, _signal); reader.Start(); new Thread(() => { while (_socket.IsConnected()) { _signal.waitForSignal(); try { reader.processMsgs(); } catch (Exception ex) { if (!_disposed) _logger.Warn(LogModule, $"Nachrichten-Reader beendet: {ex.Message}"); break; } } }) { IsBackground = true, Name = "IBKR-Reader" }.Start(); } private void SafeDisconnect() { try { _socket.eDisconnect(); } catch { /* Socket war nie offen oder ist bereits zu. */ } } // ─── Anfragen ───────────────────────────────────────────────────────────── /// /// Löst ein Symbol in einen vollständigen Kontrakt auf (inklusive ConId, damit Kurse und /// Orders dasselbe Instrument treffen). Ergebnisse werden für die Laufzeit zwischengespeichert. /// /// /// Löst ein Instrument gegen TWS in einen vollständigen Kontrakt auf – Aktie wie Option. /// /// Der Zwischenspeicher liegt auf dem Instrumentenschlüssel, nicht auf dem Ticker: sonst /// überschrieben sich alle Optionen desselben Basiswerts gegenseitig und ein Auftrag auf den /// 150er-Call träfe den zuletzt aufgelösten Kontrakt. /// public async Task ResolveContractAsync(InstrumentRef instrument, TimeSpan timeout, CancellationToken ct) { var key = instrument.Key; if (_contractCache.TryGetValue(key, out var cached)) return cached; var id = NextRequestId(); var slot = new ContractSlot(); _contracts[id] = slot; try { var query = instrument.Option is { } option ? IbkrMapping.Option(option) : IbkrMapping.Stock(instrument.Symbol, instrument.Currency); _socket.reqContractDetails(id, query); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, $"Zeitüberschreitung bei der Kontraktsuche für {key}."); return null; } if (slot.Error is not null) { _logger.Warn(LogModule, $"Kontrakt {key} nicht auflösbar: {slot.Error}"); return null; } if (slot.First is null) { _logger.Warn(LogModule, $"Kein Kontrakt für {key} gefunden."); return null; } _contractCache[key] = slot.First; return slot.First; } finally { _contracts.TryRemove(id, out _); } } /// /// Momentaufnahme des Kurses. Eine Zeitüberschreitung ist hier kein Fehler: TWS beendet den /// Snapshot nicht immer sauber, die bis dahin gelieferten Ticks reichen meist für einen Kurs. /// public async Task RequestQuoteAsync(Contract contract, string symbol, TimeSpan timeout, CancellationToken ct) { var id = NextRequestId(); var slot = new QuoteSlot(); _quotes[id] = slot; try { _socket.reqMktData(id, contract, "", true, false, null); await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false); if (slot.Error is not null) _logger.Warn(LogModule, $"Kursabfrage {symbol}: {slot.Error}"); return IbkrMapping.BuildQuote(symbol, slot.Last, slot.Bid, slot.Ask, slot.Close); } finally { _quotes.TryRemove(id, out _); } } /// /// Verfallstage und Strikes eines Basiswerts (reqSecDefOptParams). /// /// Die Abfrage braucht die Kontrakt-ID des Basiswerts, nicht nur sein Symbol – deshalb /// wird der Basiswert zuerst aufgelöst. Ohne die ID liefert TWS die Ketten aller Werte mit /// diesem Symbol an allen Börsen, und was davon zusammengehört, wäre nicht mehr zu erkennen. /// /// TWS antwortet je Börse und Handelsklasse mit einem eigenen Satz. Wir nehmen die Sätze /// von SMART, wenn es sie gibt: das ist die Serie, gegen die auch geordert wird. Die Strikes /// der übrigen Börsen zu mischen würde Ausübungspreise anbieten, die in der gehandelten Serie /// nicht existieren. /// public async Task RequestOptionChainAsync( InstrumentRef underlying, TimeSpan timeout, CancellationToken ct) { var stock = await ResolveContractAsync(underlying, timeout, ct).ConfigureAwait(false); if (stock is null) { _logger.Warn(LogModule, $"Optionskette {underlying.Key}: Basiswert nicht auflösbar."); return null; } var id = NextRequestId(); var slot = new OptionChainSlot(); _chains[id] = slot; try { _socket.reqSecDefOptParams(id, stock.Symbol, "", stock.SecType, stock.ConId); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, $"Optionskette {underlying.Key}: Zeitüberschreitung."); return null; } if (slot.Error is not null) { _logger.Warn(LogModule, $"Optionskette {underlying.Key}: {slot.Error}"); return null; } var series = BestSeries(slot, underlying.Symbol); if (series is null) { _logger.Info(LogModule, $"Optionskette {underlying.Key}: keine Serien gemeldet."); return null; } var chain = new OptionChain( underlying.Symbol, underlying.Currency, series.Exchange, series.TradingClass, series.Multiplier, [.. series.Expiries], [.. series.Strikes]); _logger.Info(LogModule, $"Optionskette: {chain}"); return chain; } finally { // Kein Abbestellen noetig: reqSecDefOptParams ist eine einmalige Antwort, kein Abo. _chains.TryRemove(id, out _); } } /// /// Wählt die Serie, gegen die tatsächlich geordert wird. /// /// TWS meldet für einen Basiswert mehrere Sätze - für AAPL etwa die Standardserie /// "AAPL" und Sonderserien wie "2AAPL", die aus Splits oder Sonderdividenden stammen und /// andere Ausübungspreise führen. Nimmt man den erstbesten, entstehen Kontrakte, die es nicht /// gibt: Am 2026-09-05 lieferte die Auswahl der Serie "2AAPL" für Strike 320 die TWS-Meldung /// 200 ("No security definition has been found"). /// /// Regel: SMART vor allen anderen Börsen, und innerhalb von SMART die Handelsklasse, die /// dem Symbol entspricht - das ist die Standardserie. Bleibt danach mehr als eine übrig, /// gewinnt die mit den meisten Ausübungspreisen; eine Sonderserie ist stets die dünnere. /// private static OptionSeries? BestSeries(OptionChainSlot slot, string symbol) { lock (slot.Series) { if (slot.Series.Count == 0) return null; return slot.Series .OrderByDescending(x => x.Exchange.Equals("SMART", StringComparison.OrdinalIgnoreCase)) .ThenByDescending(x => x.TradingClass.Equals(symbol, StringComparison.OrdinalIgnoreCase)) .ThenByDescending(x => x.Strikes.Count) .First(); } } /// /// Bewertungsgrößen eines Kontrakts (tickOptionComputation, Modell-Feld). /// /// Als Momentaufnahme angefordert. Kommt bis zum Ablauf der Frist nichts, ist das kein /// Fehler, sondern die übliche Antwort für eine Serie, die gerade niemand stellt – der /// Aufrufer bekommt null und entscheidet selbst. /// public async Task RequestGreeksAsync( OptionSpec spec, TimeSpan timeout, CancellationToken ct) { var contract = IbkrMapping.Option(spec); var id = NextRequestId(); var slot = new GreeksSlot(); _greeks[id] = slot; try { // Kein Snapshot, sondern ein Abo: TWS berechnet die Bewertungsgrößen nur im laufenden // Datenstrom und liefert sie in einem Snapshot überhaupt nicht (am 2026-09-05 gegen // das Paper-Gateway gemessen - die Kette kam, die Greeks blieben leer). Das Abo endet // beim ersten Modell-Tick, das finally bestellt es ab. _socket.reqMktData(id, contract, "", false, false, null); await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false); if (slot.Error is not null) _logger.Warn(LogModule, $"Bewertung {spec}: {slot.Error}"); return slot.Value; } finally { _greeks.TryRemove(id, out _); try { _socket.cancelMktData(id); } catch { /* siehe oben */ } } } public async Task RequestAccountAsync(TimeSpan timeout, CancellationToken ct) { var id = NextRequestId(); var slot = new AccountSlot(); _accounts[id] = slot; try { // $LEDGER:ALL liefert die Bargeldbestaende JE WAEHRUNG. Sie werden fuer die // Waehrungsdeckung gebraucht (Kapitalmodell §4 Schicht C): ein Kauf in X darf nur aus // Cash in X finanziert werden, sonst entsteht ein Sollsaldo in X - ein Margin-Kredit, // auch wenn das Konto in Summe positiv aussieht. // // ":ALL" ist noetig und nicht kosmetisch. Das nackte "$LEDGER" liefert nur die // Zusammenfassung in Kontobasiswaehrung, und die traegt als Waehrungskennung "BASE" - // also genau den Wert, den die Auswertung ueberspringen muss, damit dasselbe Geld nicht // zweimal zaehlt. Ergebnis war eine leere Aufschluesselung und damit eine Ablehnung // jedes Kaufs durch Schicht C (am 2026-09-04 gegen das Paper-Gateway beobachtet). _socket.reqAccountSummary(id, "All", "NetLiquidation,AvailableFunds,TotalCashValue,$LEDGER:ALL"); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, "Zeitüberschreitung bei der Kontoabfrage."); return null; } if (slot.Error is not null) { _logger.Warn(LogModule, $"Kontoabfrage fehlgeschlagen: {slot.Error}"); return null; } // Was der Broker an Waehrungen gemeldet hat, gehoert ins Log: ohne diese Zeile ist eine // Ablehnung durch Schicht C ("verfuegbar 0") von aussen nicht von einem echten // Geldmangel zu unterscheiden. _logger.Info(LogModule, slot.Cash.Count == 0 ? "Konto: keine Bargeldaufschluesselung je Waehrung erhalten - Schicht C lehnt Kaeufe " + "deshalb ab. Antwortet das Gateway nicht auf $LEDGER:ALL?" : "Konto: Bargeld je Waehrung " + string.Join(", ", slot.Cash.Select(kv => string.Create(CultureInfo.InvariantCulture, $"{kv.Key} {kv.Value:N2}"))) + "."); // Die Kontobasiswaehrung ist die mit Kurs 1: der Broker nennt sie nicht ausdruecklich, // aber er rechnet sie auch nicht um. var baseCurrency = slot.Rates.FirstOrDefault(kv => kv.Value == 1m).Key; if (slot.Rates.Count > 0) _logger.Info(LogModule, "Konto: Umrechnungskurse " + string.Join(", ", slot.Rates.Select(kv => string.Create(CultureInfo.InvariantCulture, $"{kv.Key} {kv.Value:0.####}"))) + (baseCurrency is null ? "." : $" (Basis {baseCurrency}).")); return new AccountState( ReadDecimal(slot.Values, "NetLiquidation"), ReadDecimal(slot.Values, "AvailableFunds"), ReadDecimal(slot.Values, "TotalCashValue"), slot.Cash.Count == 0 ? null : slot.Cash, slot.Rates.Count == 0 ? null : slot.Rates, baseCurrency); } finally { _accounts.TryRemove(id, out _); try { _socket.cancelAccountSummary(id); } catch { /* Verbindung bereits weg. */ } } } /// /// Bestand samt Bewertung. Nutzt reqAccountUpdates statt reqPositions, weil nur /// dieser Weg Marktwert und unrealisierten G/V mitliefert. Es ist ein Abonnement – wir melden /// uns nach dem ersten vollständigen Stand wieder ab. /// public async Task> RequestPositionsAsync(TimeSpan timeout, CancellationToken ct) { var account = _account; if (string.IsNullOrWhiteSpace(account)) { _logger.Warn(LogModule, "Kontonummer unbekannt – Positionen nicht abrufbar."); return Array.Empty(); } var slot = new PortfolioSlot(); if (Interlocked.CompareExchange(ref _portfolio, slot, null) is not null) { _logger.Warn(LogModule, "Es läuft bereits eine Positionsabfrage."); return Array.Empty(); } try { _socket.reqAccountUpdates(true, account); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, "Zeitüberschreitung bei der Positionsabfrage."); return Array.Empty(); } return slot.Positions; } finally { try { _socket.reqAccountUpdates(false, account); } catch { /* Verbindung bereits weg. */ } Interlocked.Exchange(ref _portfolio, null); } } /// /// Ausführungen. Kommissionen kommen über einen eigenen Callback und treffen oft erst nach /// execDetailsEnd ein – deshalb die kurze Nachlauffrist, bevor zusammengeführt wird. /// public async Task> RequestExecutionsAsync(DateTime? since, TimeSpan timeout, CancellationToken ct) { var id = NextRequestId(); var slot = new ExecutionSlot(); _executions[id] = slot; try { var filter = new ExecutionFilter(); if (since is { } from) filter.Time = IbkrMapping.FormatExecutionFilterTime(from); _socket.reqExecutions(id, filter); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, "Zeitüberschreitung bei der Abfrage der Ausführungen."); return Array.Empty(); } await Task.Delay(TimeSpan.FromSeconds(1), ct).ConfigureAwait(false); LogTimeProvenance(slot.Timestamps); return slot.Items .Select(e => slot.Commissions.TryGetValue(e.ExecId, out var c) ? e with { Commission = c.Amount, CommissionCurrency = c.Currency } : e) .OrderByDescending(e => e.Time) .ToList(); } finally { _executions.TryRemove(id, out _); } } public async Task PlaceOrderAsync(OrderRequest request, Contract contract, TimeSpan timeout, CancellationToken ct) { if (request.Type == OrderType.Limit && request.LimitPrice is null or <= 0) return OrderResult.Fail($"Limit-Order für {request.Symbol} ohne gültigen Limitpreis – nicht platziert."); var orderId = NextOrderId(); if (orderId < 0) return OrderResult.Fail("Keine gültige Order-ID von TWS erhalten – Verbindung nicht bereit."); var slot = new OrderSlot(); _orders[orderId] = slot; try { _socket.placeOrder(orderId, contract, IbkrMapping.BuildOrder(request, orderId, _account)); // Ab hier ist die Order abgesendet. Alles Weitere entscheidet nur noch darüber, ob wir // BEWEISEN können, dass beim Broker nichts liegt. Ohne Beweis lautet die Antwort // "Working" – und gebundenes Kapital bleibt gebunden. var id = orderId.ToString(CultureInfo.InvariantCulture); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { var status = string.IsNullOrEmpty(slot.Status) ? "keine Rückmeldung" : slot.Status; return OrderResult.Working(id, status, $"Order {orderId} ({request.Side} {request.Quantity}x {request.Symbol}) wurde übermittelt, " + $"blieb aber {timeout.TotalSeconds:F0}s ohne Endstatus (zuletzt: {status}). " + "Sie kann bei IBKR weiterhin aktiv sein."); } if (slot.Error is not null) return slot.Rejected ? OrderResult.Fail($"Order {orderId} abgelehnt: {slot.Error}") // Eine Meldung, die die Order nicht beweisbar beendet, etwa ein // Verbindungsabbruch: Unsere Sicht ist weg, die Order womoeglich nicht. : OrderResult.Working(id, slot.Status, $"Order {orderId}: {slot.Error} – ob sie beim Broker liegt, ist damit nicht entschieden."); if (slot.Status == "Filled" && slot.Filled > 0) return OrderResult.Filled(id, slot.Filled, (decimal)slot.AvgFillPrice); if (IbkrMapping.IsDeadStatus(slot.Status)) return OrderResult.Fail($"Order {orderId} endete ohne Ausführung (Status {slot.Status})."); return OrderResult.Working(id, slot.Status, $"Order {orderId} steht auf {slot.Status}" + (slot.Notice is null ? "" : $" – {slot.Notice}") + "."); } finally { _orders.TryRemove(orderId, out _); } } /// /// Fragt das Orderbuch des Kontos ab – alle offenen Orders, auch die anderer Client-IDs und die /// aus früheren Sitzungen. /// /// reqAllOpenOrders statt reqOpenOrders: Letzteres liefert nur, was dieser /// Client selbst abgesetzt hat. Eine Order, die der Dienst vor einem Neustart platziert hat, /// fände man damit nicht – und genau die ist die gefährliche. /// public async Task> RequestOpenOrdersAsync(TimeSpan timeout, CancellationToken ct) { var slot = new OpenOrderSlot(); _openOrders = slot; try { _socket.reqAllOpenOrders(); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, "Zeitüberschreitung bei der Abfrage der offenen Orders. Das Ergebnis ist unvollständig " + "und darf nicht als \"keine offenen Orders\" gelesen werden."); return Array.Empty(); } lock (slot.Orders) return slot.Orders.ToList(); } finally { _openOrders = null; } } /// /// Nimmt eine Order zurück und wartet auf die Bestätigung des Brokers. /// /// Ohne das Warten wäre die Rückmeldung wertlos: cancelOrder ist ein Zuruf, keine /// Zusage. Bestätigt der Broker nicht, liegt die Order weiter – und der Aufrufer muss das /// erfahren, statt sie für erledigt zu halten. /// public async Task CancelOrderAsync(int orderId, TimeSpan timeout, CancellationToken ct) { var slot = new OrderSlot(); _orders[orderId] = slot; try { _socket.cancelOrder(orderId); if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false)) { _logger.Warn(LogModule, $"Order {orderId}: keine Bestätigung der Stornierung."); return false; } var cancelled = IbkrMapping.IsDeadStatus(slot.Status) || slot.Rejected; _logger.Info(LogModule, cancelled ? $"Order {orderId} storniert." : $"Order {orderId} nicht storniert (Status {slot.Status})."); return cancelled; } finally { _orders.TryRemove(orderId, out _); } } // ─── EWrapper-Callbacks (laufen auf dem Reader-Thread) public override void openOrder(int orderId, Contract contract, Order order, OrderState state) { var slot = _openOrders; if (slot is null) return; var mapped = IbkrMapping.ToBrokerOrder(orderId, contract, order, state); lock (slot.Orders) { // TWS wiederholt dieselbe Order gelegentlich innerhalb eines Abrufs. if (!slot.Orders.Any(o => o.OrderId == mapped.OrderId && o.PermId == mapped.PermId)) slot.Orders.Add(mapped); } } public override void openOrderEnd() => _openOrders?.Complete(); public override void nextValidId(int orderId) { Interlocked.Exchange(ref _nextOrderId, orderId); _handshake.TrySetResult(true); } public override void managedAccounts(string accountsList) => _account = accountsList?.Split(',').FirstOrDefault(a => !string.IsNullOrWhiteSpace(a))?.Trim(); public override void contractDetails(int reqId, ContractDetails contractDetails) { if (_contracts.TryGetValue(reqId, out var slot)) slot.First ??= contractDetails.Contract; } public override void contractDetailsEnd(int reqId) { if (_contracts.TryGetValue(reqId, out var slot)) slot.Complete(); } public override void tickPrice(int tickerId, int field, double price, TickAttrib attribs) { if (price <= 0 || !_quotes.TryGetValue(tickerId, out var slot)) return; if (IbkrMapping.IsLastTick(field)) slot.Last = price; else if (IbkrMapping.IsBidTick(field)) slot.Bid = price; else if (IbkrMapping.IsAskTick(field)) slot.Ask = price; else if (IbkrMapping.IsCloseTick(field)) slot.Close = price; } public override void tickSnapshotEnd(int tickerId) { if (_quotes.TryGetValue(tickerId, out var slot)) slot.Complete(); } public override void accountSummary(int reqId, string account, string tag, string value, string currency) { if (!_accounts.TryGetValue(reqId, out var slot)) return; slot.Values[tag] = value; // Aus dem Ledger interessiert genau ein Wert je Waehrung: der Bargeldbestand. "BASE" ist // die Zusammenfassung in Kontobasiswaehrung und keine eigene Waehrung - wuerde sie mit // aufgenommen, zaehlte dasselbe Geld zweimal. if (tag.Equals("TotalCashBalance", StringComparison.OrdinalIgnoreCase) && !string.IsNullOrWhiteSpace(currency) && !currency.Equals("BASE", StringComparison.OrdinalIgnoreCase) && decimal.TryParse(value, NumberStyles.Any, CultureInfo.InvariantCulture, out var cash)) { slot.Cash[currency.ToUpperInvariant()] = cash; } // Der Umrechnungskurs in die Kontobasiswaehrung, ebenfalls aus dem Ledger. Damit rechnet // das Kapitalmodell Positionswerte und Reservierungen fremder Waehrungen um - und zwar mit // genau den Kursen, aus denen der Broker auch NetLiquidation bildet. if (tag.Equals("ExchangeRate", StringComparison.OrdinalIgnoreCase) && !string.IsNullOrWhiteSpace(currency) && !currency.Equals("BASE", StringComparison.OrdinalIgnoreCase) && decimal.TryParse(value, NumberStyles.Any, CultureInfo.InvariantCulture, out var rate) && rate > 0m) { slot.Rates[currency.ToUpperInvariant()] = rate; } } public override void securityDefinitionOptionParameter( int reqId, string exchange, int underlyingConId, string tradingClass, string multiplier, HashSet expirations, HashSet strikes) { if (!_chains.TryGetValue(reqId, out var slot)) return; // Jeden Satz aufnehmen. TWS meldet je Boerse UND Handelsklasse einen eigenen; welcher der // richtige ist, entscheidet sich erst, wenn alle da sind. var m = int.TryParse(multiplier, NumberStyles.Integer, CultureInfo.InvariantCulture, out var parsed) && parsed > 0 ? parsed : 100; var expiries = new SortedSet(); foreach (var raw in expirations) if (DateTime.TryParseExact(raw, "yyyyMMdd", CultureInfo.InvariantCulture, DateTimeStyles.None, out var expiry)) expiries.Add(DateOnly.FromDateTime(expiry)); var parsedStrikes = new SortedSet(); foreach (var strike in strikes.Where(x => x > 0)) parsedStrikes.Add((decimal)strike); if (expiries.Count == 0 || parsedStrikes.Count == 0) return; lock (slot.Series) slot.Series.Add(new OptionSeries(exchange, tradingClass, m, expiries, parsedStrikes)); } public override void securityDefinitionOptionParameterEnd(int reqId) { if (_chains.TryGetValue(reqId, out var slot)) slot.Complete(); } public override void tickOptionComputation( int tickerId, int field, double impliedVolatility, double delta, double optPrice, double pvDividend, double gamma, double vega, double theta, double undPrice) { if (!_greeks.TryGetValue(tickerId, out var slot)) return; // Nur das Modell-Feld. Die Varianten zu Geld-, Brief- und letztem Kurs fehlen bei // illiquiden Serien und liefern sonst ein Delta aus einem Kurs, den niemand stellt. if (!IbkrMapping.IsModelOptionTick(field)) return; slot.Value = new OptionGreeks( Sane(impliedVolatility), Sane(delta), Sane(gamma), Sane(vega), Sane(theta), Sane(optPrice), Sane(undPrice), IbkrMapping.IsDelayedOptionTick(field)); slot.Complete(); } /// /// TWS meldet fehlende Werte als double.MaxValue bzw. -1 und nicht als „kein Wert". /// Unverändert übernommen ergäben sie eine Bewertung, die formal vollständig aussieht. /// private static decimal? Sane(double value) => double.IsNaN(value) || double.IsInfinity(value) || Math.Abs(value) >= 1e10 || value == -1d ? null : (decimal)value; public override void accountSummaryEnd(int reqId) { if (_accounts.TryGetValue(reqId, out var slot)) slot.Complete(); } public override void updatePortfolio(Contract contract, double position, double marketPrice, double marketValue, double averageCost, double unrealizedPNL, double realizedPNL, string accountName) { // Glattgestellte Positionen meldet TWS mit Menge 0 weiter – die gehören nicht in den Bestand. if (_portfolio is not { } slot || position == 0) return; slot.Positions.Add(new BrokerPosition { Symbol = contract.Symbol, SecType = contract.SecType, Currency = contract.Currency, ConId = contract.ConId, Quantity = (decimal)position, AvgCost = (decimal)averageCost, MarketPrice = (decimal)marketPrice, MarketValue = (decimal)marketValue, UnrealizedPnl = (decimal)unrealizedPNL, Instrument = IbkrMapping.ToInstrument(contract) }); } public override void accountDownloadEnd(string account) => _portfolio?.Complete(); /// Antwort auf reqCurrentTime – Sekunden seit Epoch, Basis des Uhrenvergleichs. public override void currentTime(long time) => _serverTime?.TrySetResult(time); public override void execDetails(int reqId, Contract contract, Execution execution) { if (!_executions.TryGetValue(reqId, out var slot)) return; var stamp = IbkrMapping.ResolveExecutionTime(execution.Time, AppTimeZone.Current); slot.Timestamps.Add(stamp); slot.Items.Add(new BrokerExecution { ExecId = execution.ExecId, Time = stamp.Utc ?? DateTime.MinValue, Symbol = contract.Symbol, SecType = contract.SecType, Side = IbkrMapping.ParseSide(execution.Side), Quantity = (decimal)execution.Shares, Price = (decimal)execution.Price, Exchange = execution.Exchange ?? "", OrderId = execution.OrderId }); } public override void execDetailsEnd(int reqId) { if (_executions.TryGetValue(reqId, out var slot)) slot.Complete(); } public override void commissionReport(CommissionReport report) { // Der Callback trägt keine reqId; die ExecId ordnet ihn der Ausführung zu. if (report.Commission is <= 0 or >= 1e100) return; foreach (var slot in _executions.Values) slot.Commissions[report.ExecId] = ((decimal)report.Commission, report.Currency ?? ""); } public override void orderStatus(int orderId, string status, double filled, double remaining, double avgFillPrice, int permId, int parentId, double lastFillPrice, int clientId, string whyHeld, double mktCapPrice) { // reqAllOpenOrders liefert zu jeder Order beide Callbacks. openOrder kennt die Fuellmenge // nicht - sie kommt hier nach und wird in die laufende Abfrage nachgetragen. if (_openOrders is { } book) { lock (book.Orders) { var index = book.Orders.FindIndex(o => o.OrderId == orderId); if (index >= 0) book.Orders[index] = book.Orders[index] with { Status = status, Filled = (int)filled, Remaining = (int)remaining }; } } if (!_orders.TryGetValue(orderId, out var slot)) return; slot.Status = status; slot.Filled = (int)filled; slot.AvgFillPrice = avgFillPrice; if (IbkrMapping.IsTerminalStatus(status)) slot.Complete(); } public override void error(int id, int errorCode, string errorMsg) { if (IbkrMapping.IsInformational(errorCode)) { _logger.Info(LogModule, $"[{errorCode}] {errorMsg}"); return; } var text = $"[{errorCode}] {errorMsg}"; // Ordermeldungen zuerst und getrennt: Ob eine Meldung eine Order beendet, entscheidet der // Code – nicht die Tatsache, dass sie über den Fehler-Callback kam. IBKR schickt über // denselben Weg auch Hinweise zu angenommenen Orders (#488). if (id >= 0 && _orders.TryGetValue(id, out var order)) { if (IbkrMapping.IsOrderNotice(errorCode)) { order.Notice = errorMsg; _logger.Info(LogModule, $"Order {id} angenommen, Hinweis des Brokers: {text}"); return; } order.Rejected = IbkrMapping.IsOrderRejection(errorCode); order.Fail(text); return; } if (id >= 0 && FailPending(id, text)) return; _logger.Warn(LogModule, text); } public override void error(Exception e) { // Ein abgelehnter Socket heißt schlicht: TWS läuft (noch) nicht. Das ist beim trägen // Verbinden der Normalfall und wird von ConnectAsync bereits gemeldet – kein Stacktrace. if (e is System.Net.Sockets.SocketException) { _logger.Info(LogModule, $"Socket nicht erreichbar: {e.Message}"); return; } _logger.Error(LogModule, "API-Ausnahme", e); } public override void error(string str) => _logger.Error(LogModule, str); public override void connectionClosed() { _ready = false; FailAllPending("Verbindung zur TWS wurde geschlossen."); if (!_disposed) _logger.Warn(LogModule, "Verbindung zur TWS wurde geschlossen."); } // ─── Hilfsmittel ────────────────────────────────────────────────────────── private int NextRequestId() => Interlocked.Increment(ref _nextRequestId); /// /// Vergibt die nächste Order-ID. Die erste Vergabe liefert genau die von TWS über /// nextValidId gemeldete ID; ohne Handshake bleibt der Wert negativ und damit ungültig. /// private int NextOrderId() => Interlocked.Increment(ref _nextOrderId) - 1; private bool FailPending(int id, string error) { if (_quotes .TryGetValue(id, out var q)) { q.Fail(error); return true; } if (_accounts .TryGetValue(id, out var a)) { a.Fail(error); return true; } if (_contracts .TryGetValue(id, out var c)) { c.Fail(error); return true; } // _orders fehlt hier absichtlich: siehe error(int, int, string) – dort wird unterschieden, // ob eine Meldung die Order beendet oder nur begleitet. if (_executions.TryGetValue(id, out var e)) { e.Fail(error); return true; } return false; } private void FailAllPending(string error) { foreach (var slot in _quotes.Values) slot.Fail(error); foreach (var slot in _accounts.Values) slot.Fail(error); foreach (var slot in _contracts.Values) slot.Fail(error); foreach (var slot in _orders.Values) slot.Fail(error); foreach (var slot in _executions.Values) slot.Fail(error); _portfolio?.Fail(error); } private static decimal ReadDecimal(IReadOnlyDictionary values, string tag) => values.TryGetValue(tag, out var raw) && decimal.TryParse(raw, NumberStyles.Any, CultureInfo.InvariantCulture, out var parsed) ? parsed : 0m; private static TaskCompletionSource NewTcs() => NewTcs(); private static TaskCompletionSource NewTcs() => new(TaskCreationOptions.RunContinuationsAsynchronously); private static async Task WaitAsync(Task task, TimeSpan timeout, CancellationToken ct) { using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); var finished = await Task.WhenAny(task, Task.Delay(timeout, cts.Token)).ConfigureAwait(false); cts.Cancel(); // beendet den Verzögerungs-Task, wenn die Antwort zuerst da war return finished == task; } public void Dispose() { if (_disposed) return; _disposed = true; _ready = false; FailAllPending("Broker-Verbindung wird beendet."); SafeDisconnect(); _connectLock.Dispose(); } // ─── Antwort-Slots ──────────────────────────────────────────────────────── private abstract class Slot { public readonly TaskCompletionSource Done = NewTcs(); public string? Error; public void Complete() => Done.TrySetResult(true); public void Fail(string error) { Error = error; Done.TrySetResult(false); } } private sealed class QuoteSlot : Slot { public double Last, Bid, Ask, Close; } private sealed class AccountSlot : Slot { public readonly Dictionary Values = new(StringComparer.OrdinalIgnoreCase); /// Bargeld je Waehrung aus dem $LEDGER-Abschnitt. public readonly Dictionary Cash = new(StringComparer.OrdinalIgnoreCase); /// Umrechnungskurs je Waehrung in die Kontobasiswaehrung ($LEDGER, Tag ExchangeRate). public readonly Dictionary Rates = new(StringComparer.OrdinalIgnoreCase); } private sealed class ContractSlot : Slot { public Contract? First; } private sealed class OptionChainSlot : Slot { /// Alle gemeldeten Serien. Die Auswahl faellt erst am Ende - siehe BestSeries. public readonly List Series = new(); } /// Eine Options-Serie, wie TWS sie meldet: Boerse, Handelsklasse, Kontraktgroesse. private sealed record OptionSeries( string Exchange, string TradingClass, int Multiplier, SortedSet Expiries, SortedSet Strikes); private sealed class GreeksSlot : Slot { public OptionGreeks? Value; } private sealed class OrderSlot : Slot { public string Status = ""; public int Filled; public double AvgFillPrice; /// /// Steht fest, dass beim Broker nichts liegt? Nur dann darf gebundenes Kapital zurück. /// Vorgabe false: Ein Verbindungsabbruch beendet unsere Sicht auf die Order, nicht /// die Order. /// public bool Rejected; /// Hinweis des Brokers zu einer angenommenen Order (etwa 399). public string? Notice; } /// /// Das Orderbuch ist – wie das Portfolio – ein kontoweites Abonnement ohne eigene reqId. /// private sealed class OpenOrderSlot : Slot { public readonly List Orders = new(); } private sealed class PortfolioSlot : Slot { public readonly List Positions = new(); } private sealed class ExecutionSlot : Slot { public readonly List Items = new(); public readonly ConcurrentDictionary Commissions = new(); // Herkunft der Zeitangaben, damit nach dem Abruf zusammengefasst werden kann, wie viele // Zeitpunkte TWS gemeldet und wie viele wir angenommen haben. public readonly List Timestamps = new(); } }