Support other accounts of a login, e.g. a company account, via --account

README.md

@@ -14,6 +14,7 @@ __Table of Contents__
 * [Usage](#usage)
 * [Authentication](#authentication)
   * [Web login](#web-login)
+  * [Other accounts of your login](#other-accounts-of-your-login)
 * [Development](#development)
   * [Setting Up a Development Environment](#setting-up-a-development-environment)
   * [Linting and Code Formatting](#linting-and-code-formatting)
@@ -61,16 +62,18 @@ If you want to use the cutting-edge version, use this command instead:
 <!-- runcmd code:console uv run --python 3.13 pytr help --for-readme -->
 ```console
 usage: pytr [-h] [-V] [-v {warning,info,debug}] [--debug-logfile DEBUG_LOGFILE] [--debug-log-filter DEBUG_LOG_FILTER]
-            {help,login,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} ...
+            {help,login,accounts,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} ...
 
 Use "pytr command_name --help" to get detailed help to a specific command
 
 Commands:
-  {help,login,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion}
+  {help,login,accounts,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion}
                                         Desired action to perform
     help                                Print this help message
     login                               Check if credentials file exists. If not create it and ask for input. Try to
                                         login. Ask for device reset if needed
+    accounts                            List accounts of this login, e.g. a company accounts or child accounts. Use them
+                                        with --account
     portfolio                           Show current portfolio
     rates                               Fetch current prices for a list of ISINs given as direct list or CSV input
     details                             Get details for an ISIN
@@ -116,6 +119,22 @@ Web login uses the public web-login endpoints at `api.traderepublic.com`. Two va
 Both variants keep you logged in on your primary device, but you may need to re-authenticate every so often when
 running `pytr`.
 
+### Other accounts of your login
+
+A Trade Republic login can hold more than its own account, e.g. company accounts or child accounts. Trade Republic
+treats each of them as a customer of its own, and the web app switches between them after the login. Pass `--account`
+to do the same, on any subcommand that performs a login:
+
+```sh
+pytr portfolio --account LEGAL_ENTITY
+pytr dl_docs --account "Mustermann Holding GmbH" ./docs
+```
+
+The value is the account's type, its name or its customer id, as listed by `pytr accounts`. Without `--account`,
+pytr uses your own account. The choice is not remembered between runs.
+
+Unfortunately, it seems that this only works with company accounts, not with child accounts at the moment.
+
 ## Development
 
 ### Setting Up a Development Environment

pytr/account.py

@@ -4,9 +4,10 @@
 import time
 from getpass import getpass
 
+import requests
 from pygments import formatters, highlight, lexers
 
-from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi
+from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi, account_kind, account_name
 from .utils import get_logger
 
 
@@ -19,10 +20,34 @@ def get_settings(tr):
         return formatted_json
 
 
-def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False):
+def print_accounts(tr):
+    """List the accounts of this login, one per line: type, name, customer id and state.
+
+    `--account` takes the type, the name or, where those are not unique, the customer id.
+    """
+    rows = [
+        (
+            account_kind(r),
+            account_name(r),
+            str(r.get("customerId") or ""),
+            "your own account" if r.get("relationshipType") == "SELF" else str(r.get("accountState") or ""),
+        )
+        for r in tr.relationships()
+    ]
+    if not rows:
+        return ""
+    heading = ("TYPE", "NAME", "CUSTOMER ID", "STATE")
+    widths = [max(len(heading[i]), max(len(row[i]) for row in rows)) for i in range(4)]
+    all_rows = [heading] + rows
+    return "\n".join("  ".join(f"{cell:<{width}}" for cell, width in zip(row, widths)).rstrip() for row in all_rows)
+
+
+def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False, account=None):
     """
     Handle credentials parameters and store to credentials file if requested.
     If no parameters are set but are needed then ask for input
+
+    `account` selects another account of this login, e.g. a company account.
     """
     log = get_logger(__name__)
     save_cookies = True
@@ -89,4 +114,15 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh
         log.info("Logged in.")
 
     log.debug(get_settings(tr))
+
+    if account is not None:
+        try:
+            tr.switch_account(account)
+        except ValueError as e:
+            log.fatal(str(e))
+            sys.exit(1)
+        except requests.exceptions.HTTPError as e:
+            status = e.response.status_code if e.response is not None else "unknown"
+            log.fatal(f"Trade Republic refused to switch to account {account!r} (status {status}).")
+            sys.exit(1)
     return tr

pytr/api.py

@@ -32,6 +32,7 @@
 import time
 import urllib.parse
 import uuid
+from collections import deque
 from datetime import datetime
 from http.cookiejar import Cookie, MozillaCookieJar
 from typing import Any, Dict
@@ -87,6 +88,38 @@
 WEB_PLATFORM = "web-pro"
 
 
+# Seconds after which the web session is refreshed.
+SESSION_REFRESH_INTERVAL = 290
+
+
+def account_kind(relationship):
+    """The kind of account behind an entry of `relationships()`, e.g. `ADULT` or `LEGAL_ENTITY`."""
+    return str(relationship.get("accountType") or relationship.get("relationshipType") or "UNKNOWN")
+
+
+def account_name(relationship):
+    """The name of the account behind an entry of `relationships()`: the account's, else the person's."""
+    person = [str(part) for part in (relationship.get("firstName"), relationship.get("lastName")) if part]
+    return str(relationship.get("accountName") or " ".join(person))
+
+
+def _account_selectors(relationship):
+    """Everything an entry of `relationships()` can be selected by, lower-cased."""
+    selectors = [
+        relationship.get("customerId"),
+        relationship.get("relationshipType"),
+        relationship.get("accountType"),
+        relationship.get("firstName"),
+        account_name(relationship),
+    ]
+    return {str(selector).strip().casefold() for selector in selectors if selector and str(selector).strip()}
+
+
+def _describe_account(relationship):
+    name = account_name(relationship)
+    return f"{account_kind(relationship)} ({name})" if name else account_kind(relationship)
+
+
 class TradeRepublicApi:
     _default_headers = {
         "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36"
@@ -148,6 +181,10 @@ def __init__(
         if self._save_cookies:
             self._websession.cookies = MozillaCookieJar(self._cookies_file)
         self._sec_acc_no: str | None = None
+        # Customer id of the account the session acts for, once it is not the login's own.
+        self._subject_id: str | None = None
+        # Messages read off the websocket on behalf of a caller that has not asked for them yet.
+        self._held_back: deque = deque()
 
     def _fetch_waf_token_awswaf(self):
         """
@@ -516,11 +553,81 @@ def resume_websession(self):
 
     def _web_request(self, url_path, payload=None, method="GET"):
         if self._session_expires_at < time.time():
-            r = self._websession.get(f"{self._host}/api/v1/auth/web/session")
+            if self._subject_id is None:
+                # Refreshing without a subject returns the session to the login's own account.
+                # A new process starts here, so it never inherits the account an earlier run chose.
+                r = self._websession.get(f"{self._host}/api/v1/auth/web/session")
+            else:
+                r = self._post_session(self._subject_id)
             r.raise_for_status()
-            self._session_expires_at = time.time() + 290
+            self._session_expires_at = time.time() + SESSION_REFRESH_INTERVAL
         return self._websession.request(method=method, url=f"{self._host}{url_path}", data=payload)
 
+    def _post_session(self, subject_id):
+        """Ask for session cookies that act for `subject_id`, the way the web frontend does.
+
+        The frontend uses this call both to switch the account and to refresh the session.
+        """
+        return self._websession.post(
+            f"{self._host}/api/v2/auth/web/session",
+            json={"subjectId": subject_id},
+            headers=self._login_headers(),
+        )
+
+    def relationships(self):
+        """List the accounts this login can act for.
+
+        Besides the login's own account (`relationshipType` `SELF`) this is e.g. a
+        company account (`LEGAL_ENTITY_ACTOR`).
+        """
+        r = self._web_request("/api/v1/customer/relationships/detailed")
+        r.raise_for_status()
+        return r.json().get("relationships") or []
+
+    def _find_relationship(self, account):
+        """Pick one relationship by customer id, type or name. `None` picks the login's own account."""
+        relationships = self.relationships()
+        if account is None:
+            matches = [r for r in relationships if r.get("relationshipType") == "SELF"]
+            wanted = "Your own account"
+        else:
+            matches = [r for r in relationships if account.strip().casefold() in _account_selectors(r)]
+            wanted = repr(account)
+        if len(matches) == 1 and matches[0].get("customerId"):
+            return matches[0]
+        known = ", ".join(_describe_account(r) for r in relationships) or "none"
+        if len(matches) > 1:
+            raise ValueError(f"{wanted} matches more than one account, use its customer id. Accounts: {known}.")
+        raise ValueError(f"{wanted} was not found among the accounts of this login. Accounts: {known}.")
+
+    def switch_account(self, account=None):
+        """Let the session act for another account of this login, e.g. a company account.
+
+        `account` is the account's type (`LEGAL_ENTITY`), its name or its customer id, as listed
+        by `relationships()`. `None` switches back to the login's own account. Returns the chosen
+        relationship.
+
+        Trade Republic does not address these accounts by a second account number. It hands
+        out new session cookies that act for the other customer, so every call made afterwards,
+        the websocket included, sees that account only.
+        """
+        if self._ws is not None and self._ws.close_code is None:
+            raise ValueError("Close the websocket connection before switching the account.")
+
+        relationship = self._find_relationship(account)
+        r = self._post_session(relationship["customerId"])
+        r.raise_for_status()
+
+        is_self = relationship.get("relationshipType") == "SELF"
+        self._subject_id = None if is_self else relationship["customerId"]
+        self._session_expires_at = time.time() + SESSION_REFRESH_INTERVAL
+        # The securities account belongs to the previous account.
+        self._sec_acc_no = None
+        # The switch replaced the session cookies; keep the saved ones usable for the next run.
+        self.save_websession()
+        self.log.info(f"Using account: {_describe_account(relationship)}")
+        return relationship
+
     async def _get_ws(self):
         if self._ws and self._ws.close_code is None:
             return self._ws
@@ -591,6 +698,11 @@ async def unsubscribe(self, subscription_id):
         self._previous_responses.pop(subscription_id, None)
 
     async def recv(self):
+        if self._held_back:
+            return self._held_back.popleft()
+        return await self._recv_ws()
+
+    async def _recv_ws(self):
         ws = await self._get_ws()
         while True:
             response = await ws.recv()
@@ -669,11 +781,42 @@ async def portfolio_status(self):
 
     async def compact_portfolio(self):
         if self._sec_acc_no is None:
-            self.settings()
+            if self._subject_id is None:
+                self.settings()
+            else:
+                # /api/v2/auth/account answers 400 INVALID_AUTH_ACCOUNT_STATE for any account but
+                # the login's own, so the number has to come from the websocket instead.
+                self._sec_acc_no = await self._sec_acc_no_from_account_pairs()
         if self._sec_acc_no is None:
-            raise ValueError("Could not retrieve securities account number from account settings.")
+            source = "account settings" if self._subject_id is None else "account pairs"
+            raise ValueError(f"Could not retrieve securities account number from {source}.")
         return await self.subscribe({"type": "compactPortfolioByType", "secAccNo": self._sec_acc_no})
 
+    async def account_pairs(self):
+        return await self.subscribe({"type": "accountPairs"})
+
+    async def _sec_acc_no_from_account_pairs(self, timeout=5.0):
+        """Read the securities account number of the account the session acts for."""
+        subscription_id = await self.account_pairs()
+        try:
+            response = await asyncio.wait_for(self._recv_holding_back_others(subscription_id), timeout)
+        except asyncio.TimeoutError:
+            return None
+        finally:
+            await self.unsubscribe(subscription_id)
+        accounts = [a for a in response.get("accounts") or [] if a.get("securitiesAccountNumber")]
+        # Prefer the pair of the default product if the account has more than one.
+        accounts.sort(key=lambda a: a.get("productType") != "DEFAULT")
+        return accounts[0]["securitiesAccountNumber"] if accounts else None
+
+    async def _recv_holding_back_others(self, subscription_id):
+        """Wait for one subscription's answer; keep the answers to others for the next `recv()`."""
+        while True:
+            message = await self._recv_ws()
+            if message[0] == subscription_id:
+                return message[2]
+            self._held_back.append(message)
+
     async def watchlist(self):
         return await self.subscribe({"type": "watchlist"})