Discover accounts and own scope

Account discovery answers which accounts the authenticated API session can see. Application authorization answers which of those accounts this process is allowed to use. Keep those decisions separate: an account appearing in IB.managedAccounts() is not an instruction to read it, trade it, or silently add it to an existing workload.

Complete connection and readiness first. The examples below use a paper port, a read-only API connection, placeholder paper-account IDs, and no order operation.

Minimal discovery, positions, and account P&L

This runnable script uses a dedicated one-account paper worker, validates the completed global position cache, waits for the first finite account-P&L update, and cancels both continuing streams before disconnecting.

import asyncio
from math import isfinite

from ib_async import Client, IB, PnL
from ib_async.util import UNSET_DOUBLE


async def main() -> None:
    selected_account = "PAPER_ACCOUNT_A"  # Placeholder; load the real ID from config.
    ib = IB()
    pnl_started = False
    pnl_ready = asyncio.Event()

    def on_pnl(value: PnL) -> None:
        numbers = (value.dailyPnL, value.unrealizedPnL, value.realizedPnL)
        if (
            value.account == selected_account
            and value.modelCode == ""
            and all(isfinite(number) and number != UNSET_DOUBLE for number in numbers)
        ):
            pnl_ready.set()

    ib.pnlEvent += on_pnl
    try:
        await ib.connectAsync(
            host="127.0.0.1",
            port=4002,
            clientId=8,
            timeout=10,
            readonly=True,
            raiseSyncErrors=True,
        )
        accessible = ib.managedAccounts()
        if frozenset(accessible) != frozenset({selected_account}):
            raise PermissionError("paper worker requires one exact account")

        summary = await ib.accountSummaryAsync(selected_account)
        net_liquidation = [
            value
            for value in summary
            if value.tag == "NetLiquidation" and value.currency
        ]
        if not net_liquidation:
            raise RuntimeError("required summary value is unavailable")

        positions = ib.positions()  # Global cache after successful startup sync.
        if any(position.account != selected_account for position in positions):
            raise PermissionError("position scope differs from authorization")

        ib.reqPnL(selected_account, "")
        pnl_started = True
        await asyncio.wait_for(pnl_ready.wait(), timeout=10)
    finally:
        ib.pnlEvent -= on_pnl
        try:
            if pnl_started:
                ib.cancelPnL(selected_account, "")
        finally:
            try:
                if ib.isConnected():
                    client: Client = ib.client
                    client.cancelPositions()
            finally:
                ib.disconnect()


asyncio.run(main())

Do not put account IDs in source code. The placeholder makes the scope decision visible; production configuration should supply the authorized IDs from an access-controlled source.

Three account sets

Treat scope as three sets with different owners:

Set	Source	Owner	Required interpretation
Accessible	IB.managedAccounts() after a successful connection	Authenticated IBKR session and its permissions	Accounts visible to this API session now.
Authorized	Application configuration/policy	Your control plane	Accounts this process may use. It must not expand automatically.
Synchronized	IB.accountValues(), IB.portfolio(), IB.positions(), and summary caches	Current IB connection epoch	State actually received into memory; it may be a subset of accessible accounts.

Application readiness should require the intended relationship among all three. For a dedicated worker, exact equality between accessible and authorized accounts is a useful fail-closed policy. A shared worker may instead permit extra accessible accounts, but must still filter every downstream operation through an explicit authorization set.

IB.managedAccounts() returns a copy of the wrapper's account list. In the TWS protocol, the managed-accounts callback occurs automatically during initial API connection and supplies the accounts available to the logged-in user. Re-evaluate the set after every reconnect; do not carry a prior epoch's discovery decision forward.

What startup synchronizes

With pinned ib_async 2.1.0 and the default startup fields, connectAsync behaves as follows:

Condition	Startup behavior	Scope consequence
Exactly one accessible account and no account= argument	The library selects that account for traditional account updates.	Convenience, not authorization. Validate it anyway.
Explicit account= and StartupFetch.ACCOUNT_UPDATES	Traditional account/portfolio updates are requested for that account.	Supplies accountValues() and portfolio state for the selected account.
At most IB.MaxSyncedSubAccounts accessible accounts and StartupFetch.SUB_ACCOUNT_UPDATES	One account-updates-multi request starts for each accessible account.	The default limit is 50; callbacks for different accounts can interleave.
More than IB.MaxSyncedSubAccounts accessible accounts	Automatic per-subaccount updates are skipped.	A successful connection does not mean all accessible account values are synchronized.

readonly=True skips startup order synchronization; it does not disable account or position reads. fetchFields controls which startup groups run, so the application must know the connection owner's chosen flags before interpreting an empty cache.

Cached account values versus account summary

These surfaces are related but not interchangeable:

