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

R9: Echter IbkrBrokerClient über die TWS API

Browse Source 
Broker-Adapter gegen TWS/IB Gateway, aktivierbar über IBKRSettings.UseTwsApi;
NullBrokerClient bleibt Default. TradingEnabled bleibt als zweite, unabhängige
Sicherung bestehen – ohne ihn platziert der ExecutionService keine Order.

Aufteilung (src/IBKRTrader.Core/Trading/Ibkr/):
- IbkrMapping      – reine Abbildung Core <-> TWS (Kontrakt, Order, Kurs, Port-
                     und Statusregeln), vollständig unit-getestet
- IbkrConnection   – Socket-Lebenszyklus, Reader-Thread, reqId-Korrelation über
                     TaskCompletionSource
- IbkrBrokerClient – implementiert IBrokerClient, übersetzt Fehler in leere
                     Ergebnisse (Konto 0 lässt die Risikoprüfung alles ablehnen)

Bewusste Entscheidungen:
- Träges Verbinden mit Wiederholung statt Verbindungsaufbau beim Start: TWS ist
  nach einem Neustart minutenlang nicht bereit.
- Port wird gegen den Handelsmodus geprüft; Paper-Modus auf Live-Port (oder
  umgekehrt) lässt den Broker inaktiv, statt auf dem falschen Konto zu handeln.
- MarketDataType Default 4: Paper-Konten ohne Datenabo bekommen sonst keine Kurse.
- Fehlercode 10167 ist ein Statushinweis (verzögerte Daten folgen), kein Fehler.
  Als Fehler behandelt scheiterte jede einzelne Kursabfrage.

Verifiziert gegen Paper-Konto DUR371528: Verbindung, Konto (100.105,50 EUR),
Kurse (AAPL/MSFT/NVDA, verzögert), Fehlerpfade. Orderpfad bis zur Broker-Annahme
per What-If-Order geprüft (Aktie + Option, ohne Ausführung); dabei zugleich die
Optionsberechtigung des Kontos bestätigt. Offen: echte Ausführung (Fill ->
Buchung) und asynchrone Fill-Verfolgung – beides in IBKR-Integration.md notiert.

Doku: TWS-Setup-Checkliste.md (Einstellungen für Neuinstallation) neu,
IBKR-Integration.md / ARCHITECTURE.md / README.md nachgezogen.

154/154 Tests grün.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Richard
2026-08-04 17:40:00 +02:00
co-authored by Claude Opus 5
parent 2a312ca035
commit 80afcd49c1
14 changed files with 1427 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/IBKRTrader.Core/Trading/Ibkr/IbkrConnection.cs
+489
View File
@@ -0,0 +1,489 @@
using System.Collections.Concurrent;
using System.Globalization;
using IBApi;
using IBKRTrader.Core.Logging;
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<string, Contract> _contractCache = new(StringComparer.OrdinalIgnoreCase);
private TaskCompletionSource<bool> _handshake = NewTcs();
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;
_socket.reqMarketDataType(_marketDataType);
_logger.Info(LogModule,
$"Verbunden mit {_host}:{_port} (Client {_clientId}), Konto {_account ?? "unbekannt"}.");
return true;
}
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. */ }
}
}
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 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; }
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);
}
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() =>
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;
}
}