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): 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: 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"