Surface	Pinned 2.1.0 lifecycle	Important boundary
IB.accountValues(account="")	Reads the wrapper's synchronized traditional and multi-account-update cache immediately; a non-empty account filters that cache.	An empty result can mean no matching values were received, startup omitted the stream, or the scope is wrong. It is not proof of a zero balance.
IB.accountSummaryAsync(account="")	On the first call in a connection epoch, starts the library's fixed group="All" summary request and awaits its initial end callback; later calls read the summary cache.	The account argument filters locally after the all-account request. The summary remains a subscription.
IB.accountSummary(account="")	Synchronous wrapper around accountSummaryAsync.	Use the async form inside an asyncio application.
IB.reqAccountSummaryAsync()	Starts the fixed all-account, fixed-tag request used by accountSummaryAsync.	Prefer accountSummaryAsync; repeated low-level requests consume the protocol's limited summary-subscription capacity.
IB.reqAccountUpdatesMultiAsync(account, modelCode="")	Starts an account/model-scoped stream, waits for its initial end callback, and stores later values in accountValues().	Await completion is not stream cancellation. The high-level helper does not return its internal request ID.

The official account-summary contract allows only two active summary subscriptions. It sends the initial requested values, then changed values on the TWS Account Window's fixed three-minute cadence. Values are strings, availability depends on tag/account configuration, and currency is part of the identity; never collapse (tag, currency) into one unqualified number.

For introducing-broker structures with more than 50 subaccounts or on-demand account lookup, the official API does not allow reqAccountSummary with group="All". reqAccountUpdatesMulti also cannot use account="All" for an introducing broker with more than 50 subaccounts. Pinned IB.accountSummaryAsync() hard-codes group="All", so this helper is not a universal large-account fallback.

Production-pattern exact-scope gate

This pattern makes one connection owner responsible for discovery, scope authorization, summary readiness, and cleanup. It reports only mismatch counts so account identifiers do not leak into ordinary logs.

import asyncio

from ib_async import IB

EXPECTED_ACCOUNTS = frozenset({"PAPER_ACCOUNT_A", "PAPER_ACCOUNT_B"})


def require_exact_account_scope(
    accessible: list[str], expected: frozenset[str]
) -> tuple[str, ...]:
    actual = frozenset(accessible)
    missing = expected - actual
    unexpected = actual - expected
    if missing or unexpected:
        raise ValueError(
            "account scope mismatch: "
            f"missing={len(missing)}, unexpected={len(unexpected)}"
        )
    return tuple(sorted(actual))


async def main() -> None:
    ib = IB()
    try:
        await ib.connectAsync(
            host="127.0.0.1",
            port=4002,
            clientId=8,
            timeout=10,
            readonly=True,
            raiseSyncErrors=True,
        )
        scoped_accounts = require_exact_account_scope(
            ib.managedAccounts(), EXPECTED_ACCOUNTS
        )
        summary = await ib.accountSummaryAsync()

        for account in scoped_accounts:
            account_summary = [value for value in summary if value.account == account]
            if not any(value.tag == "NetLiquidation" for value in account_summary):
                raise RuntimeError("required summary tag is unavailable")

            # This is synchronized memory owned by the current connection epoch.
            cached_values = ib.accountValues(account)
            # Publish readiness only after application-specific tag/currency checks.
            _ = cached_values
    finally:
        ib.disconnect()


if __name__ == "__main__":
    asyncio.run(main())

If the application intentionally authorizes only a subset of a shared session, change the gate to require expected <= accessible, then pass the authorized set explicitly into every account, position, PnL, and order operation. Never use an empty account argument as an authorization shortcut.

Positions and P&L are separate subscriptions

Positions, account P&L, and single-position P&L do not share one completion contract:

Surface	Initial state and updates	Cancellation boundary
IB.reqPositionsAsync() / IB.reqPositions()	Requests every position in every accessible account. positionEnd completes the initial list, but later position callbacks continue to update IB.positions() and emit IB.positionEvent. A zero-size callback removes that contract from the cache while still emitting the zero observation.	The protocol pair is Client.cancelPositions(); the owner uses that exact client call or ends the connection epoch. connectAsync 2.1.0 already starts and awaits this global subscription, so do not immediately duplicate it after connection.
IB.reqPnL(account, modelCode) / IB.cancelPnL(account, modelCode)	Returns a mutable PnL object immediately. It is not ready until a matching callback supplies finite daily, unrealized, and realized values; later callbacks mutate the same object and emit IB.pnlEvent.	The exact (account, modelCode) key owns one request ID. A duplicate request asserts; cancellation removes both pinned lookup maps.
IB.reqPnLSingle(account, modelCode, conId) / IB.cancelPnLSingle(...)	Returns a mutable PnLSingle immediately. A valid callback also supplies position and market value and emits IB.pnlSingleEvent. An invalid contract ID can produce no callback.	The exact (account, modelCode, conId) key owns one request ID. Preserve an empty model code as a real part of that key.

