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()

Allow Control-C to end program.

barplot

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

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

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

dataclassAsTuple

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

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

dataclassNonDefaults

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

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

dataclassRepr

dataclassRepr(obj) -> str

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

dataclassUpdate

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

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)

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

Format date or datetime to string that IB uses.

formatSI

formatSI(n: float) -> str

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

getLoop

getLoop()

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

Not a number test.

isnamedtupleinstance

isnamedtupleinstance(x)

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

logToConsole

logToConsole(level=20)

Create a log handler that logs to the console.

logToFile

logToFile(path, level=20)

Create a log handler that logs to the given file.

parseIBDatetime

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

Parse string in IB date or datetime format to datetime.

patchAsyncio

patchAsyncio()

Patch asyncio to allow nested event loops.

run

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

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)

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

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()

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]

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]

Async version of timeRange.

tree

tree(obj)

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)

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

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

Async version of waitUntil.

timeit

timeit(title='Run')

Context manager for timing.

__enter__

__enter__(self)

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

__exit__

__exit__(self, *_args)

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