Coverage for adrs/oms/price_feed.py: 100%

1import logging 

2import time 

3from dataclasses import dataclass 

4from decimal import Decimal 

5from typing import Callable 

6 

7from cybotrade import Symbol 

8 

9logger = logging.getLogger(__name__) 

10 

11# How long since *any* frame on the connection before every quote is withheld. 

12# 

13# This is the liveness signal, not a per-quote freshness limit. @bookTicker 

14# pushes on every top-of-book change, so silence on a healthy socket means the 

15# book has not moved and the last quote is still the current one. The heartbeat 

16# subscription (BTCUSDT) was measured at 58-212 frames/sec with a 515ms worst 

17# gap over 60s, so 2s is ~4x the observed worst case: tight enough to catch a 

18# wedged socket inside one placement tick, loose enough to survive a scheduling 

19# hiccup. 

20DEFAULT_HEARTBEAT_MAX_AGE_SEC = 2.0 

21 

22# Backstop for the single failure liveness cannot see: one symbol's subscription 

23# dying while the connection keeps delivering others. Deliberately generous — a 

24# tight value would reintroduce the REST-per-tick cost on quiet symbols that this 

25# feed exists to remove. The trade is up to this long quoting from a dead 

26# subscription, versus REST on every tick for every quiet symbol. 

27DEFAULT_QUOTE_MAX_AGE_SEC = 60.0 

28 

29 

30@dataclass(frozen=True, slots=True) 

31class Quote: 

32 bid: Decimal 

33 ask: Decimal 

34 received_at: float # time.monotonic() at parse time 

35 

36 @property 

37 def mid(self) -> Decimal: 

38 # Must match cybotrade's ExchangeClient.get_current_price exactly, or the 

39 # feed and REST paths would quote differently for the same book. 

40 return (self.bid + self.ask) / Decimal("2.0") 

41 

42 

43class PriceFeed: 

44 """ 

45 Top-of-book cache with a staleness policy. Pure: it makes no exchange calls, 

46 knows nothing about websockets, and decides nothing about what to do on a 

47 miss — callers fall back to REST. 

48 

49 `heartbeat_max_age_sec=None` disables the connection-liveness gate and leaves 

50 per-symbol age as the only guard. **Do not set it without first measuring that 

51 the stream re-pushes an unchanged book for an IDLE symbol.** An earlier version 

52 of this docstring asserted Bybit's orderbook.1 re-pushes on a 3s timer; that was 

53 never verified and it contradicts the adapter actually shipped, whose rationale 

54 is the opposite — see BybitPublicWS, which subscribes a busy heartbeat symbol 

55 precisely because it treats the stream as change-driven. What was measured is 

56 only that BTCUSDT delivers 21-61 frames/sec with a 590ms worst gap, which a 

57 change-driven stream on a busy book explains just as well as a timer. Acting on 

58 the old claim would have switched off the liveness gate on the strength of an 

59 unmeasured assumption. 

60 """ 

61 

62 def __init__( 

63 self, 

64 heartbeat_max_age_sec: float | None = DEFAULT_HEARTBEAT_MAX_AGE_SEC, 

65 quote_max_age_sec: float = DEFAULT_QUOTE_MAX_AGE_SEC, 

66 clock: Callable[[], float] = time.monotonic, 

67 ): 

68 self._heartbeat_max_age_sec = heartbeat_max_age_sec 

69 self._quote_max_age_sec = quote_max_age_sec 

70 self._clock = clock 

71 self._quotes: dict[Symbol, Quote] = {} 

72 self._last_message_at: float | None = None 

73 self._served = 0 

74 self._fallback_no_liveness = 0 

75 self._fallback_unseen = 0 

76 self._fallback_backstop = 0 

77 self._rejected = 0 

78 

79 def note_message(self) -> None: 

80 """ 

81 Record that the connection delivered something. 

82 

83 Called for every frame, including ones that do not yield a quote: a 

84 malformed frame still proves the socket is alive. 

85 """ 

86 self._last_message_at = self._clock() 

87 

88 def apply(self, symbol: Symbol, bid: Decimal, ask: Decimal) -> bool: 

89 """ 

90 Store a quote. False if the book is crossed or zero, in which case this 

91 symbol's cached quote is dropped too. 

92 

93 Dropping it is the point. A symbol that starts pushing an unusable book 

94 — delisting, settlement, one side of the book emptied — otherwise has 

95 every update rejected while liveness stays fresh off other symbols, so 

96 get() would keep serving the last good quote for the full backstop 

97 window. That is the only path in this feed that can knowingly serve a 

98 wrong price; falling back to REST reads the real book instead. Liveness 

99 still counts: the frame arrived, so the socket is provably delivering. 

100 """ 

101 self.note_message() 

102 if bid <= 0 or ask <= 0 or bid >= ask: 

103 self._rejected += 1 

104 self._quotes.pop(symbol, None) 

105 logger.warning( 

106 f"[PRICE_FEED] Rejected implausible book for {symbol} and dropped " 

107 f"its cached quote: bid={bid} ask={ask}" 

108 ) 

109 return False 

110 self._quotes[symbol] = Quote(bid=bid, ask=ask, received_at=self._clock()) 

111 return True 

112 

113 def get(self, symbol: Symbol) -> Quote | None: 

114 """The current quote, or None when the caller must fall back to REST.""" 

115 now = self._clock() 

116 if self._heartbeat_max_age_sec is not None: 

117 if ( 

118 self._last_message_at is None 

119 or now - self._last_message_at > self._heartbeat_max_age_sec 

120 ): 

121 self._fallback_no_liveness += 1 

122 return None 

123 quote = self._quotes.get(symbol) 

124 if quote is None: 

125 self._fallback_unseen += 1 

126 return None 

127 if now - quote.received_at > self._quote_max_age_sec: 

128 self._fallback_backstop += 1 

129 return None 

130 self._served += 1 

131 return quote 

132 

133 def clear(self) -> None: 

134 """ 

135 Drop every quote and reset liveness. Called on reconnect: after a gap in 

136 delivery no cached quote can be trusted, so each symbol takes one REST 

137 read before rejoining the feed. 

138 """ 

139 self._quotes.clear() 

140 self._last_message_at = None 

141 

142 def invalidate(self, symbol: Symbol) -> None: 

143 """Drop one symbol. Unused on Binance; Bybit needs it on a sequence gap.""" 

144 self._quotes.pop(symbol, None) 

145 

146 def stats(self) -> dict[str, int | float | None]: 

147 now = self._clock() 

148 return { 

149 "served": self._served, 

150 "fallback_no_liveness": self._fallback_no_liveness, 

151 "fallback_unseen": self._fallback_unseen, 

152 "fallback_backstop": self._fallback_backstop, 

153 "rejected": self._rejected, 

154 "tracked_symbols": len(self._quotes), 

155 "liveness_age_sec": ( 

156 None if self._last_message_at is None else now - self._last_message_at 

157 ), 

158 }