The whole-account position stream is broader than an application account allowlist. Validate every callback before using it and never interpret an empty filtered cache as proof of zero positions. It is unavailable to introducing-broker or advisor master accounts with more than 50 subaccounts, and to broker accounts using on-demand lookup; design per-subaccount reqPositionsMulti ownership instead of treating missing global data as an empty portfolio.

Account P&L and position P&L are calculated on the TWS Portfolio Window basis and follow its configured reset schedule; they are operational projections, not a durable accounting ledger. The official single-position stream is approximately once per second (subject to change) and cannot use account="All" for an introducing-broker account configured for on-demand lookup.

Readiness must be per connection epoch and per exact subscription key. Zero is a valid P&L, position, or value; NaN, infinity, and the pinned protocol's unset-double sentinel are not ready. Persist the authorized scope, model code, contract ID, connection epoch, an aware application observation time, the configured P&L reset-window identity, and each accepted observation without placing account identifiers in ordinary logs. These callbacks carry no authoritative server observation timestamp: process duplicates idempotently by logical key, retain receipt order, and start a new reset-window/epoch record when the TWS reset configuration changes rather than comparing daily P&L across windows.

Production-pattern position and P&L owner

This owner accepts only a completed, global position receipt for the current exact account scope, attaches its own continuing position and disconnect handlers, owns exact P&L keys, and fails closed after any durable-write error. Application P&L event handlers call observe_account or observe_single; shutdown cancels every exact key and detaches every handler it started.

import asyncio
from collections.abc import Callable
from datetime import datetime, timezone
from math import isfinite
from typing import Any, NamedTuple

from ib_async import Client, IB, PnL, PnLSingle, Position
from ib_async.util import UNSET_DOUBLE


Record = dict[str, Any]


class PositionSnapshotReceipt(NamedTuple):
    connection_epoch: str
    accessible_accounts: frozenset[str]
    positions: tuple[Position, ...]
    completed_at: datetime
    proof: object


async def request_positions_snapshot(
    ib: IB,
    timeout: float,
    connection_epoch: str,
    authorized_accounts: frozenset[str],
    owner_proof: object,
    cleanup_on_failure: bool = True,
) -> PositionSnapshotReceipt:
    """Use only when the connection owner intentionally restarts this stream."""
    if (
        not isfinite(timeout)
        or timeout <= 0
        or not connection_epoch
        or not authorized_accounts
        or owner_proof is None
    ):
        raise ValueError("position deadline, epoch, and scope are required")
    typed_ib: IB = ib
    future: asyncio.Future[list[Position]] | None = None
    try:
        future = asyncio.ensure_future(typed_ib.reqPositionsAsync())
        positions = await asyncio.wait_for(asyncio.shield(future), timeout)
        accessible = frozenset(typed_ib.managedAccounts())
        if accessible != authorized_accounts:
            raise PermissionError("global position scope differs from authorization")
        return PositionSnapshotReceipt(
            connection_epoch=connection_epoch,
            accessible_accounts=accessible,
            positions=tuple(positions),
            completed_at=datetime.now(timezone.utc),
            proof=owner_proof,
        )
    except BaseException:
        if future is not None:
            future.cancel()
        if cleanup_on_failure:
            client: Client = typed_ib.client
            try:
                client.cancelPositions()
            except BaseException:
                pass
            finally:
                typed_ib.disconnect()
        raise


