"""Access to realtime market information."""from dataclasses importdataclass,fieldfrom datetime importdatetimefrom typing importClassVarfrom eventkit importEvent,Opfrom ib\_async.contract importContractfrom ib\_async.objects import(Dividends,DOMLevel,EfpData,FundamentalRatios,IBDefaults,MktDepthData,OptionComputation,TickByTickAllLast,TickByTickBidAsk,TickByTickMidPoint,TickData,)from ib\_async.util importdataclassRepr,isNannan=float("nan")
[[docs]](../../api.html#ib_async.ticker.Ticker)@dataclassclass Ticker:""" Current market data such as bid, ask, last price, etc. for a contract. Streaming level-1 ticks of type :class:`.TickData` are stored in the ``ticks`` list. Streaming level-2 ticks of type :class:`.MktDepthData` are stored in the ``domTicks`` list. The order book (DOM) is available as lists of :class:`.DOMLevel` in ``domBids`` and ``domAsks``. Streaming tick-by-tick ticks are stored in ``tickByTicks``. For options the :class:`.OptionComputation` values for the bid, ask, resp. last price are stored in the ``bidGreeks``, ``askGreeks`` resp. ``lastGreeks`` attributes. There is also ``modelGreeks`` that conveys the greeks as calculated by Interactive Brokers' option model. Events: \* ``updateEvent`` (ticker: :class:`.Ticker`) """events: ClassVar = ("updateEvent",)contract: Contract | None = Nonetime: datetime | None = Nonetimestamp: float | None = NonemarketDataType: int = 1minTick: float = nanbid: float = nanbidSize: float = nanbidExchange: str = ""ask: float = nanaskSize: float = nanaskExchange: str = ""last: float = nanlastSize: float = nanlastExchange: str = ""lastTimestamp: datetime | None = NoneprevBid: float = nanprevBidSize: float = nanprevAsk: float = nanprevAskSize: float = nanprevLast: float = nanprevLastSize: float = nanvolume: float = nanopen: float = nanhigh: float = nanlow: float = nanclose: float = nanvwap: float = nanlow13week: float = nanhigh13week: float = nanlow26week: float = nanhigh26week: float = nanlow52week: float = nanhigh52week: float = nanbidYield: float = nanaskYield: float = nanlastYield: float = nanmarkPrice: float = nanhalted: float = nanrtHistVolatility: float = nanrtVolume: float = nanrtTradeVolume: float = nanrtTime: datetime | None = NoneavVolume: float = nantradeCount: float = nantradeRate: float = nanvolumeRate: float = nanvolumeRate3Min: float = nanvolumeRate5Min: float = nanvolumeRate10Min: float = nanshortable: float = nanshortableShares: float = nanindexFuturePremium: float = nanfuturesOpenInterest: float = nanputOpenInterest: float = nancallOpenInterest: float = nanputVolume: float = nancallVolume: float = nanavOptionVolume: float = nanhistVolatility: float = nanimpliedVolatility: float = nanopenInterest: float = nanlastRthTrade: float = nanlastRegTime: str = ""optionBidExch: str = ""optionAskExch: str = ""bondFactorMultiplier: float = nancreditmanMarkPrice: float = nancreditmanSlowMarkPrice: float = nandelayedLastTimestamp: datetime | None = NonedelayedHalted: float = nanreutersMutualFunds: str = ""etfNavClose: float = nanetfNavPriorClose: float = nanetfNavBid: float = nanetfNavAsk: float = nanetfNavLast: float = nanetfFrozenNavLast: float = nanetfNavHigh: float = nanetfNavLow: float = nansocialMarketAnalytics: str = ""estimatedIpoMidpoint: float = nanfinalIpoLast: float = nandividends: Dividends | None = NonefundamentalRatios: FundamentalRatios | None = Noneticks: list[TickData] = field(default\_factory=list)tickByTicks: list[TickByTickAllLast | TickByTickBidAsk | TickByTickMidPoint] = (field(default\_factory=list))domBids: list[DOMLevel] = field(default\_factory=list)domBidsDict: dict[int, DOMLevel] = field(default\_factory=dict)domAsks: list[DOMLevel] = field(default\_factory=list)domAsksDict: dict[int, DOMLevel] = field(default\_factory=dict)domTicks: list[MktDepthData] = field(default\_factory=list)bidGreeks: OptionComputation | None = NoneaskGreeks: OptionComputation | None = NonelastGreeks: OptionComputation | None = NonemodelGreeks: OptionComputation | None = NonecustGreeks: OptionComputation | None = NonebidEfp: EfpData | None = NoneaskEfp: EfpData | None = NonelastEfp: EfpData | None = NoneopenEfp: EfpData | None = NonehighEfp: EfpData | None = NonelowEfp: EfpData | None = NonecloseEfp: EfpData | None = NoneauctionVolume: float = nanauctionPrice: float = nanauctionImbalance: float = nanregulatoryImbalance: float = nanbboExchange: str = ""snapshotPermissions: int = 0defaults: IBDefaults = field(default\_factory=IBDefaults, repr=False)created: bool = Falsedef \_\_post\_init\_\_(self):# when copying a dataclass, the \_\_post\_init\_\_ runs again, so we# want to make sure if this was \_already\_ created, we don't overwrite# everything with \_another\_ post\_init clear.if not self.created:self.updateEvent = TickerUpdateEvent("updateEvent")self.minTick = self.defaults.unsetself.bid = self.defaults.unsetself.bidSize = self.defaults.unsetself.ask = self.defaults.unsetself.askSize = self.defaults.unsetself.last = self.defaults.unsetself.lastSize = self.defaults.unsetself.prevBid = self.defaults.unsetself.prevBidSize = self.defaults.unsetself.prevAsk = self.defaults.unsetself.prevAskSize = self.defaults.unsetself.prevLast = self.defaults.unsetself.prevLastSize = self.defaults.unsetself.volume = self.defaults.unsetself.open = self.defaults.unsetself.high = self.defaults.unsetself.low = self.defaults.unsetself.close = self.defaults.unsetself.vwap = self.defaults.unsetself.low13week = self.defaults.unsetself.high13week = self.defaults.unsetself.low26week = self.defaults.unsetself.high26week = self.defaults.unsetself.low52week = self.defaults.unsetself.high52week = self.defaults.unsetself.bidYield = self.defaults.unsetself.askYield = self.defaults.unsetself.lastYield = self.defaults.unsetself.markPrice = self.defaults.unsetself.halted = self.defaults.unsetself.rtHistVolatility = self.defaults.unsetself.rtVolume = self.defaults.unsetself.rtTradeVolume = self.defaults.unsetself.avVolume = self.defaults.unsetself.tradeCount = self.defaults.unsetself.tradeRate = self.defaults.unsetself.volumeRate = self.defaults.unsetself.volumeRate3Min = self.defaults.unsetself.volumeRate5Min = self.defaults.unsetself.volumeRate10Min = self.defaults.unsetself.shortable = self.defaults.unsetself.shortableShares = self.defaults.unsetself.indexFuturePremium = self.defaults.unsetself.futuresOpenInterest = self.defaults.unsetself.putOpenInterest = self.defaults.unsetself.callOpenInterest = self.defaults.unsetself.putVolume = self.defaults.unsetself.callVolume = self.defaults.unsetself.avOptionVolume = self.defaults.unsetself.histVolatility = self.defaults.unsetself.impliedVolatility = self.defaults.unsetself.auctionVolume = self.defaults.unsetself.auctionPrice = self.defaults.unsetself.auctionImbalance = self.defaults.unsetself.regulatoryImbalance = self.defaults.unsetself.openInterest = self.defaults.unsetself.lastRthTrade = self.defaults.unsetself.bondFactorMultiplier = self.defaults.unsetself.creditmanMarkPrice = self.defaults.unsetself.creditmanSlowMarkPrice = self.defaults.unsetself.delayedHalted = self.defaults.unsetself.etfNavClose = self.defaults.unsetself.etfNavPriorClose = self.defaults.unsetself.etfNavBid = self.defaults.unsetself.etfNavAsk = self.defaults.unsetself.etfNavLast = self.defaults.unsetself.etfFrozenNavLast = self.defaults.unsetself.etfNavHigh = self.defaults.unsetself.etfNavLow = self.defaults.unsetself.estimatedIpoMidpoint = self.defaults.unsetself.finalIpoLast = self.defaults.unsetself.created = Truedef \_\_eq\_\_(self, other):return self is otherdef \_\_hash\_\_(self):return id(self)\_\_repr\_\_ = dataclassRepr\_\_str\_\_ = dataclassRepr
[[docs]](../../api.html#ib_async.ticker.Ticker.isUnset)def isUnset(self, value) -\> bool:# if default value is nan and value is nan, it is unset.# else, if value matches default value, it is unset.dev = self.defaults.unsetreturn (dev != dev and value != value) or (value == dev)

[[docs]](../../api.html#ib_async.ticker.Ticker.hasBidAsk)def hasBidAsk(self) -\> bool:"""See if this ticker has a valid bid and ask."""return (self.bid != -1and not self.isUnset(self.bid)and self.bidSize \> 0and self.ask != -1and not self.isUnset(self.ask)and self.askSize \> 0)

[[docs]](../../api.html#ib_async.ticker.Ticker.midpoint)def midpoint(self) -\> float:""" Return average of bid and ask, or defaults.unset if no valid bid and ask are available. """return (self.bid + self.ask) \* 0.5 if self.hasBidAsk() else self.defaults.unset

[[docs]](../../api.html#ib_async.ticker.Ticker.marketPrice)def marketPrice(self) -\> float:""" Return the first available one of \* last price if within current bid/ask or no bid/ask available; \* average of bid and ask (midpoint). """if self.hasBidAsk():if self.bid \<= self.last \<= self.ask:price = self.lastelse:price = self.midpoint()else:price = self.lastreturn price

[[docs]](../../api.html#ib_async.ticker.TickerUpdateEvent)class TickerUpdateEvent(Event):\_\_slots\_\_ = ()
[[docs]](../../api.html#ib_async.ticker.TickerUpdateEvent.trades)def trades(self) -\> "Tickfilter":"""Emit trade ticks."""return Tickfilter((4, 5, 48, 68, 71), self)

[[docs]](../../api.html#ib_async.ticker.TickerUpdateEvent.bids)def bids(self) -\> "Tickfilter":"""Emit bid ticks."""return Tickfilter((0, 1, 66, 69), self)

[[docs]](../../api.html#ib_async.ticker.TickerUpdateEvent.asks)def asks(self) -\> "Tickfilter":"""Emit ask ticks."""return Tickfilter((2, 3, 67, 70), self)

[[docs]](../../api.html#ib_async.ticker.TickerUpdateEvent.bidasks)def bidasks(self) -\> "Tickfilter":"""Emit bid and ask ticks."""return Tickfilter((0, 1, 66, 69, 2, 3, 67, 70), self)

[[docs]](../../api.html#ib_async.ticker.TickerUpdateEvent.midpoints)def midpoints(self) -\> "Tickfilter":"""Emit midpoint ticks."""return Midpoints((), self)

[[docs]](../../api.html#ib_async.ticker.Tickfilter)class Tickfilter(Op):"""Tick filtering event operators that ``emit(time, price, size)``."""\_\_slots\_\_ = ("\_tickTypes",)def \_\_init\_\_(self, tickTypes, source=None):Op.\_\_init\_\_(self, source)self.\_tickTypes = set(tickTypes)
[[docs]](../../api.html#ib_async.ticker.Tickfilter.on_source)def on\_source(self, ticker):for t in ticker.ticks:if t.tickType in self.\_tickTypes:self.emit(t.time, t.price, t.size)

[[docs]](../../api.html#ib_async.ticker.Tickfilter.timebars)def timebars(self, timer: Event) -\> "TimeBars":""" Aggregate ticks into time bars, where the timing of new bars is derived from a timer event. Emits a completed :class:`Bar`. This event stores a :class:`BarList` of all created bars in the ``bars`` property. Args: timer: Event for timing when a new bar starts. """return TimeBars(timer, self)

[[docs]](../../api.html#ib_async.ticker.Tickfilter.tickbars)def tickbars(self, count: int) -\> "TickBars":""" Aggregate ticks into bars that have the same number of ticks. Emits a completed :class:`Bar`. This event stores a :class:`BarList` of all created bars in the ``bars`` property. Args: count: Number of ticks to use to form one bar. """return TickBars(count, self)

[[docs]](../../api.html#ib_async.ticker.Tickfilter.volumebars)def volumebars(self, volume: int) -\> "VolumeBars":""" Aggregate ticks into bars that have the same volume. Emits a completed :class:`Bar`. This event stores a :class:`BarList` of all created bars in the ``bars`` property. Args: count: Number of ticks to use to form one bar. """return VolumeBars(volume, self)

[[docs]](../../api.html#ib_async.ticker.Midpoints)class Midpoints(Tickfilter):\_\_slots\_\_ = ()
[[docs]](../../api.html#ib_async.ticker.Midpoints.on_source)def on\_source(self, ticker):if ticker.ticks:self.emit(ticker.time, ticker.midpoint(), 0)

[[docs]](../../api.html#ib_async.ticker.Bar)@dataclassclass Bar:time: datetime | Noneopen: float = nanhigh: float = nanlow: float = nanclose: float = nanvolume: int = 0count: int = 0

[[docs]](../../api.html#ib_async.ticker.BarList)class BarList(list[Bar]):def \_\_init\_\_(self, \*args):super().\_\_init\_\_(\*args)self.updateEvent = Event("updateEvent")def \_\_eq\_\_(self, other) -\> bool:return self is other

[[docs]](../../api.html#ib_async.ticker.TimeBars)class TimeBars(Op):\_\_slots\_\_ = ("\_timer","bars",)\_\_doc\_\_ = Tickfilter.timebars.\_\_doc\_\_bars: BarListdef \_\_init\_\_(self, timer, source=None):Op.\_\_init\_\_(self, source)self.\_timer = timerself.\_timer.connect(self.\_on\_timer, None, self.\_on\_timer\_done)self.bars = BarList()
[[docs]](../../api.html#ib_async.ticker.TimeBars.on_source)def on\_source(self, time, price, size):if not self.bars:returnbar = self.bars[-1]if isNan(bar.open):bar.open = bar.high = bar.low = pricebar.high = max(bar.high, price)bar.low = min(bar.low, price)bar.close = pricebar.volume += sizebar.count += 1self.bars.updateEvent.emit(self.bars, False)
def \_on\_timer(self, time):if self.bars:bar = self.bars[-1]if isNan(bar.close) and len(self.bars) \> 1:bar.open = bar.high = bar.low = bar.close = self.bars[-2].closeself.bars.updateEvent.emit(self.bars, True)self.emit(bar)self.bars.append(Bar(time))def \_on\_timer\_done(self, timer):self.\_timer = Noneself.set\_done()

[[docs]](../../api.html#ib_async.ticker.TickBars)class TickBars(Op):\_\_slots\_\_ = ("\_count", "bars")\_\_doc\_\_ = Tickfilter.tickbars.\_\_doc\_\_bars: BarListdef \_\_init\_\_(self, count, source=None):Op.\_\_init\_\_(self, source)self.\_count = countself.bars = BarList()
[[docs]](../../api.html#ib_async.ticker.TickBars.on_source)def on\_source(self, time, price, size):if not self.bars or self.bars[-1].count == self.\_count:bar = Bar(time, price, price, price, price, size, 1)self.bars.append(bar)else:bar = self.bars[-1]bar.high = max(bar.high, price)bar.low = min(bar.low, price)bar.close = pricebar.volume += sizebar.count += 1if bar.count == self.\_count:self.bars.updateEvent.emit(self.bars, True)self.emit(self.bars)

[[docs]](../../api.html#ib_async.ticker.VolumeBars)class VolumeBars(Op):\_\_slots\_\_ = ("\_volume", "bars")\_\_doc\_\_ = Tickfilter.volumebars.\_\_doc\_\_bars: BarListdef \_\_init\_\_(self, volume, source=None):Op.\_\_init\_\_(self, source)self.\_volume = volumeself.bars = BarList()
[[docs]](../../api.html#ib_async.ticker.VolumeBars.on_source)def on\_source(self, time, price, size):if not self.bars or self.bars[-1].volume \>= self.\_volume:bar = Bar(time, price, price, price, price, size, 1)self.bars.append(bar)else:bar = self.bars[-1]bar.high = max(bar.high, price)bar.low = min(bar.low, price)bar.close = pricebar.volume += sizebar.count += 1if bar.volume \>= self.\_volume:self.bars.updateEvent.emit(self.bars, True)self.emit(self.bars)