ib_async.util

Generated from the installed ib_async 2.1.0 package. Signatures and defaults are version-specific.

Utilities.

Public constants

Name	Value
EPOCH	datetime.datetime(1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)
UNSET_DOUBLE	1.7976931348623157e+308
UNSET_INTEGER	2147483647

allowCtrlC

allowCtrlC()

Completeness: advanced-low-level · Canonical ID: ib_async.util.allowCtrlC

Advanced or protocol-level surface classified separately from application utilities.

Allow Control-C to end program.

barplot

barplot(bars, title='', upColor='blue', downColor='red')

Completeness: advanced-low-level · Canonical ID: ib_async.util.barplot

Advanced or protocol-level surface classified separately from application utilities.

Create candlestick plot for the given bars. The bars can be given as a DataFrame or as a list of bar objects.

dataclassAsDict

dataclassAsDict(obj) -> dict

Completeness: advanced-low-level · Canonical ID: ib_async.util.dataclassAsDict

Advanced or protocol-level surface classified separately from application utilities.

Return dataclass values as dict. This is a non-recursive variant of dataclasses.asdict.

dataclassAsTuple

dataclassAsTuple(obj) -> tuple[typing.Any, ...]

Completeness: advanced-low-level · Canonical ID: ib_async.util.dataclassAsTuple

Advanced or protocol-level surface classified separately from application utilities.

Return dataclass values as tuple. This is a non-recursive variant of dataclasses.astuple.

dataclassNonDefaults

dataclassNonDefaults(obj) -> dict[str, typing.Any]

Completeness: advanced-low-level · Canonical ID: ib_async.util.dataclassNonDefaults

Advanced or protocol-level surface classified separately from application utilities.

For a dataclass instance get the fields that are different from the default values and return as dict.

dataclassRepr

dataclassRepr(obj) -> str

Completeness: advanced-low-level · Canonical ID: ib_async.util.dataclassRepr

Advanced or protocol-level surface classified separately from application utilities.

Provide a culled representation of the given dataclass instance, showing only the fields with a non-default value.

dataclassUpdate

dataclassUpdate(obj, *srcObjs, **kwargs) -> object

Completeness: advanced-low-level · Canonical ID: ib_async.util.dataclassUpdate

Advanced or protocol-level surface classified separately from application utilities.

Update fields of the given dataclass object from zero or more dataclass source objects and/or from keyword arguments.

df

df(objs, labels: list[str] | None = None)

Completeness: advanced-low-level · Canonical ID: ib_async.util.df

Advanced or protocol-level surface classified separately from application utilities.

Create pandas DataFrame from the sequence of same-type objects.

Args: labels: If supplied, retain only the given labels and drop the rest.

formatIBDatetime

formatIBDatetime(t: datetime.date | datetime.datetime | str | None) -> str

Completeness: advanced-low-level · Canonical ID: ib_async.util.formatIBDatetime

Advanced or protocol-level surface classified separately from application utilities.

Format date or datetime to string that IB uses.

formatSI

formatSI(n: float) -> str

Completeness: advanced-low-level · Canonical ID: ib_async.util.formatSI

Advanced or protocol-level surface classified separately from application utilities.

Format the integer or float n to 3 significant digits + SI prefix.

getLoop

getLoop()

Completeness: advanced-low-level · Canonical ID: ib_async.util.getLoop

Advanced or protocol-level surface classified separately from application utilities.

Get asyncio event loop with smart fallback handling.

This function is designed for use in synchronous contexts or when the execution context is unknown. It will:

1. Try to get the currently running event loop (if in async context)
2. Fall back to getting the current thread's event loop via policy
3. Create a new event loop if none exists or if the existing one is closed

For performance-critical async code paths, prefer using asyncio.get_running_loop() directly instead of this function.

Note: This function does NOT cache the loop to avoid stale loop bugs when loops are closed and recreated (e.g., in testing, Jupyter notebooks).

isNan

isNan(x: float) -> bool

Completeness: advanced-low-level · Canonical ID: ib_async.util.isNan

Advanced or protocol-level surface classified separately from application utilities.

Not a number test.

isnamedtupleinstance

isnamedtupleinstance(x)

Completeness: advanced-low-level · Canonical ID: ib_async.util.isnamedtupleinstance

Advanced or protocol-level surface classified separately from application utilities.

From https://stackoverflow.com/a/2166841/6067848

logToConsole

logToConsole(level=20)

Completeness: advanced-low-level · Canonical ID: ib_async.util.logToConsole

Advanced or protocol-level surface classified separately from application utilities.

Create a log handler that logs to the console.

logToFile

logToFile(path, level=20)

Completeness: advanced-low-level · Canonical ID: ib_async.util.logToFile

Advanced or protocol-level surface classified separately from application utilities.

Create a log handler that logs to the given file.

parseIBDatetime

parseIBDatetime(s: str) -> datetime.date | datetime.datetime