class PositionPnlOwner:
    def __init__(
        self,
        ib: IB,
        authorized_accounts: frozenset[str],
        connection_epoch: str,
        pnl_reset_window: str,
        persist: Callable[[Record], None],
        clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
    ) -> None:
        if not authorized_accounts or not connection_epoch or not pnl_reset_window:
            raise ValueError("accounts, epoch, and P&L reset window are required")
        self.ib = ib
        self.authorized_accounts = authorized_accounts
        self.connection_epoch = connection_epoch
        self.pnl_reset_window = pnl_reset_window
        self.persist = persist
        self.clock = clock
        self.account_pnl: dict[tuple[str, str], PnL] = {}
        self.single_pnl: dict[tuple[str, str, int], PnLSingle] = {}
        self.ready_account_keys: set[tuple[str, str]] = set()
        self.ready_single_keys: set[tuple[str, str, int]] = set()
        self.positions_ready = False
        self.positions_owned = False
        self.positions_pending = False
        self._position_proof = object()
        self._position_buffer: list[Record] | None = None
        self._position_start_task: asyncio.Task[Any] | None = None
        self.poisoned = False
        self.closed = False
        self.ib.positionEvent += self.observe_position
        self.ib.disconnectedEvent += self._on_disconnected
        self.events_attached = True

    def _require_healthy(self) -> None:
        if self.poisoned or self.closed:
            raise RuntimeError("position/P&L owner is poisoned; replace the epoch")

    def _observed_at(self) -> str:
        value = self.clock()
        if value.tzinfo is None or value.utcoffset() is None:
            raise ValueError("observation clock must return an aware datetime")
        return value.isoformat()

    def _clear_readiness(self) -> None:
        self.positions_ready = False
        self.ready_account_keys.clear()
        self.ready_single_keys.clear()

    def _detach_events(self) -> None:
        if not self.events_attached:
            return
        self.ib.positionEvent -= self.observe_position
        self.ib.disconnectedEvent -= self._on_disconnected
        self.events_attached = False

    def _require_account(self, account: str) -> None:
        if account not in self.authorized_accounts:
            raise PermissionError("account is outside authorized scope")

    def _write(self, record: Record) -> None:
        try:
            self.persist(record)
        except BaseException as error:
            if self.positions_pending:
                self._position_buffer = None
                self._position_start_failed(error)
            else:
                self.poisoned = True
                self._clear_readiness()
                self._detach_events()
            raise

    @staticmethod
    def _finite(*values: float) -> bool:
        return all(
            isfinite(float(value)) and float(value) != UNSET_DOUBLE
            for value in values
        )

    def _position_record(self, position: Position, observed_at: str) -> Record:
        self._require_account(position.account)
        if position.contract.conId <= 0 or not self._finite(
            position.position, position.avgCost
        ):
            raise ValueError("position identity or numeric value is invalid")
        return {
            "account": position.account,
            "conId": position.contract.conId,
            "position": float(position.position),
            "avgCost": float(position.avgCost),
            "zeroTombstone": float(position.position) == 0,
            "observedAt": observed_at,
        }

    def _adopt_positions(self, receipt: PositionSnapshotReceipt) -> None:
        self._require_healthy()
        if (
            receipt.connection_epoch != self.connection_epoch
            or receipt.accessible_accounts != self.authorized_accounts
            or receipt.proof is not self._position_proof
            or receipt.completed_at.tzinfo is None
            or receipt.completed_at.utcoffset() is None
        ):
            raise ValueError("position receipt is incomplete or for another scope")
        observed_at = receipt.completed_at.isoformat()
        observations = [
            self._position_record(position, observed_at)
            for position in receipt.positions
        ]
        self._write(
            {
                "kind": "position-snapshot-ready",
                "connectionEpoch": self.connection_epoch,
                "authorizedAccounts": sorted(self.authorized_accounts),
                "completedAt": observed_at,
                "positions": observations,
            }
        )
        self._require_healthy()
        self.positions_owned = True
        self.positions_ready = True

    def _position_start_failed(self, error: BaseException) -> None:
        pending_task = self._position_start_task
        if pending_task is not None:
            try:
                current_task = asyncio.current_task()
            except RuntimeError:
                current_task = None
            if pending_task is not current_task:
                pending_task.cancel()
        self.poisoned = True
        self.closed = True
        self.positions_pending = False
        self.positions_owned = False
        self.account_pnl.clear()
        self.single_pnl.clear()
        self._clear_readiness()
        self._detach_events()
        cleanup = "cancelled"
        client: Client = self.ib.client
        try:
            client.cancelPositions()
        except BaseException:
            cleanup = "cancel-failed"
        epoch_recycled = False
        try:
            self.ib.disconnect()
            epoch_recycled = True
        except BaseException:
            pass
        recovery: Record = {
            "kind": "position-start-failed",
            "connectionEpoch": self.connection_epoch,
            "errorType": type(error).__name__,
            "cleanup": cleanup,
            "epochRecycled": epoch_recycled,
        }
        try:
            recovery["observedAt"] = self._observed_at()
            self.persist(recovery)
        except BaseException:
            pass

    async def start_positions(self, timeout: float) -> None:
        self._require_healthy()
        if self.positions_owned or self._position_buffer is not None:
            raise RuntimeError("global position stream is already owned")
        self._position_buffer = []
        self.positions_pending = True
        self._position_start_task = asyncio.current_task()
        try:
            receipt = await request_positions_snapshot(
                self.ib,
                timeout,
                self.connection_epoch,
                self.authorized_accounts,
                self._position_proof,
                cleanup_on_failure=False,
            )
            buffered = self._position_buffer
            self._position_buffer = None
            self._adopt_positions(receipt)
            for record in buffered:
                self._write(
                    {
                        "kind": "position-observation",
                        "connectionEpoch": self.connection_epoch,
                        **record,
                    }
                )
                self._require_healthy()
            self.positions_pending = False
        except BaseException as error:
            self._position_buffer = None
            if self.positions_pending:
                self._position_start_failed(error)
            raise
        finally:
            self._position_start_task = None

    def observe_position(self, position: Position) -> None:
        if self.closed:
            return
        self._require_healthy()
        try:
            record = self._position_record(position, self._observed_at())
        except BaseException as error:
            if self.positions_pending:
                self._position_buffer = None
                self._position_start_failed(error)
            else:
                self.poisoned = True
                self._clear_readiness()
                self._detach_events()
            raise
        if self._position_buffer is not None:
            self._position_buffer.append(record)
            return
        self._write(
            {
                "kind": "position-observation",
                "connectionEpoch": self.connection_epoch,
                **record,
            }
        )

    def _on_disconnected(self) -> None:
        if self.closed:
            return
        self.poisoned = True
        self.positions_owned = False
        self.account_pnl.clear()
        self.single_pnl.clear()
        self._clear_readiness()
        try:
            try:
                self.persist(
                    {
                        "kind": "position-pnl-owner-disconnected",
                        "connectionEpoch": self.connection_epoch,
                        "observedAt": self._observed_at(),
                    }
                )
            except BaseException:
                pass
        finally:
            self._detach_events()

    def _dispatch_failed(
        self,
        scope: str,
        key: tuple[str, str] | tuple[str, str, int],
        error: BaseException,
    ) -> None:
        self.poisoned = True
        self._clear_readiness()
        self._detach_events()
        self.closed = True
        cleanup = "not-installed"
        try:
            if scope == "account":
                account_key = (key[0], key[1])
                if account_key in self.ib.wrapper.pnlKey2ReqId:
                    cleanup = "cancelled"
                    self.ib.cancelPnL(account_key[0], account_key[1])
                self.account_pnl.pop(account_key, None)
            else:
                single_key = (key[0], key[1], int(key[2]))
                if single_key in self.ib.wrapper.pnlSingleKey2ReqId:
                    cleanup = "cancelled"
                    self.ib.cancelPnLSingle(
                        single_key[0], single_key[1], single_key[2]
                    )
                self.single_pnl.pop(single_key, None)
        except BaseException:
            cleanup = "cancel-failed"
        finally:
            self.ib.disconnect()
        try:
            self.persist(
                {
                    "kind": "pnl-subscription-dispatch-failed",
                    "connectionEpoch": self.connection_epoch,
                    "scope": scope,
                    "ownerKey": list(key),
                    "errorType": type(error).__name__,
                    "cleanup": cleanup,
                    "epochRecycled": True,
                    "observedAt": self._observed_at(),
                }
            )
        except BaseException:
            pass

    def start_account(self, account: str, model_code: str = "") -> PnL:
        self._require_healthy()
        self._require_account(account)
        key = (account, model_code)
        if key in self.account_pnl:
            raise RuntimeError("duplicate account P&L owner key")
        self._write(
            {
                "kind": "pnl-subscription-starting",
                "connectionEpoch": self.connection_epoch,
                "scope": "account",
                "account": account,
                "modelCode": model_code,
                "pnlResetWindow": self.pnl_reset_window,
                "observedAt": self._observed_at(),
            }
        )
        ib: IB = self.ib
        try:
            value = ib.reqPnL(account, model_code)
        except BaseException as error:
            self._dispatch_failed("account", key, error)
            raise
        self.account_pnl[key] = value
        return value

    def start_single(
        self, account: str, model_code: str, con_id: int
    ) -> PnLSingle:
        self._require_healthy()
        self._require_account(account)
        if con_id <= 0:
            raise ValueError("a qualified positive contract ID is required")
        key = (account, model_code, con_id)
        if key in self.single_pnl:
            raise RuntimeError("duplicate single-position P&L owner key")
        self._write(
            {
                "kind": "pnl-subscription-starting",
                "connectionEpoch": self.connection_epoch,
                "scope": "single-position",
                "account": account,
                "modelCode": model_code,
                "conId": con_id,
                "pnlResetWindow": self.pnl_reset_window,
                "observedAt": self._observed_at(),
            }
        )
        ib: IB = self.ib
        try:
            value = ib.reqPnLSingle(account, model_code, con_id)
        except BaseException as error:
            self._dispatch_failed("single-position", key, error)
            raise
        self.single_pnl[key] = value
        return value

    def observe_account(self, value: PnL) -> None:
        self._require_healthy()
        key = (value.account, value.modelCode)
        if self.account_pnl.get(key) is not value:
            raise RuntimeError("unowned account P&L callback")
        if not self._finite(value.dailyPnL, value.unrealizedPnL, value.realizedPnL):
            self.ready_account_keys.discard(key)
            self._write(
                {
                    "kind": "pnl-readiness-revoked",
                    "connectionEpoch": self.connection_epoch,
                    "scope": "account",
                    "account": value.account,
                    "modelCode": value.modelCode,
                    "pnlResetWindow": self.pnl_reset_window,
                    "observedAt": self._observed_at(),
                    "reason": "invalid-numeric-callback",
                }
            )
            raise ValueError("account P&L callback is not ready")
        try:
            self._write(
                {
                    "kind": "pnl-observation",
                    "connectionEpoch": self.connection_epoch,
                    "scope": "account",
                    "account": value.account,
                    "modelCode": value.modelCode,
                    "pnlResetWindow": self.pnl_reset_window,
                    "observedAt": self._observed_at(),
                    "dailyPnL": value.dailyPnL,
                    "unrealizedPnL": value.unrealizedPnL,
                    "realizedPnL": value.realizedPnL,
                }
            )
        except BaseException:
            self._cancel_account_transport(key)
            raise
        self.ready_account_keys.add(key)

    def observe_single(self, value: PnLSingle) -> None:
        self._require_healthy()
        key = (value.account, value.modelCode, value.conId)
        if self.single_pnl.get(key) is not value:
            raise RuntimeError("unowned single-position P&L callback")
        if not self._finite(
            value.position,
            value.dailyPnL,
            value.unrealizedPnL,
            value.realizedPnL,
            value.value,
        ):
            self.ready_single_keys.discard(key)
            self._write(
                {
                    "kind": "pnl-readiness-revoked",
                    "connectionEpoch": self.connection_epoch,
                    "scope": "single-position",
                    "account": value.account,
                    "modelCode": value.modelCode,
                    "conId": value.conId,
                    "pnlResetWindow": self.pnl_reset_window,
                    "observedAt": self._observed_at(),
                    "reason": "invalid-numeric-callback",
                }
            )
            raise ValueError("single-position P&L callback is not ready")
        try:
            self._write(
                {
                    "kind": "pnl-observation",
                    "connectionEpoch": self.connection_epoch,
                    "scope": "single-position",
                    "account": value.account,
                    "modelCode": value.modelCode,
                    "conId": value.conId,
                    "pnlResetWindow": self.pnl_reset_window,
                    "observedAt": self._observed_at(),
                    "position": value.position,
                    "dailyPnL": value.dailyPnL,
                    "unrealizedPnL": value.unrealizedPnL,
                    "realizedPnL": value.realizedPnL,
                    "value": value.value,
                }
            )
        except BaseException:
            self._cancel_single_transport(key)
            raise
        self.ready_single_keys.add(key)

    def _cancel_account_transport(
        self, key: tuple[str, str], recycle_on_failure: bool = True
    ) -> None:
        self.account_pnl.pop(key, None)
        self.ready_account_keys.discard(key)
        ib: IB = self.ib
        account, model_code = key
        try:
            ib.cancelPnL(account, model_code)
        except BaseException as error:
            if recycle_on_failure:
                self._cancel_transport_failed("account", key, error)
            else:
                self.poisoned = True
                self._clear_readiness()
            raise

    def _cancel_single_transport(
        self, key: tuple[str, str, int], recycle_on_failure: bool = True
    ) -> None:
        self.single_pnl.pop(key, None)
        self.ready_single_keys.discard(key)
        ib: IB = self.ib
        account, model_code, con_id = key
        try:
            ib.cancelPnLSingle(account, model_code, con_id)
        except BaseException as error:
            if recycle_on_failure:
                self._cancel_transport_failed("single-position", key, error)
            else:
                self.poisoned = True
                self._clear_readiness()
            raise

    def _cancel_transport_failed(
        self,
        scope: str,
        key: tuple[str, str] | tuple[str, str, int],
        error: BaseException,
    ) -> None:
        self.poisoned = True
        self.closed = True
        self.positions_owned = False
        self.account_pnl.clear()
        self.single_pnl.clear()
        self._clear_readiness()
        self._detach_events()
        epoch_recycled = False
        try:
            self.ib.disconnect()
            epoch_recycled = True
        except BaseException:
            pass
        recovery: Record = {
            "kind": "pnl-cancellation-failed",
            "connectionEpoch": self.connection_epoch,
            "scope": scope,
            "ownerKey": list(key),
            "errorType": type(error).__name__,
            "epochRecycled": epoch_recycled,
        }
        try:
            recovery["observedAt"] = self._observed_at()
            self.persist(recovery)
        except BaseException:
            pass

    def cancel_account(self, account: str, model_code: str = "") -> None:
        self._require_healthy()
        key = (account, model_code)
        if key not in self.account_pnl:
            raise KeyError("account P&L key is not owned")
        self._cancel_account_transport(key)
        self._write(
            {
                "kind": "pnl-subscription-cancelled",
                "connectionEpoch": self.connection_epoch,
                "scope": "account",
                "account": account,
                "modelCode": model_code,
                "pnlResetWindow": self.pnl_reset_window,
                "observedAt": self._observed_at(),
            }
        )

    def cancel_single(self, account: str, model_code: str, con_id: int) -> None:
        self._require_healthy()
        key = (account, model_code, con_id)
        if key not in self.single_pnl:
            raise KeyError("single-position P&L key is not owned")
        self._cancel_single_transport(key)
        self._write(
            {
                "kind": "pnl-subscription-cancelled",
                "connectionEpoch": self.connection_epoch,
                "scope": "single-position",
                "account": account,
                "modelCode": model_code,
                "conId": con_id,
                "pnlResetWindow": self.pnl_reset_window,
                "observedAt": self._observed_at(),
            }
        )

    def close(self) -> None:
        if self.closed:
            return
        self._detach_events()
        self.closed = True
        pending_positions = self.positions_pending
        pending_task = self._position_start_task
        if pending_task is not None:
            try:
                current_task = asyncio.current_task()
            except RuntimeError:
                current_task = None
            if pending_task is not current_task:
                pending_task.cancel()
        failures: list[BaseException] = []
        outcomes: list[Record] = []
        for key in tuple(self.single_pnl):
            try:
                self._cancel_single_transport(key, recycle_on_failure=False)
                outcomes.append({"scope": "single-position", "status": "cancelled"})
            except BaseException as error:
                failures.append(error)
                outcomes.append(
                    {
                        "scope": "single-position",
                        "status": "cancel-failed",
                        "errorType": type(error).__name__,
                    }
                )
        for key in tuple(self.account_pnl):
            try:
                self._cancel_account_transport(key, recycle_on_failure=False)
                outcomes.append({"scope": "account", "status": "cancelled"})
            except BaseException as error:
                failures.append(error)
                outcomes.append(
                    {
                        "scope": "account",
                        "status": "cancel-failed",
                        "errorType": type(error).__name__,
                    }
                )
        if self.positions_owned or pending_positions:
            client: Client = self.ib.client
            try:
                client.cancelPositions()
                outcomes.append({"scope": "positions", "status": "cancelled"})
            except BaseException as error:
                failures.append(error)
                outcomes.append(
                    {
                        "scope": "positions",
                        "status": "cancel-failed",
                        "errorType": type(error).__name__,
                    }
                )
        self.positions_owned = False
        self.positions_pending = False
        self._clear_readiness()
        if pending_positions:
            epoch_recycled = False
            try:
                self.ib.disconnect()
                epoch_recycled = True
            except BaseException as error:
                failures.append(error)
                outcomes.append(
                    {
                        "scope": "connection",
                        "status": "disconnect-failed",
                        "errorType": type(error).__name__,
                    }
                )
            pending_record: Record = {
                "kind": "position-start-cancelled",
                "connectionEpoch": self.connection_epoch,
                "outcomes": outcomes,
                "epochRecycled": epoch_recycled,
            }
            try:
                pending_record["observedAt"] = self._observed_at()
                self.persist(pending_record)
            except BaseException:
                pass
        if failures:
            self.poisoned = True
            epoch_recycled = False
            try:
                self.ib.disconnect()
                epoch_recycled = True
            except BaseException as error:
                outcomes.append(
                    {
                        "scope": "connection",
                        "status": "disconnect-failed",
                        "errorType": type(error).__name__,
                    }
                )
            recovery: Record = {
                "kind": "position-pnl-cleanup-failed",
                "connectionEpoch": self.connection_epoch,
                "outcomes": outcomes,
                "epochRecycled": epoch_recycled,
            }
            try:
                recovery["observedAt"] = self._observed_at()
                self.persist(recovery)
            except BaseException:
                pass
            raise failures[0]

