Utilities
ib_interface.api.util
Utilities.
df
def df(objs, labels: Optional[List[str]] = None)
Create pandas DataFrame from the sequence of same-type objects.
Args: labels: If supplied, retain only the given labels and drop the rest.
dataclassAsDict
def dataclassAsDict(obj) -> dict
Return dataclass values as dict.
This is a non-recursive variant of dataclasses.asdict.
dataclassAsTuple
def dataclassAsTuple(obj) -> tuple
Return dataclass values as tuple.
This is a non-recursive variant of dataclasses.astuple.
dataclassNonDefaults
def dataclassNonDefaults(obj) -> dict
For a dataclass instance get the fields that are different from the
default values and return as dict.
dataclassUpdate
def dataclassUpdate(obj, srcObjs = (), kwargs = {}) -> object
Update fields of the given dataclass object from zero or more
dataclass source objects and/or from keyword arguments.
dataclassRepr
def dataclassRepr(obj) -> str
Provide a culled representation of the given dataclass instance,
showing only the fields with a non-default value.
isnamedtupleinstance
def isnamedtupleinstance(x)
From https://stackoverflow.com/a/2166841/6067848
tree
def tree(obj)
Convert object to a tree of lists, dicts and simple values. The result can be serialized to JSON.
barplot
def 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.
allowCtrlC
def allowCtrlC()
Allow Control-C to end program.
logToFile
def logToFile(path, level = logging.INFO)
Create a log handler that logs to the given file.
logToConsole
def logToConsole(level = logging.INFO)
Create a log handler that logs to the console.
isNan
def isNan(x: float) -> bool
Not a number test.
formatSI
def formatSI(n: float) -> str
Format the integer or float n to 3 significant digits + SI prefix.
timeit
Context manager for timing.
run
def run(awaitables: Awaitable = (), timeout: Optional[float] = 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
def schedule(time: Time_t, callback: Callable, args = ())
Schedule the callback to be run at the given time with the given arguments. This will return the Event Handle.
Args:
time (datetime.time): Time to run callback. If given as
:py:class:datetime.time then use today as date.
callback (callable): Callable scheduled to run.
args (tuple): Arguments for to call callback with.
sleep
def 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.
timeRange
def timeRange(start: Time_t, end: Time_t, step: float) -> Iterator[dt.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
waitUntil
def waitUntil(t: Time_t) -> 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.
timeRangeAsync
def timeRangeAsync(start: Time_t, end: Time_t, step: float) -> AsyncIterator[dt.datetime]
Async version of :meth:timeRange.
waitUntilAsync
def waitUntilAsync(t: Time_t) -> bool
Async version of :meth:waitUntil.
patchAsyncio
def patchAsyncio()
Patch asyncio to allow nested event loops.
getLoop
def getLoop()
Get the asyncio event loop for the current thread.
startLoop
def startLoop()
Use nested asyncio event loop for Jupyter notebooks.
formatIBDatetime
def formatIBDatetime(t: Union[dt.date, dt.datetime, str, None]) -> str
Format date or datetime to string that IB uses.
parseIBDatetime
def parseIBDatetime(s: str) -> Union[dt.date, dt.datetime]
Parse string in IB date or datetime format to datetime.