Completeness: advanced-low-level · Canonical ID: ib_async.util.parseIBDatetime

Advanced or protocol-level surface classified separately from application utilities.

Parse string in IB date or datetime format to datetime.

patchAsyncio

patchAsyncio()

Completeness: advanced-low-level · Canonical ID: ib_async.util.patchAsyncio

Advanced or protocol-level surface classified separately from application utilities.

Patch asyncio to allow nested event loops.

run

run(*awaitables: collections.abc.Awaitable, timeout: float | None = None)

Completeness: advanced-low-level · Canonical ID: ib_async.util.run

Advanced or protocol-level surface classified separately from application utilities.

By default run the event loop forever.

When awaitables (like Tasks, Futures or coroutines) are given then run the event loop until each has completed and return their results.

An optional timeout (in seconds) can be given that will raise asyncio.TimeoutError if the awaitables are not ready within the timeout period.

schedule

schedule(time: datetime.time | datetime.datetime, callback: collections.abc.Callable, *args)

Completeness: advanced-low-level · Canonical ID: ib_async.util.schedule

Advanced or protocol-level surface classified separately from application utilities.

Schedule the callback to be run at the given time with the given arguments. This will return the Event Handle.

Args: time: Time to run callback. If given as :pydatetime.time then use today as date. callback: Callable scheduled to run. args: Arguments for to call callback with.

sleep

sleep(secs: float = 0.02) -> bool

Completeness: advanced-low-level · Canonical ID: ib_async.util.sleep

Advanced or protocol-level surface classified separately from application utilities.

Wait for the given amount of seconds while everything still keeps processing in the background. Never use time.sleep().

Args: secs (float): Time in seconds to wait.

startLoop

startLoop()

Completeness: advanced-low-level · Canonical ID: ib_async.util.startLoop

Advanced or protocol-level surface classified separately from application utilities.

Use nested asyncio event loop for Jupyter notebooks.

timeRange

timeRange(start: datetime.time | datetime.datetime, end: datetime.time | datetime.datetime, step: float) -> collections.abc.Iterator[datetime.datetime]

Completeness: advanced-low-level · Canonical ID: ib_async.util.timeRange

Advanced or protocol-level surface classified separately from application utilities.

Iterator that waits periodically until certain time points are reached while yielding those time points.

Args: start: Start time, can be specified as datetime.datetime, or as datetime.time in which case today is used as the date end: End time, can be specified as datetime.datetime, or as datetime.time in which case today is used as the date step (float): The number of seconds of each period

timeRangeAsync

timeRangeAsync(start: datetime.time | datetime.datetime, end: datetime.time | datetime.datetime, step: float) -> collections.abc.AsyncIterator[datetime.datetime]

Completeness: advanced-low-level · Canonical ID: ib_async.util.timeRangeAsync

Advanced or protocol-level surface classified separately from application utilities.

Async version of timeRange.

tree

tree(obj)

Completeness: advanced-low-level · Canonical ID: ib_async.util.tree

Advanced or protocol-level surface classified separately from application utilities.

Convert object to a tree of lists, dicts and simple values. The result can be serialized to JSON.

useQt

useQt(qtLib: str = 'PyQt5', period: float = 0.01)

Completeness: advanced-low-level · Canonical ID: ib_async.util.useQt

Advanced or protocol-level surface classified separately from application utilities.

Run combined Qt5/asyncio event loop.

Args: qtLib: Name of Qt library to use:

• PyQt5
• PyQt6
• PySide2
• PySide6 period: Period in seconds to poll Qt.

waitUntil

waitUntil(t: datetime.time | datetime.datetime) -> bool

Completeness: advanced-low-level · Canonical ID: ib_async.util.waitUntil

Advanced or protocol-level surface classified separately from application utilities.

Wait until the given time t is reached.

Args: t: The time t can be specified as datetime.datetime, or as datetime.time in which case today is used as the date.

waitUntilAsync

waitUntilAsync(t: datetime.time | datetime.datetime) -> bool

Completeness: advanced-low-level · Canonical ID: ib_async.util.waitUntilAsync

Advanced or protocol-level surface classified separately from application utilities.

Async version of waitUntil.

timeit

timeit(title='Run')

Completeness: advanced-low-level · Canonical ID: ib_async.util.timeit

Advanced or protocol-level surface classified separately from application utilities.

Context manager for timing.

enter

__enter__(self)

Completeness: advanced-low-level · Canonical ID: ib_async.util.timeit.__enter__

Advanced or protocol-level surface classified separately from application utilities.

No library docstring is provided; consult the signature, type fields, and operational guides.

exit

__exit__(self, *_args)

Completeness: advanced-low-level · Canonical ID: ib_async.util.timeit.__exit__

Advanced or protocol-level surface classified separately from application utilities.

No library docstring is provided; consult the signature, type fields, and operational guides.