Source code for ib_insync.ticker

"""Access to realtime market information."""

from dataclasses import dataclass, field
from datetime import datetime
from typing import ClassVar, List, Optional, Union

from eventkit import Event, Op

from ib_insync.contract import Contract
from ib_insync.objects import (
    DOMLevel, Dividends, FundamentalRatios, MktDepthData,
    OptionComputation, TickByTickAllLast, TickByTickBidAsk, TickByTickMidPoint,
    TickData)
from ib_insync.util import dataclassRepr, isNan

nan = float('nan')


[docs]@dataclass class 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: Optional[Contract] = None time: Optional[datetime] = None marketDataType: int = 1 bid: float = nan bidSize: float = nan ask: float = nan askSize: float = nan last: float = nan lastSize: float = nan prevBid: float = nan prevBidSize: float = nan prevAsk: float = nan prevAskSize: float = nan prevLast: float = nan prevLastSize: float = nan volume: float = nan open: float = nan high: float = nan low: float = nan close: float = nan vwap: float = nan low13week: float = nan high13week: float = nan low26week: float = nan high26week: float = nan low52week: float = nan high52week: float = nan bidYield: float = nan askYield: float = nan lastYield: float = nan markPrice: float = nan halted: float = nan rtHistVolatility: float = nan rtVolume: float = nan rtTradeVolume: float = nan rtTime: Optional[datetime] = None avVolume: float = nan tradeCount: float = nan tradeRate: float = nan volumeRate: float = nan shortableShares: float = nan indexFuturePremium: float = nan futuresOpenInterest: float = nan putOpenInterest: float = nan callOpenInterest: float = nan putVolume: float = nan callVolume: float = nan avOptionVolume: float = nan histVolatility: float = nan impliedVolatility: float = nan dividends: Optional[Dividends] = None fundamentalRatios: Optional[FundamentalRatios] = None ticks: List[TickData] = field(default_factory=list) tickByTicks: List[Union[ TickByTickAllLast, TickByTickBidAsk, TickByTickMidPoint]] = \ field(default_factory=list) domBids: List[DOMLevel] = field(default_factory=list) domAsks: List[DOMLevel] = field(default_factory=list) domTicks: List[MktDepthData] = field(default_factory=list) bidGreeks: Optional[OptionComputation] = None askGreeks: Optional[OptionComputation] = None lastGreeks: Optional[OptionComputation] = None modelGreeks: Optional[OptionComputation] = None auctionVolume: float = nan auctionPrice: float = nan auctionImbalance: float = nan regulatoryImbalance: float = nan def __post_init__(self): self.updateEvent = TickerUpdateEvent('updateEvent') def __eq__(self, other): return self is other def __hash__(self): return id(self) __repr__ = dataclassRepr __str__ = dataclassRepr
[docs] def hasBidAsk(self) -> bool: """See if this ticker has a valid bid and ask.""" return ( self.bid != -1 and not isNan(self.bid) and self.bidSize > 0 and self.ask != -1 and not isNan(self.ask) and self.askSize > 0)
[docs] def midpoint(self) -> float: """ Return average of bid and ask, or NaN if no valid bid and ask are available. """ return (self.bid + self.ask) * 0.5 if self.hasBidAsk() else nan
[docs] def marketPrice(self) -> float: """ Return the first available one of * last price if within current bid/ask; * average of bid and ask (midpoint); * close price. """ price = self.last if ( self.hasBidAsk() and self.bid <= self.last <= self.ask) else \ self.midpoint() if isNan(price): price = self.close return price
[docs]class TickerUpdateEvent(Event): __slots__ = ()
[docs] def trades(self) -> "Tickfilter": """Emit trade ticks.""" return Tickfilter((4, 5, 48, 68, 71), self)
[docs] def bids(self) -> "Tickfilter": """Emit bid ticks.""" return Tickfilter((0, 1, 66, 69), self)
[docs] def asks(self) -> "Tickfilter": """Emit ask ticks.""" return Tickfilter((2, 3, 67, 70), self)
[docs] def bidasks(self) -> "Tickfilter": """Emit bid and ask ticks.""" return Tickfilter((0, 1, 66, 69, 2, 3, 67, 70), self)
[docs] def midpoints(self) -> "Tickfilter": """Emit midpoint ticks.""" return Midpoints((), self)
[docs]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] 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] 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] 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]class Midpoints(Tickfilter): __slots__ = ()
[docs] def on_source(self, ticker): if ticker.ticks: self.emit(ticker.time, ticker.midpoint(), 0)
[docs]@dataclass class Bar: time: Optional[datetime] open: float = nan high: float = nan low: float = nan close: float = nan volume: int = 0 count: int = 0
[docs]class BarList(List[Bar]): def __init__(self, *args): super().__init__(*args) self.updateEvent = Event('updateEvent') def __eq__(self, other): return self is other def __hash__(self): return id(self)
[docs]class TimeBars(Op): __slots__ = ('_timer', 'bars',) __doc__ = Tickfilter.timebars.__doc__ bars: BarList def __init__(self, timer, source=None): Op.__init__(self, source) self._timer = timer self._timer.connect(self._on_timer, None, self._on_timer_done) self.bars = BarList()
[docs] def on_source(self, time, price, size): if not self.bars: return bar = self.bars[-1] if isNan(bar.open): bar.open = bar.high = bar.low = price bar.high = max(bar.high, price) bar.low = min(bar.low, price) bar.close = price bar.volume += size bar.count += 1 self.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].close self.bars.updateEvent.emit(self.bars, True) self.emit(bar) self.bars.append(Bar(time)) def _on_timer_done(self, timer): self._timer = None self.set_done()
[docs]class TickBars(Op): __slots__ = ('_count', 'bars') __doc__ = Tickfilter.tickbars.__doc__ bars: BarList def __init__(self, count, source=None): Op.__init__(self, source) self._count = count self.bars = BarList()
[docs] 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 = price bar.volume += size bar.count += 1 if bar.count == self._count: self.bars.updateEvent.emit(self.bars, True) self.emit(self.bars)