PositionPnlOwner.start_positions attaches the continuing position handler before it requests the snapshot, explicitly awaits the global position-end boundary, publishes the completed snapshot, and then replays callbacks buffered across the completion handoff. The owner-bound receipt cannot be adopted by another owner, and cache inspection alone cannot prove readiness. A failed bounded refresh attempts to cancel the local future and protocol subscription, then disconnects in a guaranteed finalizer so partial cache and late same-key callbacks cannot contaminate another attempt.

Attach P&L event handlers before calling start_account or start_single; a callback can arrive immediately. The accepted observation is the readiness signal, not construction of the mutable return object, and a later invalid numeric callback revokes that key's readiness. A dispatch exception poisons the owner, attempts exact cleanup, persists the recovery outcome, and recycles the epoch. On an initial-wait timeout, cancel the exact key and treat the stream as not ready. close() clears readiness, attempts the global position cancellation plus every P&L key even when one cancellation fails, and detaches handlers; any partial cleanup failure is persisted and forces epoch recycling.

Signals, ordering, and failure handling

Signal or condition	Interpretation	Required response
Managed-accounts callback during connection	Accessible IDs for this authenticated session arrived.	Compare with application policy before publishing account readiness.
Account-update-multi payloads followed by an end callback	One request's initial snapshot completed. Other account requests may still be in flight.	Track readiness per intended account/model; tolerate interleaving.
Account-summary payloads followed by summary-end	The initial summary response completed.	Validate required account/tag/currency tuples; do not assume every possible tag exists.
Later account-value or summary event	A synchronized value changed. Duplicate logical keys replace cached values.	Make consumers idempotent and include account, tag, currency, and model where applicable.
Empty accessible set	Connection startup did not yield an account scope usable by this application.	Clear readiness and fail closed; investigate permissions/session configuration.
Missing or unexpected account	Authenticated visibility drifted from policy.	Do not broaden scope automatically; stop the workload and alert without exposing IDs in public telemetry.
More than 50 accessible accounts	Automatic subaccount synchronization is skipped by pinned library defaults.	Use an explicitly designed account/profile batching strategy; do not infer readiness from connection success.
Timeout or reconnect	The initial request may be incomplete, and disconnect resets wrapper state.	Clear account readiness, end the epoch, reconnect, rediscover, and rebuild caches.

