initial release: GizerBridge v1.0
This commit is contained in:
+520
@@ -0,0 +1,520 @@
|
||||
"""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 – empirisch ermittelte Offsets (480-Byte-Pakete):
|
||||
# 0 uint16 Size
|
||||
# 2 uint16 Type
|
||||
# 4 int32 RequestID
|
||||
# 8 int32 TotalNumberMessages
|
||||
# 12 int32 MessageNumber
|
||||
# 16 char[64] Symbol
|
||||
# 80 char[16] Exchange
|
||||
# 96 char[32] ??? (ASCII-Feld, Inhalt noch unbekannt)
|
||||
# 128 uint32 BuySell (1=BUY, 2=SELL)
|
||||
# 132 uint32 ??? (4 Bytes Padding/unbekannt)
|
||||
# 136 double Price (136 % 8 == 0, 8-Byte-aligned)
|
||||
# 144 int64 DateTime (Unix-Sekunden)
|
||||
# 152 double Volume/Qty
|
||||
# 160 char[32] FillIdentifier/TradeID
|
||||
# 192 char[32] ???
|
||||
# 224 char[32] TradeAccount
|
||||
# 256 uint8 NoOrderFills
|
||||
# 257 char[96] InfoText
|
||||
# 353 char[64] UniqueExecutionID
|
||||
# 417 char[32] ExchangeOrderID
|
||||
|
||||
def _parse_fill_resp(data):
|
||||
if len(data) < 260:
|
||||
return None
|
||||
try:
|
||||
no_fills = bool(struct.unpack_from("<B", data, 256)[0]) if len(data) > 256 else False
|
||||
price = struct.unpack_from("<d", data, 136)[0]
|
||||
qty = struct.unpack_from("<d", data, 152)[0]
|
||||
buysell = struct.unpack_from("<I", data, 128)[0]
|
||||
return {
|
||||
"req_id": struct.unpack_from("<i", data, 4)[0],
|
||||
"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": no_fills,
|
||||
"unique_id": _unpack_str(data, 353, 64) if len(data) >= 417 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]
|
||||
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
|
||||
if len(data) < 16:
|
||||
continue
|
||||
if struct.unpack_from("<i", data, 4)[0] != req_id:
|
||||
continue
|
||||
|
||||
total = struct.unpack_from("<i", data, 8)[0]
|
||||
msgno = struct.unpack_from("<i", data, 12)[0]
|
||||
|
||||
f = _parse_fill_resp(data)
|
||||
if not f:
|
||||
continue
|
||||
if f["no_fills"] or (total > 0 and msgno >= total):
|
||||
break
|
||||
# Gültige Fills: BuySell muss 1 (BUY) oder 2 (SELL) sein, Qty > 0
|
||||
if f["buysell"] in (_BUY, _SELL) and f["qty"] > 0:
|
||||
fills.append(f)
|
||||
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}"
|
||||
|
||||
return {
|
||||
"trade_id": f"{account}_{symbol}_{ep:.4f}_{xp:.4f}_{entry_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
|
||||
Reference in New Issue
Block a user