refactor(stream): clarify lifecycle and raw frame ownership

docs/streaming.md

@@ -22,6 +22,42 @@ async def main():
 asyncio.run(main())
 ```
 
+## Stream lifecycle and adapter hooks
+
+`start()` schedules one background receiver and returns the stream immediately.
+It needs a running asyncio loop. It does not wait for an open connection,
+completed subscriptions or a ready book. `async with stream` starts receiving
+on entry and closes on exit; `async for event in stream` starts receiving if
+needed and removes events from the stream's single queue. Multiple readers
+share that queue rather than each receiving a copy. Closing ends iteration
+when the queued shutdown marker reaches the reader.
+
+For tests or a custom transport, pass
+`websocket_connection_opener=async_opener`. The opener is called on every
+connection attempt, including reconnects, with the URL and
+`additional_headers=stream.headers()`. It must return a connected socket
+supporting the adapter's send, receive and close operations. The default
+opener remains unchanged.
+
+For custom WebSocket adapters, implement `handle_raw(raw: str) -> list[Event]`.
+The adapter owns framing, decoding, cached-state updates and any native/Python
+selection. Return an empty list for a handled frame with no events; never
+return `None`. The shared receiver counts frames and emits the returned
+events. gRPC adapters retain their decoded-message `handle()` hook on
+`GrpcStream`.
+
+Migration on the experimental SDK branch:
+
+- Rename constructor argument `connect` to `websocket_connection_opener`.
+- Move a WebSocket subclass's old `handle(message)` logic behind `handle_raw()`;
+  the base class no longer decodes JSON or chooses a fallback handler.
+- Replace calls to the removed `status()` helper with the appropriate private
+  `_record_*_and_emit_status()` operation. These operations own the related
+  counters and lifecycle state. `_emit_status_event()` only publishes a status.
+- Instrumentation overriding private hooks must follow `_task` →
+  `_receiver_task`, `_run()` → `_run_connection_and_receive_loop()` and
+  `_dispatch()` → `_process_raw_frame_and_emit_events()`.
+
 ## Streams
 
 | Stream | Endpoint | Credentials | Carries |

synpath/__init__.py

@@ -84,6 +84,7 @@
 
 __version__ = "0.8.1"
 
+# Every venue this library speaks, by id.
 exchanges: dict[str, type[Exchange]] = {
     "kalshi": Kalshi,
     "polymarket": Polymarket,
@@ -96,7 +97,6 @@
     "betfair": Betfair,
     "smarkets": Smarkets,
 }
-"""Every venue this library speaks, by id."""
 
 
 def exchange(venue: str, **kwargs) -> Exchange:

synpath/_deprecated.py

@@ -23,8 +23,8 @@
 
 T = TypeVar("T")
 
+# The rename every contract-count field and argument went through.
 QUANTITY = {"amount": "quantity"}
-"""The rename every contract-count field and argument went through."""
 
 
 def warn(old: str, new: str, *, stacklevel: int = 3) -> None:

synpath/_native.py

@@ -23,8 +23,8 @@
     except ImportError:  # pragma: no cover - only without a built extension
         core = None
 
+# True when the Rust core is in use.
 ENABLED = core is not None
-"""True when the Rust core is in use."""
 
 
 def pick(name: str, fallback: Any) -> Any:

synpath/base.py

@@ -34,15 +34,24 @@
 from . import ids
 from .types import BookSide, Candle, Event, FeeSchedule, Market, OrderBook, Series, Trade
 
+# What a venue can do, in the shape ccxt's `has` uses.
+#
+#   True       — supported; the method's docstring says how the venue answers it
+#   "partial"  — supported, but the result is thinner than the type suggests;
+#                the method's docstring says exactly how
+#   False      — not available; calling raises `NotSupported`
 Capability = Literal[True, False, "partial"]
-"""What a venue can do, in the shape ccxt's `has` uses.
-
-  True       — supported; the method's docstring says how the venue answers it
-  "partial"  — supported, but the result is thinner than the type suggests;
-               the method's docstring says exactly how
-  False      — not available; calling raises `NotSupported`
-"""
 
+# Every question `has` can be asked, for every venue.
+#
+# This list is the single place a capability is named. `Exchange.__init_subclass__`
+# fills a venue's `has` from it, so a venue that says nothing about a capability
+# gets `False` rather than a hole. `has[key]` therefore never raises KeyError on
+# any venue — which matters, because the whole point of `has` is to be safe to
+# read *before* you know whether something is supported.
+#
+# The read-only adapters answer `False` to every trading key; a trading
+# adapter for the same venue subclasses one and says what it adds.
 CAPABILITIES: tuple[str, ...] = (
     # -- read --
     "fetch_markets",
@@ -94,64 +103,49 @@
     "watch_positions",
     "watch_balance",
 )
-"""Every question `has` can be asked, for every venue.
-
-This list is the single place a capability is named. `Exchange.__init_subclass__`
-fills a venue's `has` from it, so a venue that says nothing about a capability
-gets `False` rather than a hole. `has[key]` therefore never raises KeyError on
-any venue — which matters, because the whole point of `has` is to be safe to
-read *before* you know whether something is supported.
-
-The read-only adapters answer `False` to every trading key; a trading
-adapter for the same venue subclasses one and says what it adds.
-"""
 
 LEGAL_CAPABILITY_VALUES = frozenset({True, False, "partial"})
 
+# The status words every venue understands, with one meaning each.
+#
+#   open    — listed and trading
+#   closed  — trading stopped, outcome not yet final
+#   settled — outcome final and paid
+#   all     — no filter
+#
+# A venue that cannot filter by one of these raises `NotSupported` for it. What
+# it must not do is accept the word and quietly answer a different question:
+# `status="settled"` returning live markets on one venue and settled ones on
+# another is the kind of disagreement a unified API exists to remove.
 MARKET_STATUSES: tuple[str, ...] = ("open", "closed", "settled", "all")
-"""The status words every venue understands, with one meaning each.
-
-  open    — listed and trading
-  closed  — trading stopped, outcome not yet final
-  settled — outcome final and paid
-  all     — no filter
-
-A venue that cannot filter by one of these raises `NotSupported` for it. What
-it must not do is accept the word and quietly answer a different question:
-`status="settled"` returning live markets on one venue and settled ones on
-another is the kind of disagreement a unified API exists to remove.
-"""
 
+# How a page of markets can be ordered.
+#
+# Polymarket orders at the venue. Kalshi and Polymarket US accept a sort
+# parameter and ignore it, so on those the page is ordered here after it is
+# read, the way ccxt and pmxt do it: `sort` then means "this page, ordered by",
+# not "the first page of the whole catalog ordered by". Each adapter's
+# `fetch_markets` docstring says which venue figure the key reads.
 MARKET_SORTS: tuple[str, ...] = ("volume", "liquidity", "newest")
-"""How a page of markets can be ordered.
-
-Polymarket orders at the venue. Kalshi and Polymarket US accept a sort
-parameter and ignore it, so on those the page is ordered here after it is
-read, the way ccxt and pmxt do it: `sort` then means "this page, ordered by",
-not "the first page of the whole catalog ordered by". Each adapter's
-`fetch_markets` docstring says which venue figure the key reads.
-"""
 
+# Rows one catalog call can return, on every venue.
+#
+# Uniform rather than per-venue. Polymarket's keyset endpoint hard-caps at 100 no
+# matter what is asked, so a higher ceiling elsewhere would mean `limit=500`
+# returning 500 rows on one venue and 100 on another, with nothing saying it had
+# been reduced. Ask for more than this and it is clamped; the cursor is how you
+# get the rest.
 MAX_PAGE_LIMIT = 100
-"""Rows one catalog call can return, on every venue.
-
-Uniform rather than per-venue. Polymarket's keyset endpoint hard-caps at 100 no
-matter what is asked, so a higher ceiling elsewhere would mean `limit=500`
-returning 500 rows on one venue and 100 on another, with nothing saying it had
-been reduced. Ask for more than this and it is clamped; the cursor is how you
-get the rest.
-"""
 
+# Longest this library will sleep inside one call, in seconds.
+#
+# A venue's `Retry-After` is a hint, not an instruction. Honouring a large one
+# literally would let the venue park a caller's thread for as long as it likes --
+# and in the server that thread belongs to a pool shared with every other
+# request, so a handful of rate-limited calls would take the whole service down.
+# The header's real value is still on the raised `RateLimitExceeded`, so a caller
+# that wants to wait longer can decide that for itself.
 MAX_RETRY_WAIT = 10.0
-"""Longest this library will sleep inside one call, in seconds.
-
-A venue's `Retry-After` is a hint, not an instruction. Honouring a large one
-literally would let the venue park a caller's thread for as long as it likes --
-and in the server that thread belongs to a pool shared with every other
-request, so a handful of rate-limited calls would take the whole service down.
-The header's real value is still on the raised `RateLimitExceeded`, so a caller
-that wants to wait longer can decide that for itself.
-"""
 
 
 class RateLimiter:
@@ -282,9 +276,9 @@ def __init__(
         self.base_url = base_url.rstrip("/")
         self.limiter = limiter
         self.attempts = attempts
+        # Named at the front of every error this client raises, so a caller
+        #         reads `kalshi: insufficient balance`, not a class name and a path.
         self.venue = venue
-        """Named at the front of every error this client raises, so a caller
-        reads `kalshi: insufficient balance`, not a class name and a path."""
         self._client = client or httpx.Client(timeout=timeout, follow_redirects=True)
 
     def request(self, method: str, path: str, **kwargs: Any) -> Any:
@@ -355,9 +349,9 @@ def __init__(
         self.base_url = base_url.rstrip("/")
         self.limiter = limiter
         self.attempts = attempts
+        # Named at the front of every error this client raises, so a caller
+        #         reads `kalshi: insufficient balance`, not a class name and a path.
         self.venue = venue
-        """Named at the front of every error this client raises, so a caller
-        reads `kalshi: insufficient balance`, not a class name and a path."""
         self._client = client or httpx.AsyncClient(
             timeout=timeout, follow_redirects=True, headers=headers,
         )
@@ -534,12 +528,11 @@ class Exchange(ABC):
 
     id: str
     name: str
+    # What this venue can do. Always complete: see `__init_subclass__`.
+    #
+    #     A subclass declares only what differs from "no". Everything in
+    #     `CAPABILITIES` it stays silent about is filled in as `False`.
     has: dict[str, Capability] = {}
-    """What this venue can do. Always complete: see `__init_subclass__`.
-
-    A subclass declares only what differs from "no". Everything in
-    `CAPABILITIES` it stays silent about is filled in as `False`.
-    """
     face_value: float = 1.0
     book_model: str = "native_per_outcome"
 

synpath/betfair.py

@@ -76,6 +76,8 @@
 
 VENUE = "betfair"
 
+# Where an account logs in and calls the API, by the Betfair site it is
+# registered on. An Italian account cannot log in on .com, and so on.
 JURISDICTIONS: dict[str, dict[str, str]] = {
     "com": {"identity": "identitysso.betfair.com", "cert": "identitysso-cert.betfair.com", "api": "api.betfair.com"},
     "it": {"identity": "identitysso.betfair.it", "cert": "identitysso-cert.betfair.it", "api": "api.betfair.it"},
@@ -86,67 +88,65 @@
     "com.au": {"identity": "identitysso.betfair.com.au", "cert": "identitysso-cert.betfair.com.au",
                "api": "api.betfair.com"},
 }
-"""Where an account logs in and calls the API, by the Betfair site it is
-registered on. An Italian account cannot log in on .com, and so on."""
 
 BETTING_PATH = "/exchange/betting/rest/v1.0"
 
+# Betfair publishes no request rate for market data, only a weight per
+# request; kept modest and shared process-wide.
 LIMITER = RateLimiter(5.0, burst=10)
-"""Betfair publishes no request rate for market data, only a weight per
-request; kept modest and shared process-wide."""
 
+# Seconds after which the session is extended with `keepAlive` before the
+# next call. A session lasts 12 hours (24 on UK and Irish accounts) from its
+# last keepAlive, and API calls do not extend it, so a long-running process
+# has to.
 SESSION_REFRESH_AFTER = 4 * 3600.0
-"""Seconds after which the session is extended with `keepAlive` before the
-next call. A session lasts 12 hours (24 on UK and Irish accounts) from its
-last keepAlive, and API calls do not extend it, so a long-running process
-has to."""
 
+# A request's projection weight times its number of markets may not pass
+# this, or Betfair refuses the whole request with `TOO_MUCH_DATA`.
 MAX_WEIGHT = 200
-"""A request's projection weight times its number of markets may not pass
-this, or Betfair refuses the whole request with `TOO_MUCH_DATA`."""
 
 CATALOGUE_WEIGHTS: dict[str, int] = {
     "COMPETITION": 0, "EVENT": 0, "EVENT_TYPE": 0, "MARKET_START_TIME": 0, "RUNNER_DESCRIPTION": 0,
     "MARKET_DESCRIPTION": 1, "RUNNER_METADATA": 1,
 }
 BOOK_WEIGHTS: dict[str, int] = {"EX_BEST_OFFERS": 5, "EX_TRADED": 17, "EX_ALL_OFFERS": 17}
+# Betfair's weights for the pairs it prices below their sum.
 BOOK_COMBINED_WEIGHTS: dict[frozenset[str], int] = {
     frozenset({"EX_BEST_OFFERS", "EX_TRADED"}): 20,
     frozenset({"EX_ALL_OFFERS", "EX_TRADED"}): 32,
 }
-"""Betfair's weights for the pairs it prices below their sum."""
+# `listMarketBook` with no price projection still weighs 2 a market.
 BOOK_BASE_WEIGHT = 2
-"""`listMarketBook` with no price projection still weighs 2 a market."""
 
 LISTING_PROJECTION = ("EVENT", "EVENT_TYPE", "COMPETITION", "MARKET_START_TIME", "MARKET_DESCRIPTION",
                       "RUNNER_DESCRIPTION")
 LOOKUP_PROJECTION = LISTING_PROJECTION + ("RUNNER_METADATA",)
+# The only price projection a delayed key serves in full: three levels a side.
 BOOK_PRICE_DATA = ("EX_BEST_OFFERS",)
-"""The only price projection a delayed key serves in full: three levels a side."""
 
+# `listMarketCatalogue` returns at most this many markets, weight permitting.
 MAX_CATALOGUE_RESULTS = 1000
-"""`listMarketCatalogue` returns at most this many markets, weight permitting."""
 
+# Most catalogue requests one listing call makes while filling a page.
 MAX_PAGES = 50
-"""Most catalogue requests one listing call makes while filling a page."""
 
+# The Market Base Rate on most .com markets, used when a market's own
+# `marketBaseRate` was not read.
 DEFAULT_COMMISSION = 0.05
-"""The Market Base Rate on most .com markets, used when a market's own
-`marketBaseRate` was not read."""
 
+# Betfair sorts its catalogue by matched or available money; it has no
+# listing time to sort by.
 SORTS = {"volume": "MAXIMUM_TRADED", "liquidity": "MAXIMUM_AVAILABLE"}
-"""Betfair sorts its catalogue by matched or available money; it has no
-listing time to sort by."""
 
 SESSION_ERRORS = frozenset({"INVALID_SESSION_INFORMATION", "NO_SESSION"})
 AUTH_ERRORS = SESSION_ERRORS | {"INVALID_APP_KEY", "NO_APP_KEY", "ACCESS_DENIED", "APP_KEY_CREATION_FAILED"}
 RATE_ERRORS = frozenset({"TOO_MANY_REQUESTS"})
 
+# Betfair's CLASSIC odds ladder: from each band's start to its end, in its step.
 CLASSIC_LADDER_BANDS: tuple[tuple[str, str, str], ...] = (
     ("1.01", "2", "0.01"), ("2", "3", "0.02"), ("3", "4", "0.05"), ("4", "6", "0.1"), ("6", "10", "0.2"),
     ("10", "20", "0.5"), ("20", "30", "1"), ("30", "50", "2"), ("50", "100", "5"), ("100", "1000", "10"),
 )
-"""Betfair's CLASSIC odds ladder: from each band's start to its end, in its step."""
 
 
 def _classic_ladder() -> tuple[Decimal, ...]:
@@ -160,8 +160,8 @@ def _classic_ladder() -> tuple[Decimal, ...]:
     return tuple(ticks)
 
 
+# Every price Betfair accepts on a CLASSIC market, 1.01 to 1000 (350 ticks).
 CLASSIC_LADDER: tuple[Decimal, ...] = _classic_ladder()
-"""Every price Betfair accepts on a CLASSIC market, 1.01 to 1000 (350 ticks)."""
 _LADDER_SET = frozenset(CLASSIC_LADDER)
 
 
@@ -247,9 +247,9 @@ def raw_levels(rows: Any) -> list[list[float]]:
     return [[row["price"], row["size"]] for row in rows or [] if row and row.get("price") is not None]
 
 
+# A runner's final status as the YES result. A removed runner (a non-runner)
+# has its bets voided. PLACED is the place leg of an each-way market.
 RESULTS: dict[str, str] = {"WINNER": "yes", "LOSER": "no", "PLACED": "yes", "REMOVED": "void"}
-"""A runner's final status as the YES result. A removed runner (a non-runner)
-has its bets voided. PLACED is the place leg of an each-way market."""
 
 
 def status_of(market_status: str | None, runner_status: str | None) -> tuple[str, bool]:

synpath/bucket.py

@@ -30,14 +30,14 @@
 
 
 class BucketMember(_Base):
+    # A Synpath id, `venue:native`.
     market_id: str
-    """A Synpath id, `venue:native`."""
+    # True when this member's YES is the opposite of the bucket's proposition.
     flip: bool = False
-    """True when this member's YES is the opposite of the bucket's proposition."""
+    # False switches the member off: it stays in the definition but is not
+    # routed to, not in the merged bucket book, and not streamed for it.
+    # Definitions stored before this field existed load with every member on.
     enabled: bool = True
-    """False switches the member off: it stays in the definition but is not
-    routed to, not in the merged bucket book, and not streamed for it.
-    Definitions stored before this field existed load with every member on."""
 
     def book_side(self) -> BookSide:
         """Which of the member's two books is the bucket's YES book. Reading