From 047a4e307dff1a26e2f0289bacf5a1383adab071 Mon Sep 17 00:00:00 2001 From: Simone Busoli Date: Wed, 12 Aug 2026 12:08:16 +0200 Subject: [PATCH 1/3] Send the commands Darwin accepts for portfolio and account info MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_portfolio() sent GETPORTFOLIO and get_account_info() sent GETACCTINFO. Darwin rejects both with ERR;;1004, so neither call could ever return data. Verified against Darwin 2.5.1: > GETPORTFOLIO ERR;GETPORTFOLIO;1004 > INFOSTOCKS STOCK;;;... (one line per position) > GETACCTINFO ERR;GETACCTINFO;1004 > INFOACCOUNT INFOACCOUNT;;;... The parsers were already written for the correct commands — the docstrings of parse_portfolio_response and parse_account_info_response name INFOSTOCKS and INFOACCOUNT respectively — so only the commands sent were wrong. Co-Authored-By: Claude Opus 5 (1M context) --- directa_api/trading.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/directa_api/trading.py b/directa_api/trading.py index 6f2a9d5..7cd1980 100644 --- a/directa_api/trading.py +++ b/directa_api/trading.py @@ -310,7 +310,10 @@ def get_portfolio(self, parse: bool = True) -> Union[Dict[str, Any], str]: if self.simulation_mode: return self.simulation.get_portfolio() - command = "GETPORTFOLIO" + # INFOSTOCKS, not GETPORTFOLIO: Darwin answers the latter with + # ERR;GETPORTFOLIO;1004 (verified against Darwin 2.5.1). Note that + # parse_portfolio_response already documents INFOSTOCKS as its input. + command = "INFOSTOCKS" response = self.send_command(command) if parse: @@ -330,7 +333,10 @@ def get_account_info(self, parse: bool = True) -> Union[Dict[str, Any], str]: if self.simulation_mode: return self.simulation.get_account_info() - command = "GETACCTINFO" + # INFOACCOUNT, not GETACCTINFO: Darwin answers the latter with + # ERR;GETACCTINFO;1004 (verified against Darwin 2.5.1). Note that + # parse_account_info_response already documents INFOACCOUNT as its input. + command = "INFOACCOUNT" response = self.send_command(command) if parse: From 4cb977de79b8b0923c747c8363fb300a191d5497 Mon Sep 17 00:00:00 2001 From: Simone Busoli Date: Wed, 12 Aug 2026 12:08:16 +0200 Subject: [PATCH 2/3] Read complete responses on the trading port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TradingConnection.send_command returned only the first line matching the expected prefix, discarding the rest of a multi-line response. Against a live account this meant ORDERLIST returned 4 orders on the socket and get_orders() returned 1, and a 15-position portfolio arrived as a single position. Nothing reported that data had been dropped. Two further problems fed into the same failure: - The read loop stopped at the first chunk containing a newline, so a list spanning several TCP reads was cut wherever the segment boundary fell. - Darwin pushes unsolicited traffic on this port: a full portfolio and order snapshot on connect, then position and order updates as they happen. Those lines were read as if they answered the command in flight, and pushed STOCK lines are indistinguishable from INFOSTOCKS rows. The response is now read line by line, with the leftover bytes buffered across calls, until the response is actually complete. Completion is determined by an end marker rather than by silence: connect() enables FLOWPOINT, so Darwin wraps list responses in BEGIN/END STOCKLIST and BEGIN/END ORDERLIST. If Darwin declines, the reader falls back to an idle gap and logs a warning. Lines are then selected by prefix, and a list command returns every matching row. The connect-time push is drained into TradingConnection.pushed_lines instead of being read as a response — it is a genuine snapshot, so it is kept rather than dropped. ERR codes 1024-1028 announce trading/datafeed link events and arrive unprompted; they no longer fail whichever command happens to be in flight. Co-Authored-By: Claude Opus 5 (1M context) --- directa_api/connection.py | 413 +++++++++++++++++++++++++------------- 1 file changed, 269 insertions(+), 144 deletions(-) diff --git a/directa_api/connection.py b/directa_api/connection.py index 34de07b..52a82c1 100644 --- a/directa_api/connection.py +++ b/directa_api/connection.py @@ -41,7 +41,9 @@ def __init__(self, host: str = "127.0.0.1", port: int = 10002, self.connected = False self.last_status = None self.connection_status = "UNKNOWN" - + # Whatever the service pushed before any command was sent + self.initial_data_text = "" + # Connection tracking self.connection_attempts = 0 self.last_connection_time = None @@ -128,7 +130,12 @@ def connect(self, initial_timeout: float = 5.0, normal_timeout: float = 2.0) -> if initial_data: initial_text = initial_data.decode('utf-8') self.logger.debug(f"Initial data received on connect: {initial_text.strip()}") - + + # Keep it: on the trading port this opening push carries a + # full portfolio and order snapshot, which subclasses expose + # rather than discard. + self.initial_data_text = initial_text + # Check for darwin status in initial data self._check_status_response(initial_text) except Exception as e: @@ -302,169 +309,287 @@ def get_connection_metrics(self) -> Dict[str, Any]: class TradingConnection(DirectaConnection): """Connection handler for Directa Trading API (port 10002)""" - + + # Response line prefix each command answers with. + RESPONSE_PREFIXES = { + "DARWINSTATUS": "DARWIN_STATUS", + "INFOACCOUNT": "INFOACCOUNT", + "INFOAVAILABILITY": "AVAILABILITY", + "INFOSTOCKS": "STOCK", + "ORDERLIST": "ORDER", + "ORDERLISTPENDING": "ORDER", + "GETPOSITION": "STOCK", + "FLOWPOINT": "FLOWPOINT", + } + + # Commands that legitimately answer with more than one line: one row per + # position or per order. Their responses must never be reduced to a + # single line. + LIST_COMMANDS = frozenset({"INFOSTOCKS", "ORDERLIST", "ORDERLISTPENDING"}) + + # Darwin closes a framed list with one of these when FLOWPOINT is enabled, + # which lets a list response terminate immediately instead of waiting for + # the socket to fall quiet. + LIST_TERMINATORS = ("END STOCKLIST", "END ORDERLIST") + + # ERR codes that announce a trading/datafeed link event rather than + # refusing the command in flight. They arrive unprompted, so attributing + # them to the current command turns an unrelated notice into a failure. + PUSH_NOTIFICATION_CODES = frozenset({"1024", "1025", "1026", "1027", "1028"}) + + # How long a list response may pause before it is considered finished, + # used only when BEGIN/END markers are not enabled. + IDLE_GAP = 0.4 + def __init__(self, host: str = "127.0.0.1", port: int = 10002, buffer_size: int = 4096): super().__init__(host, port, buffer_size, "DirectaTrading") - + # Trading-specific attributes self.is_trading_connected = False - + # Bytes read from the socket but not yet consumed as a complete line. + # Kept across calls so a response arriving in the same TCP segment as + # the next pushed update is not discarded. + self._read_buffer = b"" + # Whether Darwin agreed to frame list responses with BEGIN/END markers + self.flowpoint = False + # The portfolio and order snapshot Darwin pushes on connect + self.pushed_lines = [] + def set_connection_status(self, status: str, is_connected: bool) -> None: """Override to also update trading connection status""" super().set_connection_status(status, is_connected) self.is_trading_connected = is_connected + + def connect(self, initial_timeout: float = 5.0, normal_timeout: float = 2.0) -> bool: + """ + Connect, clear Darwin's unsolicited opening push, and enable framing + + On connect Darwin sends a greeting followed by a full portfolio and + order snapshot, unprompted. Left in the socket, those lines would be + read as the response to whatever command is sent first — and, worse, + pushed STOCK lines are indistinguishable from the rows of an INFOSTOCKS + response. They are collected into pushed_lines instead, and FLOWPOINT + is enabled so that list responses are delimited explicitly. + """ + if not super().connect(initial_timeout, normal_timeout): + return False + + self._read_buffer = b"" + self.pushed_lines = [ + line.strip() for line in self.initial_data_text.splitlines() if line.strip() + ] + self.pushed_lines.extend(self._drain_pushed_lines()) + self.flowpoint = self._enable_flowpoint() + if not self.flowpoint: + self.logger.warning( + "Darwin did not enable FLOWPOINT; list responses have no end " + "marker and will be read until the connection falls idle" + ) + return True + + def _drain_pushed_lines(self, idle: float = 1.0) -> list: + """Collect the lines Darwin pushed before any command was sent""" + lines = [] + while self._await_more_data(idle): + line = self._next_line(time.time() + idle) + if line is None: + break + if line: + lines.append(line) + return lines + + def _enable_flowpoint(self) -> bool: + """ + Ask Darwin to wrap list responses in BEGIN/END markers + + With framing on, a list response ends at its END marker rather than + whenever the socket happens to go quiet, and unsolicited lines arriving + mid-response fall outside the markers. + """ + try: + response = self.send_command("FLOWPOINT TRUE") + except (ConnectionError, socket.error) as e: + self.logger.warning(f"Could not enable FLOWPOINT: {str(e)}") + return False + return "FLOWPOINT;TRUE" in response.upper() def send_command(self, command: str, timeout: float = None) -> str: """ - Send a command to the Trading API with specialized handling - + Send a command to the Trading API and return its response + + The response is read line by line and only the lines belonging to this + command are returned. Two things make that necessary: + + - Darwin pushes unsolicited traffic on this port (a full portfolio and + order list on connect, then position and order updates as they + happen), so "whatever arrives next" is not necessarily the answer. + - A list command answers with one line per row, and every row matters. + Args: command: The command string to send - timeout: Optional override for socket timeout - + timeout: Optional override for how long to wait for the response + Returns: - str: The response from the server + str: The response lines for this command, newline-separated. List + commands return every matching row, not just the first. """ if not self.connected or not self.socket: raise ConnectionError("Not connected to Directa Trading API") - - # Ensure command ends with newline - if not command.endswith('\n'): - command += '\n' - + + command = command.strip() + name = command.split()[0] if command else "" + prefix = self.RESPONSE_PREFIXES.get(name, "") + expects_list = name in self.LIST_COMMANDS + + if timeout is None: + timeout = 5.0 if expects_list else 3.0 + try: - self.logger.debug(f"Sending command: {command.strip()}") - self.socket.sendall(command.encode('utf-8')) - - # Use different timeouts for different commands - if command.strip() == "DARWINSTATUS": - # Use longer timeout for status checks - self.socket.settimeout(3.0) - else: - # Use standard timeout for other commands - self.socket.settimeout(2.0 if timeout is None else timeout) - - response = b"" - - # Read response with a timeout loop - start_time = time.time() - max_time = 3.0 # Maximum 3 seconds - - while time.time() - start_time < max_time: - try: - chunk = self.socket.recv(self.buffer_size) - if not chunk: # If no data, break - if response: # But only if we already have some data - break - # Otherwise keep waiting a bit - time.sleep(0.1) - continue - - response += chunk - - # Special handling for DARWINSTATUS - if command.strip() == "DARWINSTATUS" and b"DARWIN_STATUS" in response: - # When we get a DARWIN_STATUS response, wait a bit more for complete data - time.sleep(0.1) - try: - # Try to get any additional data - self.socket.settimeout(0.2) - more_data = self.socket.recv(self.buffer_size) - if more_data: - response += more_data - except (socket.timeout, BlockingIOError): - pass # No more data available, which is fine - # Found our status response, can break - break - - # For other commands, if we see a complete response, stop waiting - if b'\n' in chunk: - # If we have a command-specific response, we can break - if (command.strip() == "INFOACCOUNT" and b"INFOACCOUNT" in response) or \ - (command.strip() == "INFOSTOCKS" and (b"STOCK" in response or b"ERR" in response)) or \ - (command.strip() == "ORDERLIST" and (b"ORDER" in response or b"ERR" in response)) or \ - (b"ERR" in response): # Always break on error - break - - # For other responses, wait a short time for any additional data - time.sleep(0.1) - try: - self.socket.settimeout(0.1) - more_data = self.socket.recv(self.buffer_size) - if more_data: - response += more_data - except (socket.timeout, BlockingIOError): - pass # No more data, which is fine - break - except socket.timeout: - # No data received within timeout - if response: # If we already have data, we can stop - break - - # Restore standard timeout - self.socket.settimeout(2.0) - - # If we didn't get any response, raise an error - if not response: - raise ConnectionError("No response received from server") - - response_text = response.decode('utf-8') - self.logger.debug(f"Received response: {response_text.strip()}") - - # Check for darwin status in the response - self._check_status_response(response_text) - - # Special handling for multi-line responses - lines = response_text.strip().split('\n') - if len(lines) > 1: - # Find the right response line based on the command - cmd_name = command.strip() - cmd_prefix = "" - - # Map commands to expected response prefixes - if cmd_name == "DARWINSTATUS": - cmd_prefix = "DARWIN_STATUS" - elif cmd_name == "INFOACCOUNT": - cmd_prefix = "INFOACCOUNT" - elif cmd_name == "INFOAVAILABILITY": - cmd_prefix = "AVAILABILITY" - elif cmd_name == "INFOSTOCKS": - cmd_prefix = "STOCK" - elif cmd_name == "ORDERLIST": - cmd_prefix = "ORDER" - - # Search for matching response line - for line in lines: - # Direct match with prefix - if line.startswith(cmd_prefix): - return line - # Check for contained prefix (e.g., DARWIN_STATUS in a larger line) - if cmd_prefix and cmd_prefix in line: - return line - # Always prioritize error responses - if line.startswith("ERR;"): - return line - - # Special cases for specific commands - if cmd_name == "DARWINSTATUS": - for line in lines: - if "DARWIN_STATUS" in line: - return line - - if cmd_name == "INFOAVAILABILITY": - for line in lines: - if line.startswith("AVAILABILITY"): - return line - - # If no specific match found, return the last non-empty line - for line in reversed(lines): - if line.strip(): - return line - - return response_text + self.logger.debug(f"Sending command: {command}") + self.socket.sendall((command + "\r\n").encode('utf-8')) + lines = self._read_response(prefix, expects_list, timeout) except socket.error as e: self.logger.error(f"Socket error: {str(e)}") raise + finally: + try: + self.socket.settimeout(2.0) + except (socket.error, AttributeError): + pass + + if not lines: + raise ConnectionError("No response received from server") + + response_text = "\n".join(lines) + self.logger.debug(f"Received response: {response_text}") + + # Check for darwin status in the response + self._check_status_response(response_text) + + return self._select_response(lines, prefix, expects_list) + + def _read_response(self, prefix: str, expects_list: bool, timeout: float) -> list: + """ + Read response lines until this command's response is complete + + Stops on an error line, on a BEGIN/END list terminator, on the expected + line for a single-line command, or — for a list without BEGIN/END + markers — once no further line arrives within IDLE_GAP. + """ + deadline = time.time() + timeout + lines = [] + while True: + line = self._next_line(deadline) + if line is None: + break + if not line: + continue + + if self._is_push_notification(line): + # A link notice, not this command's answer. Keep it in the + # response so _check_status_response sees it, but do not let it + # end the read. + lines.append(line) + continue + + lines.append(line) + + if line.startswith("ERR;"): + break + if any(line.startswith(marker) for marker in self.LIST_TERMINATORS): + break + if expects_list: + if self.flowpoint: + # Framed: only the END marker ends the list, so a pause + # mid-response cannot cut it short. + continue + if not self._await_more_data(self.IDLE_GAP): + break + continue + if not prefix or line.startswith(prefix) or prefix in line: + break + return lines + + def _next_line(self, deadline: float): + """ + Return the next complete line, or None once deadline passes + + Bytes beyond the newline stay in the buffer for the next read. + """ + while True: + newline = self._read_buffer.find(b'\n') + if newline >= 0: + raw = self._read_buffer[:newline] + self._read_buffer = self._read_buffer[newline + 1:] + return raw.decode('utf-8', errors='replace').strip() + + remaining = deadline - time.time() + if remaining <= 0: + return None + self.socket.settimeout(min(0.5, remaining)) + try: + chunk = self.socket.recv(self.buffer_size) + except socket.timeout: + continue + if not chunk: + raise ConnectionError("Connection closed by Directa Trading API") + self._read_buffer += chunk + + def _await_more_data(self, idle: float) -> bool: + """True if another line is available, or arrives within idle seconds""" + if b'\n' in self._read_buffer: + return True + deadline = time.time() + idle + while True: + remaining = deadline - time.time() + if remaining <= 0: + return b'\n' in self._read_buffer + self.socket.settimeout(remaining) + try: + chunk = self.socket.recv(self.buffer_size) + except socket.timeout: + return False + if not chunk: + return False + self._read_buffer += chunk + if b'\n' in self._read_buffer: + return True + + def _is_push_notification(self, line: str) -> bool: + """True for an ERR line announcing a link event rather than a refusal""" + if not line.startswith("ERR;"): + return False + parts = line.split(';') + return len(parts) > 2 and parts[2].strip() in self.PUSH_NOTIFICATION_CODES + + def _select_response(self, lines: list, prefix: str, expects_list: bool) -> str: + """ + Pick the lines that answer this command out of everything that was read + + For a list command this returns every matching row. Returning only the + first — as this method used to — silently turned a 15-position + portfolio into one position, with no error to show anything was lost. + """ + for line in lines: + if line.startswith("ERR;") and not self._is_push_notification(line): + return line + + candidates = [line for line in lines if not self._is_push_notification(line)] + if not candidates: + return "\n".join(lines) + + if not prefix: + return candidates[0] if len(candidates) == 1 else "\n".join(candidates) + + matching = [line for line in candidates if line.startswith(prefix)] + if not matching: + matching = [line for line in candidates if prefix in line] + if not matching: + return candidates[-1] + + if expects_list: + return "\n".join(matching) + return matching[0] class HistoricalConnection(DirectaConnection): From e4502fba22879d9420a55777157867dd9a657d08 Mon Sep 17 00:00:00 2001 From: Simone Busoli Date: Wed, 12 Aug 2026 12:08:16 +0200 Subject: [PATCH 3/3] Add tests for the trading connection Covers the reads against a stand-in for Darwin's socket, so they can run without the platform: complete multi-line responses, lines split across TCP reads, the connect-time push not being mistaken for a response, link notifications not failing a command, and the two command names Darwin accepts. Response shapes follow a live Darwin 2.5.1; values are synthetic. Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_trading_connection.py | 277 +++++++++++++++++++++++++++++++ 1 file changed, 277 insertions(+) create mode 100644 tests/test_trading_connection.py diff --git a/tests/test_trading_connection.py b/tests/test_trading_connection.py new file mode 100644 index 0000000..66709b2 --- /dev/null +++ b/tests/test_trading_connection.py @@ -0,0 +1,277 @@ +"""Tests for TradingConnection against a stand-in for Darwin's socket. + +The response shapes follow what a live Darwin 2.5.1 actually sends; the values +are synthetic. The behaviours pinned down are the ones that made real use fail: +a multi-line response reduced to one line, the two commands Darwin rejects, and +the unsolicited traffic Darwin pushes on this port. +""" + +import socket +import threading +import time + +import pytest + +from directa_api import DirectaTrading + +STATUS = ( + "DARWIN_STATUS;CONN_OK;FALSE;Release 2.5.1 build 04/02/2025 11:00:00 " + "more info at http://app1.directatrading.com/trading-api-directa/index.html" +) +ACCOUNT = "INFOACCOUNT;10:00:00;A0000;1500.0;2500;100000.0;104000.0;PROD" + +# Fifteen positions, because the bug being pinned reduced any list to one row. +STOCKS = [ + "STOCK;ENI.MI;10:00:00;100;0;0;13.5;120", + "STOCK;ISP.MI;10:00:00;500;0;0;3.2;45", + "STOCK;UCG.MI;10:00:00;200;0;0;35.4;-80", + "STOCK;STLAM.MI;10:00:00;300;0;0;8.75;210", + "STOCK;ENEL.MI;10:00:00;400;0;0;6.1;90", + "STOCK;RACE.MI;10:00:00;10;0;0;380.0;150", + "STOCK;G.MI;10:00:00;250;0;0;24.8;-35", + "STOCK;PST.MI;10:00:00;150;0;0;12.4;60", + "STOCK;MB.MI;10:00:00;180;0;0;14.9;25", + "STOCK;BAMI.MI;10:00:00;900;0;0;7.3;110", + "STOCK;AZM.MI;10:00:00;120;0;0;22.6;-15", + "STOCK;CPR.MI;10:00:00;220;0;0;9.85;40", + "STOCK;MONC.MI;10:00:00;60;0;0;52.3;-70", + "STOCK;SRG.MI;10:00:00;700;0;0;4.55;85", + "STOCK;TIT.MI;10:00:00;5000;0;-5000;0.28;-20", +] + +# One symbol accumulates several order records with different states. +ORDERS = [ + "ORDER;ENI.MI;09:58:54;ORD1;VENAZ;13.9;0.0;100;2004", + "ORDER;ENI.MI;09:59:26;ORD2;VENAZ;13.85;0.0;100;2000", + "ORDER;ENI.MI;09:58:38;ORD3;VENAZ;13.95;0.0;100;2004", + "ORDER;ENI.MI;09:59:26;ORD4;VENAZ;13.8;0.0;100;2004", +] + + +class FakeDarwin: + """Serves one connection on an ephemeral port, imitating Darwin's dAPI.""" + + def __init__(self, responses=None, push=(), chunk_size=None, flowpoint=True): + self.responses = dict(responses or {}) + self.push = list(push) + self.chunk_size = chunk_size + self.flowpoint = flowpoint + self.received = [] + self.inject_before_next = [] + self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._listener.bind(("127.0.0.1", 0)) + self._listener.listen(1) + self.port = self._listener.getsockname()[1] + self._stopping = threading.Event() + self._thread = threading.Thread(target=self._serve) + self._thread.daemon = True + + def __enter__(self): + self._thread.start() + return self + + def __exit__(self, *exc_info): + self.stop() + + def stop(self): + self._stopping.set() + try: + self._listener.close() + except OSError: + pass + if self._thread.is_alive(): + self._thread.join(timeout=2.0) + + def _write(self, conn, lines): + data = "".join("%s\r\n" % line for line in lines).encode("latin-1") + if not self.chunk_size: + conn.sendall(data) + return + for start in range(0, len(data), self.chunk_size): + conn.sendall(data[start:start + self.chunk_size]) + time.sleep(0.002) + + def _serve(self): + try: + conn, _ = self._listener.accept() + except OSError: + return + try: + if self.push: + self._write(conn, self.push) + buffer = b"" + while not self._stopping.is_set(): + try: + chunk = conn.recv(4096) + except OSError: + return + if not chunk: + return + buffer += chunk + while b"\n" in buffer: + raw, buffer = buffer.split(b"\n", 1) + command = raw.decode("latin-1").strip() + if not command: + continue + self.received.append(command) + if self.inject_before_next: + self._write(conn, self.inject_before_next) + self.inject_before_next = [] + self._write(conn, self._answer(command)) + finally: + conn.close() + + def _answer(self, command): + if command == "FLOWPOINT TRUE": + return ["FLOWPOINT;TRUE"] if self.flowpoint else ["ERR;FLOWPOINT;1004"] + response = self.responses.get(command) + if response is None: + return ["ERR;%s;1003" % command.split()[0]] + return response + + +def framed(begin, end, rows, flowpoint=True): + """A list response, with or without BEGIN/END markers.""" + if not flowpoint: + return list(rows) + return [begin] + list(rows) + [end] + + +def responses(flowpoint=True): + return { + "DARWINSTATUS": [STATUS], + "INFOACCOUNT": [ACCOUNT], + "INFOSTOCKS": framed("BEGIN STOCKLIST", "END STOCKLIST", STOCKS, flowpoint), + "ORDERLIST": framed("BEGIN ORDERLIST", "END ORDERLIST", ORDERS, flowpoint), + } + + +@pytest.fixture +def darwin_factory(): + """Yields a factory so each test can shape the fake, and always cleans up.""" + servers = [] + + def make(**kwargs): + server = FakeDarwin(**kwargs) + server.__enter__() + servers.append(server) + return server + + yield make + for server in servers: + server.stop() + + +def trading(server): + return DirectaTrading(host="127.0.0.1", port=server.port, simulation_mode=False) + + +class TestCommandNames: + """Darwin rejects GETPORTFOLIO and GETACCTINFO with ERR 1004.""" + + def test_portfolio_uses_infostocks(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS]) + with trading(server) as api: + api.get_portfolio() + assert "INFOSTOCKS" in server.received + assert "GETPORTFOLIO" not in server.received + + def test_account_info_uses_infoaccount(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS]) + with trading(server) as api: + api.get_account_info() + assert "INFOACCOUNT" in server.received + assert "GETACCTINFO" not in server.received + + +class TestMultiLineResponses: + """The regression: a list response used to come back as a single line.""" + + def test_portfolio_returns_every_position(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS]) + with trading(server) as api: + result = api.get_portfolio() + assert result["success"] is True + positions = result["data"] + assert len(positions) == 15 + assert [p["symbol"] for p in positions] == [s.split(";")[1] for s in STOCKS] + + def test_order_list_returns_every_order(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS]) + with trading(server) as api: + result = api.get_orders() + assert result["success"] is True + assert len(result["data"]) == 4 + + def test_works_without_flowpoint_markers(self, darwin_factory): + """If Darwin refuses FLOWPOINT, the list is read until it falls idle — + still complete, just less deterministic.""" + server = darwin_factory( + responses=responses(flowpoint=False), push=[STATUS], flowpoint=False + ) + with trading(server) as api: + result = api.get_portfolio() + assert len(result["data"]) == 15 + + def test_lines_split_across_reads_are_reassembled(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS], chunk_size=4) + with trading(server) as api: + positions = api.get_portfolio() + orders = api.get_orders() + assert len(positions["data"]) == 15 + assert len(orders["data"]) == 4 + + +class TestUnsolicitedTraffic: + """Darwin pushes a portfolio snapshot on connect and updates thereafter.""" + + def test_connect_push_is_not_read_as_a_response(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS] + STOCKS + ORDERS) + with trading(server) as api: + info = api.get_account_info() + assert info["success"] is True + assert info["data"]["account_code"] == "A0000" + + def test_connect_push_is_kept_for_inspection(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS] + STOCKS + ORDERS) + with trading(server) as api: + pushed = api.connection.pushed_lines + assert any(line.startswith("STOCK;") for line in pushed) + + def test_pushed_update_before_response_is_skipped(self, darwin_factory): + server = darwin_factory(responses=responses(), push=[STATUS]) + with trading(server) as api: + server.inject_before_next = ["STOCK;ENI.MI;10:05:00;100;0;0;13.5;135"] + info = api.get_account_info() + assert info["data"]["account_code"] == "A0000" + + def test_link_notification_does_not_fail_the_command(self, darwin_factory): + """ERR 1027 reports a datafeed drop; it is not a refusal of this + command, and must not be returned in its place.""" + server = darwin_factory(responses=responses(), push=[STATUS]) + with trading(server) as api: + server.inject_before_next = ["ERR;DATAFEED;1027"] + info = api.get_account_info() + assert info["success"] is True + assert info["data"]["equity"] == 104000.0 + + +class TestErrors: + def test_refusal_is_reported(self, darwin_factory): + overrides = responses() + overrides["INFOACCOUNT"] = ["ERR;INFOACCOUNT;1004"] + server = darwin_factory(responses=overrides, push=[STATUS]) + with trading(server) as api: + info = api.get_account_info() + assert info["success"] is False + assert info["error_code"] == "1004" + + def test_empty_portfolio_is_reported_as_such(self, darwin_factory): + overrides = responses() + overrides["INFOSTOCKS"] = ["ERR;INFOSTOCKS;1018"] + server = darwin_factory(responses=overrides, push=[STATUS]) + with trading(server) as api: + result = api.get_portfolio() + assert result["success"] is False + assert result["error_code"] == "1018"