Tiingo WebSocket streaming: re-enable the level-breach monitor

The monitor has been dead since the Oanda feed switch (Oanda's provider
does not stream). New config.json scalar data_providers.streaming selects
a tick-stream provider independent of the candle feed; telegram_bot
resolves it via _get_streaming_dp() (separate instance, candle cache
stays on the active provider) and falls back to the active provider when
unset — behavior is byte-identical to before when the key is absent.

TiingoDataProvider grows the two sanctioned streaming overrides:
supports_streaming=True and iter_ticks() on wss://api.tiingo.com/fx
(mid price only — the no-spread rule extends to the stream). Two
deliberate divergences from the TD reference implementation, both
because Tiingo authenticates inside the socket: server "E" messages
raise TiingoDataError (a swallowed auth error would end the generator
cleanly and reconnect-hammer with zero backoff), and on_connect fires on
the first received message rather than after the subscribe send. Field
order of quote frames was live-verified: ask is data[7], not data[6] as
the docs table suggests (data[6] is askSize).

make_data_provider(config, name=None) now takes an explicit provider
name; keyless streaming providers disable the monitor with one warning
instead of a traceback-per-minute loop.

Tests 266 → 273 (WS parser, ticker round-trip, streaming config scalar,
explicit-name dispatch, streaming-dp resolution, monitor auto-disable).

Claude-Session: https://claude.ai/code/session_01Nsd8HcrV67GbCZXxZEn1Ma

providers/__init__.py

@@ -6,12 +6,13 @@ caching, validation, boundary guarantees, and range pagination — the
 behavior the signal pipeline depends on).
 
 Adding a provider = one module implementing `_fetch_raw` + one registry
-entry + one config.json `data_providers` block. The live feed is selected
-by config.json `data_providers.active`; registration alone changes
-nothing.
+entry + one config.json `data_providers` block. The live candle feed is
+selected by config.json `data_providers.active`; the optional
+`data_providers.streaming` scalar selects a different provider for the
+tick stream (level-breach monitor). Registration alone changes nothing.
 """
 
-from typing import Dict
+from typing import Dict, Optional
 
 from app_config import AppConfigError, get_app_config
 from config import PlatformConfig
@@ -29,9 +30,11 @@ PROVIDER_REGISTRY: Dict[str, type] = {
 }
 
 
-def make_data_provider(config: PlatformConfig) -> BaseDataProvider:
-    """Build the ACTIVE data provider (config.json: data_providers.active)."""
-    name = get_app_config().active_provider
+def make_data_provider(config: PlatformConfig,
+                       name: Optional[str] = None) -> BaseDataProvider:
+    """Build a provider by name (default: config.json data_providers.active)."""
+    if name is None:
+        name = get_app_config().active_provider
     cls = PROVIDER_REGISTRY.get(name)
     if cls is None:
         raise AppConfigError(

providers/tiingo.py

@@ -10,15 +10,18 @@ Tiingo's JSON into the normalized frame.
 
 Every failure mode raises `TiingoDataError` (a `DataIntegrityError`)
 with a specific message — the scan loops log it and carry on; nothing
-fails silently. NOTE: this is the CANDLE feed only. Tiingo's free
+fails silently. NOTE: candles and mid-price ticks only. Tiingo's free
 `/tiingo/fx/top` quote endpoint publishes indicative two-tier spreads
 and must not be used for spread display (removed 2026-07-23) —
-`get_spread` deliberately stays on the base-class default.
+`get_spread` deliberately stays on the base-class default, and the
+WebSocket stream (`iter_ticks`) exposes the mid price only, never
+bid/ask or a derived spread.
 """
 
+import json
 import logging
 from datetime import datetime
-from typing import Optional
+from typing import AsyncIterator, Callable, Optional, Sequence, Tuple
 
 import pandas as pd
 import requests
@@ -198,3 +201,106 @@ class TiingoDataProvider(BaseDataProvider):
             "Fetched %d bars for %s @ %s from Tiingo",
             len(df), ticker, resample)
         return df