Subscription ownership and persistence

accountSummaryAsync() and reqAccountUpdatesMultiAsync() await an initial completion signal, but their protocol requests establish continuing updates. Do not call them repeatedly as if they were independent snapshots. Let the connection owner create the required subscriptions once per epoch and use the cached accessors/events afterward.

The low-level client exposes request IDs and matching cancellation operations for applications that truly need dynamic subscription ownership. That path must retain each request ID, cancel it exactly once, and keep callbacks from old scopes out of new state. The high-level helpers above hide the request ID; deliberate IB.disconnect() is their connection-epoch cleanup boundary.

Persist the authorized account policy and the fact that a particular connection epoch passed scope validation in an access-controlled store. Do not persist wrapper objects as the durable account record, and do not treat account-summary or account-value streams as a financial ledger. Later workflow steps add positions, PnL, orders, executions, and reconciliation on top of this scope gate.

For reporting-period account controls rather than live TWS state, start with the Flex reporting boundary and retrieve each report instance through the version 3 request lifecycle.

Sources and applicability

• IBKR Campus: managed accounts callback — official-current; retrieved 2026-07-14T19:13:54.602937Z; supports automatic accessible-account discovery during initial connection.
• IBKR Campus: account summary and requesting account summary — official-current; retrieved 2026-07-14T19:13:54.602937Z; support subscription capacity, initial/end behavior, update cadence, group/tag selection, and large-account restrictions.
• IBKR Campus: account updates multi — official-current; retrieved 2026-07-14T19:13:54.602937Z; supports account/model scope and the introducing-broker All restriction.
• IBKR Campus: positions and position cancellation — official-current; retrieved 2026-07-14T19:13:54.602937Z; support the all-accessible-account initial/update stream, initial-end boundary, and global cancellation pair.
• IBKR Campus: profit and loss, account P&L, and single-position P&L — official-current; retrieved 2026-07-14T19:13:54.602937Z; support TWS reset/calculation scope, account/model/contract subscription identity, current fields, approximate cadence, invalid-contract silence, and the on-demand broker boundary.
• IB cached account accessors — library-source, ib_async 2.1.0; retrieved 2026-07-15T03:42:42Z; supports account discovery, cache filtering, and the synchronous summary wrapper.
• IB position/P&L accessors and owners, request/cancel pairs, and wrapper mutation/events — library-source, 2.1.0; retrieved 2026-07-15T03:42:42Z; support cache filtering, exact owner keys, mutable return objects, cancellation cleanup, zero-size removal, event ordering, and initial position completion.
• IB.connectAsync account startup — library-source, 2.1.0; retrieved 2026-07-15T03:42:42Z; supports single-account selection, startup flags, and the 50-subaccount synchronization boundary.
• IB async account helpers and wrapper account callbacks — library-source, 2.1.0; retrieved 2026-07-15T03:42:42Z; support initial completion, cache keys, events, local filtering, fixed summary tags, and continuing synchronized state.

No live account query is required for this page. CI compiles all three examples as ordinary scripts, binds typed ib_async calls to 2.1.0 signatures, and executes scope, cache/zero-update, completed-receipt, timeout/recycle, exact cancellation, disconnect, dispatch-failure, reset-window, and durable-write behavior against offline state.