Improve stream reading and caching

betfair_parser/cache.py

@@ -18,6 +18,7 @@
     OCM,
     PV,
     ChangeType,
+    MarketDefinition,
     MatchedOrder,
     Order,
     OrderRunnerChange,
@@ -154,7 +155,7 @@ def update_meta(self, msg: Union[MCM, OCM]) -> None:
             self.clear()
 
     def clear(self) -> None:
-        return
+        raise NotImplementedError()
 
 
 class MarketCache(ChangeCache):
@@ -166,8 +167,10 @@ class MarketCache(ChangeCache):
     """
 
     def __init__(self):
-        self.order_book: defaultdict[str, defaultdict] = defaultdict(lambda: defaultdict(RunnerOrderBook))
-        self.market_definitions = {}
+        self.order_book: defaultdict[str, defaultdict[int, RunnerOrderBook]] = defaultdict(
+            lambda: defaultdict(RunnerOrderBook)
+        )
+        self.market_definitions: dict[str, MarketDefinition] = {}
 
     def clear(self) -> None:
         self.order_book.clear()
@@ -177,6 +180,9 @@ def update(self, mcm: MCM) -> None:
         self.update_meta(mcm)
         if mcm.is_heartbeat:
             return
+        if not mcm.mc:
+            return
+
         for mc in mcm.mc:
             if mc.img:
                 self.order_book.pop(mc.id, None)
@@ -236,7 +242,7 @@ class OrderCache(ChangeCache):
     """
 
     def __init__(self):
-        self.orders: defaultdict[str, defaultdict] = defaultdict(lambda: defaultdict(RunnerOrders))
+        self.orders: defaultdict[str, defaultdict[int, RunnerOrders]] = defaultdict(lambda: defaultdict(RunnerOrders))
 
     def clear(self) -> None:
         self.orders.clear()
@@ -245,6 +251,8 @@ def update(self, ocm: OCM) -> None:
         self.update_meta(ocm)
         if ocm.is_heartbeat:
             return
+        if not ocm.oc:
+            return
 
         for oc in ocm.oc:
             if oc.full_image:

betfair_parser/spec/streaming/__init__.py

@@ -48,20 +48,23 @@
     StartingPriceBack,
     StartingPriceLay,
     StrategyMatchChange,
+    StreamRef,
     Trade,
 )
 
 
-STREAM_REQUEST = Union[Authentication, MarketSubscription, OrderSubscription, Heartbeat]
-STREAM_RESPONSE = Union[Connection, Status, MCM, OCM]
-CHANGE_MESSAGE = Union[MCM, OCM]
-_STREAM_MESSAGES = Union[STREAM_RESPONSE, list[STREAM_RESPONSE], STREAM_REQUEST]
-_STREAM_DECODER = Decoder(_STREAM_MESSAGES, strict=False)
+StreamRequestType = Union[Authentication, MarketSubscription, OrderSubscription, Heartbeat]
+StreamResponseType = Union[Connection, Status, MCM, OCM]
+SubscriptionType = Union[MarketSubscription, OrderSubscription]
+ChangeMessageType = Union[MCM, OCM]
+StreamMessageType = Union[StreamResponseType, list[StreamResponseType], StreamRequestType]
 
+_STREAM_DECODER = Decoder(StreamMessageType, strict=False)
 
-def stream_decode(raw: Union[str, bytes]):
+
+def stream_decode(raw: Union[str, bytes]) -> StreamMessageType:
     return _STREAM_DECODER.decode(raw)
 
 
-def stream_decode_lines(raw: Union[str, bytes]):
+def stream_decode_lines(raw: Union[str, bytes]) -> list[StreamMessageType]:
     return _STREAM_DECODER.decode_lines(raw)

betfair_parser/spec/streaming/messages.py

@@ -2,9 +2,9 @@
 Definition of the betfair streaming API messages as defined in:
 - https://docs.developer.betfair.com/display/1smk3cen4v3lu3yomq5qye0ni/Exchange+Stream+API
 - https://github.com/betfair/stream-api-sample-code/blob/master/ESASwaggerSchema.json
- """
+"""
 
-from typing import Literal, Optional, Union
+from typing import Literal, Optional
 
 from betfair_parser.spec.common import BaseMessage, first_lower
 from betfair_parser.spec.streaming.enums import ChangeType, SegmentType, StatusErrorCode
@@ -14,19 +14,20 @@
     MarketFilter,
     OrderFilter,
     OrderMarketChange,
+    StreamRef,
 )
 
 
 class _StreamRequest(BaseMessage, tag_field="op", tag=first_lower, frozen=True):
     """Common parent class for any stream request."""
 
-    id: Optional[Union[int, str]] = None  # Client generated unique id to link request with response (like json rpc)
+    id: Optional[StreamRef] = None  # Client generated unique id to link request with response (like json rpc)
 
 
 class _StreamResponse(BaseMessage, tag_field="op", tag=str.lower, frozen=True):
     """Common parent class for any stream response."""
 
-    id: Optional[Union[int, str]] = None  # Client generated unique id to link request with response (like json rpc)
+    id: Optional[StreamRef] = None  # Client generated unique id to link request with response (like json rpc)
 
 
 class Authentication(_StreamRequest, kw_only=True, frozen=True):

betfair_parser/spec/streaming/type_definitions.py

@@ -21,6 +21,8 @@
 from betfair_parser.spec.streaming.enums import LapseStatusReasonCode, MarketDataFilterFields, PriceLadderDefinitionType
 
 
+StreamRef = Union[int, str]
+
 # Request objects