+
+    # ── Streaming quotes ───────────────────────────────────────────
+
+    supports_streaming = True
+
+    WS_URL = "wss://api.tiingo.com/fx"
+
+    @staticmethod
+    def _parse_ws_message(raw: str) -> Optional[Tuple[str, float]]:
+        """Parse one WS message; return (tiingo_ticker, mid_price) for
+        quote updates, None for heartbeats ("H"), subscription acks
+        ("I"), malformed JSON, or quotes without a usable price.
+
+        Server error messages ("E", e.g. bad token) RAISE TiingoDataError:
+        Tiingo authenticates inside the socket via the subscribe message,
+        so an auth failure arrives as a data message followed by a clean
+        close — swallowing it would make the caller's reconnect loop
+        hammer the endpoint with no backoff (backoff only grows on
+        exceptions).
+        """
+        try:
+            msg = json.loads(raw)
+        except json.JSONDecodeError:
+            logger.warning("Tiingo WS: malformed JSON, skipping")
+            return None
+
+        mtype = msg.get("messageType")
+        if mtype == "H":
+            return None
+        if mtype == "I":
+            logger.info("Tiingo WS: subscription acknowledged")
+            return None
+        if mtype == "E":
+            raise TiingoDataError(
+                f"Tiingo WS server error: {str(msg.get('response'))[:200]}")
+        if mtype != "A":
+            return None
+
+        data = msg.get("data")
+        if not isinstance(data, list) or len(data) < 8 or data[0] != "Q":
+            return None
+        # Live-verified field order (2026-08-03): ["Q", ticker, timestamp,
+        # bidSize, bid, mid, askSize, ask] — ask is index 7, NOT 6 as the
+        # docs' table suggests (index 6 is askSize).
+        ticker = data[1]
+        bid, mid, ask = data[4], data[5], data[7]
+        if mid is None and bid is not None and ask is not None:
+            mid = (bid + ask) / 2
+        try:
+            return ticker, float(mid)
+        except (TypeError, ValueError):
+            logger.warning("Tiingo WS: quote for %s has no usable price",
+                           ticker)
+            return None
+
+    async def iter_ticks(
+        self,
+        symbols: Sequence[str],
+        on_connect: Optional[Callable[[], None]] = None,
+    ) -> AsyncIterator[Tuple[str, float]]:
+        """One connection's lifetime of (canonical_symbol, mid) ticks.
+
+        The caller owns the reconnect/backoff loop around this generator.
+        Divergences from the Twelve Data reference implementation, both
+        deliberate: `on_connect` fires on the FIRST RECEIVED message (the
+        "I" ack in practice) rather than after the subscribe send, so a
+        rejected token cannot reset the caller's backoff; and there is no
+        client heartbeat task — Tiingo sends "H" frames itself and the
+        websockets library's built-in ping keepalive detects dead sockets.
+        """
+        import websockets
+
+        if self._api_key is None:
+            raise TiingoDataError(
+                "No Tiingo API key. Set the TIINGO_API_KEY env var or "
+                "data_providers.tiingo.api_key in config.json.")
+
+        ws_to_pair = {self._to_tiingo_ticker(s): s for s in symbols}
+        async with websockets.connect(self.WS_URL) as ws:
+            await ws.send(json.dumps({
+                "eventName": "subscribe",
+                "authorization": self._api_key,
+                "eventData": {
+                    "thresholdLevel": 5,
+                    "tickers": sorted(ws_to_pair),
+                },
+            }))
+            logger.info(
+                "Tiingo WS connected, subscribing to %d tickers",
+                len(ws_to_pair))
+            connected = False
+            async for raw in ws:
+                parsed = self._parse_ws_message(raw)
+                if not connected:
+                    connected = True
+                    if on_connect is not None:
+                        on_connect()
+                if parsed is None:
+                    continue
+                pair_symbol = ws_to_pair.get(parsed[0])
+                if pair_symbol is None:
+                    continue
+                yield pair_symbol, parsed[1]