523 lines
19 KiB
Python
523 lines
19 KiB
Python
"""DTC Binary protocol client for Sierra Chart (default: localhost:11099).
|
||
|
||
Binary wire format (no pragma pack → default C alignment):
|
||
Every message: [uint16 Size][uint16 Type][payload...]
|
||
Size = total byte count including the header.
|
||
|
||
Struct offsets used here match MSVC/GCC x64 default alignment:
|
||
- int64/double are 8-byte aligned (padding inserted before them when needed)
|
||
|
||
Supported messages:
|
||
Send: LOGON_REQUEST, HEARTBEAT, HISTORICAL_ORDER_FILLS_REQUEST,
|
||
OPEN_ORDERS_REQUEST
|
||
Receive: LOGON_RESPONSE, HEARTBEAT, HISTORICAL_ORDER_FILL_RESPONSE,
|
||
ORDER_UPDATE
|
||
"""
|
||
|
||
import re
|
||
import socket
|
||
import struct
|
||
import threading
|
||
import time
|
||
from datetime import datetime
|
||
|
||
# ── DTC message type constants ────────────────────────────────────────────────
|
||
|
||
_LOGON_REQUEST = 1
|
||
_LOGON_RESPONSE = 2
|
||
_HEARTBEAT = 3
|
||
_OPEN_ORDERS_REQ = 300
|
||
_ORDER_UPDATE = 301
|
||
_HIST_FILLS_REQ = 303
|
||
_HIST_FILL_RESP = 304
|
||
|
||
_PROTOCOL_VER = 8
|
||
_LOGON_SUCCESS = 1
|
||
_BUY = 1
|
||
_SELL = 2
|
||
|
||
|
||
# ── Encoding helpers ──────────────────────────────────────────────────────────
|
||
|
||
def _s(text, n):
|
||
"""Return exactly n bytes: ASCII-encoded, null-padded."""
|
||
b = (text or "").encode("ascii", errors="replace")[:n]
|
||
return b.ljust(n, b"\x00")
|
||
|
||
|
||
def _unpack_str(data, offset, n):
|
||
return data[offset:offset + n].rstrip(b"\x00").decode("ascii", errors="replace")
|
||
|
||
|
||
def _unpack_ts(data, offset):
|
||
"""Read int64 Unix timestamp at offset → datetime or None."""
|
||
raw = struct.unpack_from("<q", data, offset)[0]
|
||
try:
|
||
return datetime.fromtimestamp(raw) if 0 < raw < 4_102_444_800 else None
|
||
except (OSError, OverflowError):
|
||
return None
|
||
|
||
|
||
def _base_symbol(sc_symbol, instruments):
|
||
"""'MNQM6' → 'MNQ' if 'MNQ' is in instruments, else None."""
|
||
m = re.search(r'([A-Z]{2,4})[A-Z]\d', sc_symbol)
|
||
if m and m.group(1) in instruments:
|
||
return m.group(1)
|
||
return sc_symbol if sc_symbol in instruments else None
|
||
|
||
|
||
# ── Message builders ──────────────────────────────────────────────────────────
|
||
|
||
# Format strings (little-endian, explicit padding to match C struct alignment)
|
||
_FMT_LOGON = "<HHi32s32s64siiii32s64s32si" # 284 bytes
|
||
_FMT_HEARTBEAT = "<HHIq" # 16 bytes
|
||
_FMT_HIST_FILLS = "<HHi32si32sB7xq" # 92 bytes (7x = padding before int64)
|
||
_FMT_OPEN_ORD = "<HHii" # 12 bytes
|
||
|
||
|
||
def _build_logon(heartbeat_secs=10):
|
||
size = struct.calcsize(_FMT_LOGON)
|
||
return struct.pack(
|
||
_FMT_LOGON,
|
||
size, _LOGON_REQUEST, _PROTOCOL_VER,
|
||
_s("", 32), _s("", 32), _s("GizerBridge", 64),
|
||
0, 0, heartbeat_secs, 0,
|
||
_s("", 32), _s("", 64), _s("gizerbridge", 32), 0,
|
||
)
|
||
|
||
|
||
def _build_heartbeat():
|
||
size = struct.calcsize(_FMT_HEARTBEAT)
|
||
return struct.pack(_FMT_HEARTBEAT, size, _HEARTBEAT, 0, int(time.time()))
|
||
|
||
|
||
def _build_hist_request(req_id, account, num_days):
|
||
size = struct.calcsize(_FMT_HIST_FILLS)
|
||
return struct.pack(
|
||
_FMT_HIST_FILLS,
|
||
size, _HIST_FILLS_REQ,
|
||
req_id,
|
||
_s("", 32), # ServerOrderID – empty = all orders
|
||
num_days,
|
||
_s(account, 32), # TradeAccount
|
||
1, # RequestAllHistoricalFills = true
|
||
0, # StartDateTime = 0 (unused when num_days > 0)
|
||
)
|
||
|
||
|
||
def _build_open_orders_request():
|
||
size = struct.calcsize(_FMT_OPEN_ORD)
|
||
return struct.pack(_FMT_OPEN_ORD, size, _OPEN_ORDERS_REQ, 1, 0)
|
||
|
||
|
||
# ── Message parsers ───────────────────────────────────────────────────────────
|
||
|
||
# HISTORICAL_ORDER_FILL_RESPONSE – Offsets empirisch bestätigt (480-Byte-Pakete, Rithmic):
|
||
# 0 uint16 Size
|
||
# 2 uint16 Type
|
||
# 4 int32 RequestID
|
||
# 8 int32 TotalNumberMessages
|
||
# 12 int32 MessageNumber
|
||
# 16 char[64] Symbol (z.B. "MNQM6.CME")
|
||
# 80 char[16] Exchange (leer bei Rithmic, Exchange steckt im Symbol)
|
||
# 96 char[32] FillIdentifier (laufende Fill-Nr. als String, z.B. "592")
|
||
# 128 uint32 BuySell (1=BUY, 2=SELL)
|
||
# 132 uint32 unbekannt (immer 0)
|
||
# 136 double Price × 100 (z.B. 2902225.0 → 29022.25 Punkte)
|
||
# 144 int64 DateTime (Unix-Sekunden)
|
||
# 152 double Volume/Qty (Anzahl Kontrakte, z.B. 1.0)
|
||
# 160 char[32] TradeID (Rithmic Order-ID, z.B. "1031001")
|
||
# 192 char[32] unbekannt
|
||
# 224 char[32] TradeAccount (z.B. "4PRO-RQTPUT")
|
||
# 256 uint8 FillStatus (1=partial, 2=final – KEIN NoOrderFills-Flag)
|
||
# 257 char[96] InfoText (z.B. "Rithmic Direct - DTC (Filled)[final]")
|
||
# 353 char[64] unbekannt
|
||
# 416 char[64] UniqueExecutionID (z.B. "2537652419")
|
||
|
||
def _parse_fill_resp(data):
|
||
if len(data) < 260:
|
||
return None
|
||
try:
|
||
raw_price = struct.unpack_from("<d", data, 136)[0]
|
||
price = raw_price / 100.0 # Rithmic liefert price × 100
|
||
qty = struct.unpack_from("<d", data, 152)[0]
|
||
buysell = struct.unpack_from("<I", data, 128)[0]
|
||
total = struct.unpack_from("<i", data, 8)[0]
|
||
msgno = struct.unpack_from("<i", data, 12)[0]
|
||
return {
|
||
"req_id": struct.unpack_from("<i", data, 4)[0],
|
||
"total": total,
|
||
"msgno": msgno,
|
||
"symbol": _unpack_str(data, 16, 64),
|
||
"price": price,
|
||
"qty": qty,
|
||
"datetime": _unpack_ts(data, 144),
|
||
"buysell": buysell,
|
||
"account": _unpack_str(data, 224, 32),
|
||
"no_fills": total > 0 and msgno >= total,
|
||
"unique_id": _unpack_str(data, 416, 64) if len(data) >= 480 else "",
|
||
}
|
||
except struct.error:
|
||
return None
|
||
|
||
|
||
# ORDER_UPDATE field offsets (C struct, default alignment):
|
||
# 0 uint16 Size
|
||
# 2 uint16 Type
|
||
# 4 int32 RequestID
|
||
# 8 int32 TotalNumMessages
|
||
# 12 int32 MessageNumber
|
||
# 16 char[64] Symbol
|
||
# 80 char[16] Exchange
|
||
# 96 char[32] PreviousServerOrderID
|
||
# 128 char[32] ServerOrderID
|
||
# 160 char[32] ClientOrderID
|
||
# 192 char[32] ExchangeOrderID
|
||
# 224 uint32 OrderStatus
|
||
# 228 uint32 OrderUpdateReason
|
||
# 232 uint32 OrderType
|
||
# 236 uint32 BuySell
|
||
# 240 double Price1 (240 % 8 == 0 ✓)
|
||
# 248 double Price2
|
||
# 256 uint32 TimeInForce
|
||
# 260 [4 bytes padding]
|
||
# 264 int64 GoodTillDateTime (264 % 8 == 0 ✓)
|
||
# 272 double OrderQuantity
|
||
# 280 double FilledQuantity
|
||
# 288 double RemainingQuantity
|
||
# 296 double AverageFillPrice
|
||
# 304 double LastFillPrice
|
||
# 312 int64 LastFillDateTime
|
||
# 320 double LastFillQuantity
|
||
# 328 char[64] UniqueExecutionID
|
||
# 392 char[32] TradeAccount
|
||
|
||
def _parse_order_update(data):
|
||
if len(data) < 328:
|
||
return None
|
||
try:
|
||
qty = struct.unpack_from("<d", data, 320)[0]
|
||
price = struct.unpack_from("<d", data, 304)[0] / 100.0
|
||
if qty <= 0 or price <= 0:
|
||
return None # not a fill event
|
||
uid = _unpack_str(data, 328, 64) if len(data) >= 392 else ""
|
||
return {
|
||
"symbol": _unpack_str(data, 16, 64),
|
||
"buysell": struct.unpack_from("<I", data, 236)[0],
|
||
"price": price,
|
||
"qty": qty,
|
||
"datetime": _unpack_ts(data, 312),
|
||
"account": _unpack_str(data, 392, 32) if len(data) >= 424 else "",
|
||
"unique_id": uid,
|
||
}
|
||
except struct.error:
|
||
return None
|
||
|
||
|
||
# ── DTCClient ─────────────────────────────────────────────────────────────────
|
||
|
||
class DTCClient:
|
||
"""Low-level DTC Binary protocol client (not thread-safe for concurrent sends)."""
|
||
|
||
def __init__(self, host="localhost", port=11099):
|
||
self._host = host
|
||
self._port = port
|
||
self._sock = None
|
||
self._buf = b""
|
||
self._lock = threading.Lock()
|
||
|
||
def connect(self):
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(10)
|
||
sock.connect((self._host, self._port))
|
||
sock.settimeout(None)
|
||
self._sock = sock
|
||
|
||
def disconnect(self):
|
||
if self._sock:
|
||
try:
|
||
self._sock.close()
|
||
except OSError:
|
||
pass
|
||
self._sock = None
|
||
|
||
def logon(self, heartbeat_secs=10):
|
||
"""Send LOGON_REQUEST and block until LOGON_RESPONSE received."""
|
||
self._send(_build_logon(heartbeat_secs))
|
||
for msg_type, data in self._read_iter(timeout=10):
|
||
if msg_type == _LOGON_RESPONSE and len(data) >= 16:
|
||
result = struct.unpack_from("<i", data, 8)[0]
|
||
text = _unpack_str(data, 12, 96)
|
||
if result != _LOGON_SUCCESS:
|
||
raise ConnectionError(f"DTC Logon abgelehnt: {text}")
|
||
return
|
||
raise TimeoutError("Kein LOGON_RESPONSE erhalten (Timeout)")
|
||
|
||
def request_fills(self, account, num_days=30):
|
||
"""Request historical fills for account. Returns list of fill dicts."""
|
||
req_id = int(time.time() * 1000) & 0x7FFFFFFF
|
||
self._send(_build_hist_request(req_id, account, num_days))
|
||
|
||
fills = []
|
||
for msg_type, data in self._read_iter(timeout=15):
|
||
if msg_type != _HIST_FILL_RESP:
|
||
continue
|
||
|
||
f = _parse_fill_resp(data)
|
||
if not f or f["req_id"] != req_id:
|
||
continue
|
||
if f["buysell"] in (_BUY, _SELL) and f["qty"] > 0 and f["price"] > 0:
|
||
fills.append(f)
|
||
if f["no_fills"]:
|
||
break
|
||
return fills
|
||
|
||
def subscribe_order_updates(self):
|
||
"""Send OPEN_ORDERS_REQUEST so SC starts pushing ORDER_UPDATE messages."""
|
||
self._send(_build_open_orders_request())
|
||
|
||
def send_heartbeat(self):
|
||
self._send(_build_heartbeat())
|
||
|
||
def read_messages(self, timeout=5):
|
||
"""Non-blocking read: return all complete messages buffered within timeout."""
|
||
self._sock.settimeout(timeout)
|
||
msgs = []
|
||
try:
|
||
chunk = self._sock.recv(65536)
|
||
if not chunk:
|
||
raise ConnectionError("DTC-Verbindung vom Server geschlossen")
|
||
self._buf += chunk
|
||
except socket.timeout:
|
||
pass
|
||
while len(self._buf) >= 4:
|
||
size = struct.unpack_from("<H", self._buf, 0)[0]
|
||
if size < 4 or len(self._buf) < size:
|
||
break
|
||
msg_type = struct.unpack_from("<H", self._buf, 2)[0]
|
||
msgs.append((msg_type, bytes(self._buf[:size])))
|
||
self._buf = self._buf[size:]
|
||
return msgs
|
||
|
||
# ── Internal ──────────────────────────────────────────────────────────────
|
||
|
||
def _send(self, data):
|
||
with self._lock:
|
||
self._sock.sendall(data)
|
||
|
||
def _read_iter(self, timeout):
|
||
"""Generator yielding (msg_type, data) tuples until timeout."""
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
remaining = max(0.1, deadline - time.time())
|
||
self._sock.settimeout(remaining)
|
||
try:
|
||
chunk = self._sock.recv(65536)
|
||
if not chunk:
|
||
raise ConnectionError("DTC-Verbindung geschlossen")
|
||
self._buf += chunk
|
||
except socket.timeout:
|
||
pass
|
||
while len(self._buf) >= 4:
|
||
size = struct.unpack_from("<H", self._buf, 0)[0]
|
||
if size < 4 or len(self._buf) < size:
|
||
break
|
||
msg_type = struct.unpack_from("<H", self._buf, 2)[0]
|
||
data = bytes(self._buf[:size])
|
||
self._buf = self._buf[size:]
|
||
yield msg_type, data
|
||
|
||
|
||
# ── Trade reconstruction from fills ──────────────────────────────────────────
|
||
|
||
def fills_to_trades(fills, instruments):
|
||
"""Convert list of fill dicts to completed round-trip trade dicts.
|
||
|
||
Groups fills by (account, base_symbol), sorts by datetime, then matches
|
||
buys and sells until the net position returns to zero – that's one trade.
|
||
"""
|
||
groups = {}
|
||
for f in sorted(fills, key=lambda x: x.get("datetime") or datetime.min):
|
||
base = _base_symbol(f["symbol"], instruments)
|
||
if not base:
|
||
continue
|
||
groups.setdefault((f["account"], base), []).append(f)
|
||
|
||
trades = []
|
||
for (account, symbol), group in groups.items():
|
||
trades.extend(_match_fills(group, account, symbol, instruments))
|
||
return trades
|
||
|
||
|
||
def _match_fills(fills, account, symbol, instruments):
|
||
trades, pending, position = [], [], 0.0
|
||
for f in fills:
|
||
delta = f["qty"] if f["buysell"] == _BUY else -f["qty"]
|
||
position += delta
|
||
pending.append(f)
|
||
if abs(position) < 0.01:
|
||
t = _pending_to_trade(pending, account, symbol, instruments)
|
||
if t:
|
||
trades.append(t)
|
||
pending, position = [], 0.0
|
||
return trades
|
||
|
||
|
||
def _pending_to_trade(fills, account, symbol, instruments):
|
||
buys = [f for f in fills if f["buysell"] == _BUY]
|
||
sells = [f for f in fills if f["buysell"] == _SELL]
|
||
if not buys or not sells:
|
||
return None
|
||
|
||
direction = "Long" if fills[0]["buysell"] == _BUY else "Short"
|
||
entry_fs = buys if direction == "Long" else sells
|
||
exit_fs = sells if direction == "Long" else buys
|
||
|
||
e_qty = sum(f["qty"] for f in entry_fs)
|
||
x_qty = sum(f["qty"] for f in exit_fs)
|
||
ep = sum(f["price"] * f["qty"] for f in entry_fs) / e_qty
|
||
xp = sum(f["price"] * f["qty"] for f in exit_fs) / x_qty
|
||
lots = int(round(e_qty))
|
||
|
||
_fmt = "%Y-%m-%d %H:%M:%S"
|
||
entry_ts = min((f["datetime"] for f in entry_fs if f["datetime"]), default=None)
|
||
exit_ts = max((f["datetime"] for f in exit_fs if f["datetime"]), default=None)
|
||
entry_str = entry_ts.strftime(_fmt) if entry_ts else datetime.now().strftime(_fmt)
|
||
exit_str = exit_ts.strftime(_fmt) if exit_ts else datetime.now().strftime(_fmt)
|
||
|
||
cfg = instruments[symbol]
|
||
tick = cfg["tick_size"]
|
||
tick_val = cfg["tick_value"]
|
||
pnl_ticks = round((xp - ep if direction == "Long" else ep - xp) / tick) * lots
|
||
pnl = round(pnl_ticks * tick_val, 2)
|
||
|
||
if direction == "Long":
|
||
execs = f"Buy {lots} @ {ep:.4f} / {entry_str},Sell {lots} @ {xp:.4f} / {exit_str}"
|
||
else:
|
||
execs = f"Sell {lots} @ {ep:.4f} / {entry_str},Buy {lots} @ {xp:.4f} / {exit_str}"
|
||
|
||
# Trade-ID uses only actual fill datetimes — never datetime.now().
|
||
# This makes IDs identical whether generated from the historical-fill path
|
||
# or from the real-time ORDER_UPDATE path for the same trade.
|
||
_id_ts = entry_ts or exit_ts
|
||
_id_str = _id_ts.strftime(_fmt) if _id_ts else ""
|
||
|
||
return {
|
||
"trade_id": f"{account}_{symbol}_{ep:.4f}_{xp:.4f}_{_id_str}",
|
||
"instrument": symbol,
|
||
"direction": direction,
|
||
"lots": lots,
|
||
"entry_price": round(ep, 4),
|
||
"exit_price": round(xp, 4),
|
||
"entry_datetime": entry_str,
|
||
"exit_datetime": exit_str,
|
||
"pnl": pnl,
|
||
"pnl_ticks": pnl_ticks,
|
||
"executions": execs,
|
||
"account": account,
|
||
}
|
||
|
||
|
||
# ── Real-time monitor ─────────────────────────────────────────────────────────
|
||
|
||
def monitor_realtime(host, port, accounts, instruments, on_trade, stop_event, log_fn,
|
||
on_ready=None):
|
||
"""Connect to SC DTC server, load existing fills, then stream ORDER_UPDATE.
|
||
|
||
Calls on_trade(trade_dict) for each newly completed round-trip trade.
|
||
Calls on_ready(existing_ids) once the connection is established and the
|
||
historical fill scan is complete – before entering the real-time loop.
|
||
Returns set of trade IDs that were already present at startup (caller marks
|
||
these as sent so they don't get re-uploaded after a restart).
|
||
Returns None if the connection fails immediately.
|
||
"""
|
||
client = DTCClient(host, port)
|
||
try:
|
||
client.connect()
|
||
client.logon(heartbeat_secs=10)
|
||
except Exception as e:
|
||
log_fn(f"DTC-Verbindungsfehler: {e}")
|
||
return None
|
||
|
||
# ── Scan existing fills so we know what's already been uploaded ────────────
|
||
existing_ids = set()
|
||
all_fills = []
|
||
for account in accounts:
|
||
try:
|
||
fills = client.request_fills(account, num_days=30)
|
||
all_fills.extend(fills)
|
||
except Exception as e:
|
||
log_fn(f"Historische Fills ({account}): {e}")
|
||
|
||
for t in fills_to_trades(all_fills, instruments):
|
||
existing_ids.add(t["trade_id"])
|
||
|
||
# ── Subscribe to real-time ORDER_UPDATE messages ──────────────────────────
|
||
try:
|
||
client.subscribe_order_updates()
|
||
except Exception as e:
|
||
log_fn(f"ORDER_UPDATE-Subscription fehlgeschlagen: {e}")
|
||
|
||
if on_ready:
|
||
on_ready(existing_ids)
|
||
|
||
# Per-(account, symbol) state for real-time fill accumulation
|
||
rt_pending = {} # (account, symbol) → [fill dicts]
|
||
rt_position = {} # (account, symbol) → float
|
||
seen_uids = set()
|
||
|
||
last_hb = time.time()
|
||
|
||
while not stop_event.is_set():
|
||
try:
|
||
msgs = client.read_messages(timeout=3)
|
||
except Exception as e:
|
||
log_fn(f"DTC Lesefehler: {e}")
|
||
break
|
||
|
||
if time.time() - last_hb >= 10:
|
||
try:
|
||
client.send_heartbeat()
|
||
except Exception:
|
||
pass
|
||
last_hb = time.time()
|
||
|
||
for msg_type, data in msgs:
|
||
if msg_type == _HEARTBEAT:
|
||
try:
|
||
client.send_heartbeat()
|
||
except Exception:
|
||
pass
|
||
|
||
elif msg_type == _ORDER_UPDATE:
|
||
f = _parse_order_update(data)
|
||
if not f:
|
||
continue
|
||
uid = f["unique_id"]
|
||
if uid and uid in seen_uids:
|
||
continue
|
||
if uid:
|
||
seen_uids.add(uid)
|
||
|
||
base = _base_symbol(f["symbol"], instruments)
|
||
if not base or not f["account"]:
|
||
continue
|
||
|
||
key = (f["account"], base)
|
||
rt_position.setdefault(key, 0.0)
|
||
rt_pending.setdefault(key, [])
|
||
|
||
rt_position[key] += f["qty"] if f["buysell"] == _BUY else -f["qty"]
|
||
rt_pending[key].append(f)
|
||
|
||
if abs(rt_position[key]) < 0.01:
|
||
group = rt_pending.pop(key)
|
||
rt_position[key] = 0.0
|
||
t = _pending_to_trade(group, f["account"], base, instruments)
|
||
if t and t["trade_id"] not in existing_ids:
|
||
existing_ids.add(t["trade_id"])
|
||
on_trade(t)
|
||
|
||
client.disconnect()
|
||
return existing_ids
|