RichardandClaude Opus 5
87194bfc48
ResolveExecutionTime liefert neben dem UTC-Zeitpunkt jetzt die Herkunft der verwendeten Zeitzone (gemeldet / angenommen / unbekannt / unlesbar). Bisher war im Nachhinein nicht unterscheidbar, ob ein Buchungszeitpunkt von TWS stammte oder eine Annahme war - genau der Fehler, der beim Umzug zwischen EU- und US-Host lautlos entsteht. IbkrConnection schreibt beim Verbinden einmalig Betriebszeitzone, Systemzeitzone und den Versatz zur TWS-Serverzeit ins Log; ab 5 s Abweichung gilt die Uhr des Hosts als verstellt. TWS-Setup-Checkliste um den Linux-Abschnitt ergaenzt (Betrieb und Umgebung unterscheiden sich, das Protokoll nicht). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
755 lines
30 KiB
C#
755 lines
30 KiB
C#
using System.Collections.Concurrent;
|
||
using System.Globalization;
|
||
using IBApi;
|
||
using IBKRTrader.Core.Logging;
|
||
using IBKRTrader.Core.Time;
|
||
|
||
namespace IBKRTrader.Core.Trading.Ibkr;
|
||
|
||
/// <summary>
|
||
/// 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 <see cref="TaskCompletionSource{TResult}"/> 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.
|
||
/// </summary>
|
||
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<int, QuoteSlot> _quotes = new();
|
||
private readonly ConcurrentDictionary<int, AccountSlot> _accounts = new();
|
||
private readonly ConcurrentDictionary<int, ContractSlot> _contracts = new();
|
||
private readonly ConcurrentDictionary<int, OrderSlot> _orders = new();
|
||
private readonly ConcurrentDictionary<int, ExecutionSlot> _executions = new();
|
||
private readonly ConcurrentDictionary<string, Contract> _contractCache = new(StringComparer.OrdinalIgnoreCase);
|
||
|
||
// Portfolio-Abruf ist ein kontoweites Abonnement, keine reqId-Anfrage – daher nur ein Slot.
|
||
private PortfolioSlot? _portfolio;
|
||
|
||
private TaskCompletionSource<bool> _handshake = NewTcs();
|
||
private TaskCompletionSource<long>? _serverTime;
|
||
|
||
/// <summary>Ab diesem Versatz zur TWS-Serverzeit gilt die Uhr des Hosts als verstellt.</summary>
|
||
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);
|
||
}
|
||
|
||
/// <summary>Kontonummer, die TWS beim Verbinden gemeldet hat (z. B. "DUR371528").</summary>
|
||
public string? Account => _account;
|
||
|
||
// ─── Verbindung ───────────────────────────────────────────────────────────
|
||
|
||
public async Task<bool> 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<bool> 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;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Schreibt den vollständigen Zeitkontext einer Verbindung ins Log: Betriebszeitzone,
|
||
/// Systemzeitzone und den Versatz zur Uhr des TWS-Servers.
|
||
///
|
||
/// <para><b>Wozu:</b> 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.</para>
|
||
/// </summary>
|
||
private async Task LogTimeContextAsync(CancellationToken ct)
|
||
{
|
||
var context = $"Zeitkontext: Betriebszeitzone {AppTimeZone.CurrentId}, " +
|
||
$"Systemzeitzone {TimeZoneInfo.Local.Id}";
|
||
|
||
var pending = NewTcs<long>();
|
||
_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;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Fasst nach jedem Abruf zusammen, wie viele Zeitstempel TWS mit Zonenangabe gemeldet hat und
|
||
/// wie viele über die Betriebszeitzone <b>angenommen</b> wurden. Die angenommenen sind die
|
||
/// Stelle, an der ein Wechsel zwischen EU- und US-Host lautlos danebenliegt.
|
||
/// </summary>
|
||
private void LogTimeProvenance(IReadOnlyList<IbkrMapping.ExecutionTimestamp> 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 ─────────────────────────────────────────────────────────────
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
public async Task<Contract?> ResolveContractAsync(string symbol, TimeSpan timeout, CancellationToken ct)
|
||
{
|
||
if (_contractCache.TryGetValue(symbol, out var cached)) return cached;
|
||
|
||
var id = NextRequestId();
|
||
var slot = new ContractSlot();
|
||
_contracts[id] = slot;
|
||
try
|
||
{
|
||
_socket.reqContractDetails(id, IbkrMapping.Stock(symbol));
|
||
|
||
if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false))
|
||
{
|
||
_logger.Warn(LogModule, $"Zeitüberschreitung bei der Kontraktsuche für {symbol}.");
|
||
return null;
|
||
}
|
||
if (slot.Error is not null)
|
||
{
|
||
_logger.Warn(LogModule, $"Kontrakt {symbol} nicht auflösbar: {slot.Error}");
|
||
return null;
|
||
}
|
||
if (slot.First is null)
|
||
{
|
||
_logger.Warn(LogModule, $"Kein Kontrakt für {symbol} gefunden.");
|
||
return null;
|
||
}
|
||
|
||
_contractCache[symbol] = slot.First;
|
||
return slot.First;
|
||
}
|
||
finally
|
||
{
|
||
_contracts.TryRemove(id, out _);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
public async Task<Quote?> 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 _);
|
||
}
|
||
}
|
||
|
||
public async Task<AccountState?> RequestAccountAsync(TimeSpan timeout, CancellationToken ct)
|
||
{
|
||
var id = NextRequestId();
|
||
var slot = new AccountSlot();
|
||
_accounts[id] = slot;
|
||
try
|
||
{
|
||
_socket.reqAccountSummary(id, "All", "NetLiquidation,AvailableFunds");
|
||
|
||
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;
|
||
}
|
||
|
||
return new AccountState(
|
||
ReadDecimal(slot.Values, "NetLiquidation"),
|
||
ReadDecimal(slot.Values, "AvailableFunds"));
|
||
}
|
||
finally
|
||
{
|
||
_accounts.TryRemove(id, out _);
|
||
try { _socket.cancelAccountSummary(id); }
|
||
catch { /* Verbindung bereits weg. */ }
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Bestand samt Bewertung. Nutzt <c>reqAccountUpdates</c> statt <c>reqPositions</c>, 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.
|
||
/// </summary>
|
||
public async Task<IReadOnlyList<BrokerPosition>> RequestPositionsAsync(TimeSpan timeout, CancellationToken ct)
|
||
{
|
||
var account = _account;
|
||
if (string.IsNullOrWhiteSpace(account))
|
||
{
|
||
_logger.Warn(LogModule, "Kontonummer unbekannt – Positionen nicht abrufbar.");
|
||
return Array.Empty<BrokerPosition>();
|
||
}
|
||
|
||
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<BrokerPosition>();
|
||
}
|
||
|
||
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<BrokerPosition>();
|
||
}
|
||
|
||
return slot.Positions;
|
||
}
|
||
finally
|
||
{
|
||
try { _socket.reqAccountUpdates(false, account); }
|
||
catch { /* Verbindung bereits weg. */ }
|
||
Interlocked.Exchange(ref _portfolio, null);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ausführungen. Kommissionen kommen über einen eigenen Callback und treffen oft erst nach
|
||
/// <c>execDetailsEnd</c> ein – deshalb die kurze Nachlauffrist, bevor zusammengeführt wird.
|
||
/// </summary>
|
||
public async Task<IReadOnlyList<BrokerExecution>> 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<BrokerExecution>();
|
||
}
|
||
|
||
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<OrderResult> 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));
|
||
|
||
if (!await WaitAsync(slot.Done.Task, timeout, ct).ConfigureAwait(false))
|
||
{
|
||
// Die Order ist übermittelt und liegt womöglich aktiv bei IBKR. Sie hier als
|
||
// reinen Fehlschlag zu buchen wäre falsch – deshalb Order-ID und letzter Status
|
||
// in die Meldung, damit der Betreiber sie in TWS wiederfindet.
|
||
var status = string.IsNullOrEmpty(slot.Status) ? "keine Rückmeldung" : slot.Status;
|
||
return OrderResult.Fail(
|
||
$"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 und muss dort geprüft werden.");
|
||
}
|
||
|
||
if (slot.Error is not null)
|
||
return OrderResult.Fail($"Order {orderId} abgelehnt: {slot.Error}");
|
||
|
||
if (slot.Status == "Filled" && slot.Filled > 0)
|
||
return OrderResult.Filled(orderId.ToString(CultureInfo.InvariantCulture),
|
||
slot.Filled, (decimal)slot.AvgFillPrice);
|
||
|
||
return OrderResult.Fail($"Order {orderId} endete ohne Ausführung (Status {slot.Status}).");
|
||
}
|
||
finally
|
||
{
|
||
_orders.TryRemove(orderId, out _);
|
||
}
|
||
}
|
||
|
||
// ─── EWrapper-Callbacks (laufen auf dem Reader-Thread) ────────────────────
|
||
|
||
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)) slot.Values[tag] = 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
|
||
});
|
||
}
|
||
|
||
public override void accountDownloadEnd(string account) => _portfolio?.Complete();
|
||
|
||
/// <summary>Antwort auf <c>reqCurrentTime</c> – Sekunden seit Epoch, Basis des Uhrenvergleichs.</summary>
|
||
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)
|
||
{
|
||
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}";
|
||
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);
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
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; }
|
||
if (_orders .TryGetValue(id, out var o)) { o.Fail(error); return true; }
|
||
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<string, string> values, string tag) =>
|
||
values.TryGetValue(tag, out var raw) &&
|
||
decimal.TryParse(raw, NumberStyles.Any, CultureInfo.InvariantCulture, out var parsed)
|
||
? parsed
|
||
: 0m;
|
||
|
||
private static TaskCompletionSource<bool> NewTcs() => NewTcs<bool>();
|
||
|
||
private static TaskCompletionSource<T> NewTcs<T>() =>
|
||
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
|
||
private static async Task<bool> 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<bool> 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<string, string> Values = new(StringComparer.OrdinalIgnoreCase);
|
||
}
|
||
|
||
private sealed class ContractSlot : Slot
|
||
{
|
||
public Contract? First;
|
||
}
|
||
|
||
private sealed class OrderSlot : Slot
|
||
{
|
||
public string Status = "";
|
||
public int Filled;
|
||
public double AvgFillPrice;
|
||
}
|
||
|
||
private sealed class PortfolioSlot : Slot
|
||
{
|
||
public readonly List<BrokerPosition> Positions = new();
|
||
}
|
||
|
||
private sealed class ExecutionSlot : Slot
|
||
{
|
||
public readonly List<BrokerExecution> Items = new();
|
||
public readonly ConcurrentDictionary<string, (decimal Amount, string Currency)> 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<IbkrMapping.ExecutionTimestamp> Timestamps = new();
|
||
}
|
||
}
|