From f7d8f1113cbd4cb1658f06d0b4d8944e2d3f84fa Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Mon, 3 Aug 2026 13:11:36 -0400 Subject: [PATCH] Tiingo WebSocket streaming: re-enable the level-breach monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CLAUDE.md | 8 +-- app_config.py | 15 ++++- config.json.example | 1 + providers/__init__.py | 17 +++--- providers/tiingo.py | 112 +++++++++++++++++++++++++++++++++- telegram_bot.py | 37 +++++++++++- test_all.py | 137 +++++++++++++++++++++++++++++++++++++++++- 7 files changed, 308 insertions(+), 19 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 8b550b7..10f1a01 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -11,7 +11,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest ## Commands ```bash -python test_all.py # QA suite: 266 tests, 18 modules, no network (mocked) +python test_all.py # QA suite: 273 tests, 18 modules, no network (mocked) python main.py --pair XAUUSD --timeframe H1 --detail # CLI signal report (--json for full dump) python telegram_bot.py # the bot (refuses to boot without credentials) python oanda_daemon.py # auto-execution daemon on Oanda practice account @@ -30,7 +30,7 @@ Python 3.12; a `venv/` exists in-repo. Dependencies in `requirements.txt` (numpy ## Configuration & credentials -Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_ACCOUNT_ID`/`OANDA_API_TOKEN`/`OANDA_BASE_URL`. +Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `data_providers.streaming` (optional tick-stream provider for the level-breach monitor), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_ACCOUNT_ID`/`OANDA_API_TOKEN`/`OANDA_BASE_URL`. ## Architecture @@ -49,12 +49,12 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz ### Layers -- **`providers/`** — one class per provider subclassing `BaseDataProvider` (`providers/base.py`), which owns everything the pipeline depends on: TTL+boundary cache, forming-candle drop, freshness check, validate/clean, paginated `fetch_range` (feeds the backtester from any provider), optional tick streaming. Subclasses implement only `_fetch_raw` + symbol/interval maps. Registry + `make_data_provider` in `providers/__init__.py`; config.json `data_providers.active` selects the live feed. A test enforces that subclasses don't override shared base methods. Tiingo is candles-only — do not re-add spread display from its free FX feed (two-tier indicative quotes, removed 2026-07-23). +- **`providers/`** — one class per provider subclassing `BaseDataProvider` (`providers/base.py`), which owns everything the pipeline depends on: TTL+boundary cache, forming-candle drop, freshness check, validate/clean, paginated `fetch_range` (feeds the backtester from any provider), optional tick streaming. Subclasses implement only `_fetch_raw` + symbol/interval maps. Registry + `make_data_provider(config, name=None)` in `providers/__init__.py`; config.json `data_providers.active` selects the live candle feed and the optional `data_providers.streaming` scalar selects a different provider for the tick stream. A test enforces that subclasses don't override shared base methods (`supports_streaming`/`iter_ticks` are the sanctioned streaming override points). Tiingo streams mid-price ticks (`wss://api.tiingo.com/fx`) — do not re-add spread display from its free FX feed (two-tier indicative quotes, removed 2026-07-23); the no-spread rule extends to the WS stream. - **`engine.py`** — `TechnicalEngine`: 9 scored indicators (ema_trend, rsi, macd, bollinger, adx, roc, keltner, divergence, candlestick) + ATR as context. Regime classification (trending_up/down, `trending` = mixed/direction-UNKNOWN, transitioning, ranging) drives per-indicator weight multipliers and regime-aware RSI reinterpretation. Compute order matters (ATR first; MACD always computed even when not a pillar — divergence needs the histogram). Private series ride on the result as `_underscore` attrs; scorers/formatters skip anything `_`-prefixed or weight-0. - **`signal_aggregator.py`** — MTF agreement, diminishing-decay penalty system, exhaustion hard block, quality grading, then an ordered chain of post-grade gates (session, XAU SELL quality/vol/hour, regime blocks, M5 noise gates), then three setup generators: XAU ATR (A+/A), generic swing-based, plus the M5 pullback (`XAU_PB`) and chart-pattern (`XAU_CP`, `chart_patterns.py`) entry paths. `TradeSetup.strategy` selects trailing params; trail params are stamped on the setup from the *producing* config so per-TF overrides reach formatters. - **`pair_signals/`** — per-pair config factories. **Always use the factory (`produce_signal` / `make_config`), never bare `PlatformConfig()`** — the factory applies the config.json overlay first, then the validated per-TF deltas (M5 hour-9 block, M15 trail arm 0.70, M15 SELL-hour block, etc.; evidence in the docstring). A test enforces factory ≡ inline-config equivalence. Confirmation map: M5→[M15,H1], M15→[M5,H1], H1→[M15,H4], H4→[H1,D1]. - **`news_engine.py`** — Forex Factory calendar (MyFXBook RSS fallback) + RSS keyword sentiment with gold-inverse keywords. Class-level caches (fresh engine per scan). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`). -- **`telegram_bot.py`** — owner-locked bot + subscriber channel; candle-boundary auto-scan loops (M5/M15/H1/H4, boundary+10 s), alert quality threshold + fingerprint dedup (committed only after a successful send), WebSocket level-breach monitor (auto-disabled when the active provider has `supports_streaming=False` — Oanda does not stream). Rotating log: `logs/telegram.log`, 3-h rotation × 240 backups. +- **`telegram_bot.py`** — owner-locked bot + subscriber channel; candle-boundary auto-scan loops (M5/M15/H1/H4, boundary+10 s), alert quality threshold + fingerprint dedup (committed only after a successful send), WebSocket level-breach monitor fed by the streaming provider (`data_providers.streaming`, falling back to the active provider; auto-disabled when the resolved provider can't stream or has no API key — Oanda does not stream, Tiingo does). Rotating log: `logs/telegram.log`, 3-h rotation × 240 backups. - **`oanda_daemon.py` / `oanda_executor.py`** — practice-account auto-execution. Book-parity rules (see daemon docstring): no broker-side TP (trailing model — TP1/TP2 advisory), 5-bar per-TF cooldown, trail ratchets at bar close exactly once per bar. Accepted divergence: one position total. - **`backtest/gpu_backtest.py`** — bar-by-bar replay through the real `analyze_from_dataframes` (news off). Realism model (do not regress): next-bar-open entries re-anchored to the raw price path, round-trip `--spread` charged per trade, trade resolution on RAW bars (cleaning applies only to signal-side data), same-bar SL+TP tie → SL first, gap fills at bar open, trail ratchet at bar close only. Indicators are precomputed once per TF (CuPy GPU or NumPy fallback — check the `Backend:` log line); replay fans out over worker processes and the 5-bar cooldown is applied in an ordered pass afterwards, so parallel and sequential books are identical. Parquet cache in `backtest/cache/` keyed by provider/symbol/TF/window. `backtest/*.csv` are gitignored pinned trade books from past campaigns — used for manual before/after comparison. `backtest/instructions_for_backtest.txt` lists what every backtest report must record. diff --git a/app_config.py b/app_config.py index 1f68f0c..72e639b 100644 --- a/app_config.py +++ b/app_config.py @@ -106,6 +106,7 @@ class AppConfig: telegram_channel_id: str = "" # "" = channel feed disabled alert_min_quality: str = "C" active_provider: str = "twelve_data" + streaming_provider: Optional[str] = None # None = active provider provider_settings: Dict[str, Dict[str, Any]] = field(default_factory=dict) scan_pairs: List[str] = field(default_factory=lambda: [XAUUSD]) scan_timeframes: List[Timeframe] = field(default_factory=lambda: [ @@ -162,7 +163,8 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig: app.active_provider = str(dp.get("active", app.active_provider)) app.provider_settings = {} for name, block in dp.items(): - if name == "active": + # "active" and "streaming" are reserved scalars, not blocks. + if name in ("active", "streaming"): continue if not isinstance(block, dict): raise AppConfigError( @@ -172,6 +174,17 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig: raise AppConfigError( f'config.json: data_providers.active "{app.active_provider}" ' "has no matching provider block") + streaming = dp.get("streaming") + if streaming is not None: + if not isinstance(streaming, str): + raise AppConfigError( + 'config.json: data_providers.streaming must be a string ' + "(a provider name)") + if streaming not in app.provider_settings: + raise AppConfigError( + f'config.json: data_providers.streaming "{streaming}" ' + "has no matching provider block") + app.streaming_provider = streaming scanning = raw.get("scanning", {}) _check_keys(scanning, _KNOWN_SCANNING, '"scanning"') diff --git a/config.json.example b/config.json.example index 09e4e8d..785607a 100644 --- a/config.json.example +++ b/config.json.example @@ -7,6 +7,7 @@ }, "data_providers": { "active": "oanda", + "streaming": "tiingo", "twelve_data": { "api_key": "REPLACE_WITH_YOUR_TWELVE_DATA_API_KEY" }, diff --git a/providers/__init__.py b/providers/__init__.py index 5a60083..39f9ad3 100644 --- a/providers/__init__.py +++ b/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( diff --git a/providers/tiingo.py b/providers/tiingo.py index 14fbcfe..797fc6f 100644 --- a/providers/tiingo.py +++ b/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] diff --git a/telegram_bot.py b/telegram_bot.py index 2ba0f06..802c237 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -12,7 +12,9 @@ Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set. M15 fires at :00/:15/:30/:45 + 10s buffer. H1 fires at :00 + 10s buffer. H4 fires at :00 every 4 hours + 10s buffer. -WebSocket: Streams live ticks from TwelveData for XAU/USD. +WebSocket: Streams live ticks from the streaming provider — config.json + `data_providers.streaming` when set (e.g. tiingo), else the + active provider if it streams; auto-disabled otherwise. When price crosses a key structure level (swing S/R, BB, EMA200) cached from the last boundary scan, triggers an immediate M15 scan and sends a Level Break Alert. @@ -203,6 +205,27 @@ def _get_shared_dp() -> BaseDataProvider: return _shared_dp +_streaming_dp: Optional[BaseDataProvider] = None + + +def _get_streaming_dp() -> BaseDataProvider: + """Provider for the tick stream: config.json data_providers.streaming + when set (a SEPARATE instance — the candle cache stays on the active + provider), else the shared active provider. Reads app config at call + time, not the import-time _APP snapshot, so tests can swap it via + set_app_config().""" + global _streaming_dp + if _streaming_dp is None: + app = get_app_config() + if app.streaming_provider is None or \ + app.streaming_provider == app.active_provider: + _streaming_dp = _get_shared_dp() + else: + _streaming_dp = make_data_provider( + PlatformConfig(), app.streaming_provider) + return _streaming_dp + + # ── Signal formatting ───────────────────────────────────────────────── _SIGNAL_EMOJI = { @@ -755,12 +778,19 @@ async def _ws_monitor(bot) -> None: """Level-breach tick monitor. The provider owns the connection and protocol (`iter_ticks`); this loop owns reconnect/backoff and breach dispatch. Providers without streaming quotes simply disable it.""" - dp = _get_shared_dp() + dp = _get_streaming_dp() if not dp.supports_streaming: logger.info( "Data provider %r has no streaming quotes — " "level-breach monitor disabled", dp.name) return + if getattr(dp, "_api_key", "unused") is None: + # Guard, not a traceback loop: without a key every connect + # attempt would raise and re-log a full traceback every ≤60s. + logger.warning( + "Streaming provider %r has no API key — " + "level-breach monitor disabled", dp.name) + return backoff = 1 while _feed_active(): @@ -951,7 +981,8 @@ def _status_text() -> str: lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)") if _ws_task and not _ws_task.done(): - lines.append("📡 WebSocket: *connected*") + provider = _streaming_dp.name if _streaming_dp else "?" + lines.append(f"📡 WebSocket: *connected* ({provider})") else: lines.append("❌ WebSocket: *disconnected*") diff --git a/test_all.py b/test_all.py index 37deef3..d552584 100644 --- a/test_all.py +++ b/test_all.py @@ -3001,6 +3001,49 @@ def _(): set_app_config(_TEST_APP) +@test("providers: make_data_provider explicit name overrides active") +def _(): + from providers import (TiingoDataProvider as _Tii, + TwelveDataProvider, make_data_provider) + # active is twelve_data under _TEST_APP; explicit name wins + with patch.dict(os.environ, {"TIINGO_API_KEY": "k"}): + assert isinstance(make_data_provider(PlatformConfig(), "tiingo"), _Tii) + assert isinstance(make_data_provider(PlatformConfig()), TwelveDataProvider) + try: + make_data_provider(PlatformConfig(), "polygon") + assert False, "unregistered explicit name must raise" + except AppConfigError as e: + assert "polygon" in str(e) + + +@test("app_config: data_providers.streaming scalar selects the tick stream") +def _(): + app = load_app_config(_write_json({ + "data_providers": {"active": "oanda", "streaming": "tiingo", + "oanda": {}, "tiingo": {"api_key": "k"}}})) + assert app.streaming_provider == "tiingo" + assert set(app.provider_settings) == {"oanda", "tiingo"}, \ + "reserved scalars must not become provider blocks" + app = load_app_config(_write_json( + {"data_providers": {"active": "oanda", "oanda": {}}})) + assert app.streaming_provider is None + + +@test("app_config: streaming provider must be a string with a settings block") +def _(): + for payload in ( + {"data_providers": {"active": "oanda", "oanda": {}, + "streaming": "polygon"}}, # no block + {"data_providers": {"active": "oanda", "oanda": {}, + "streaming": 5}}, # not a string + ): + try: + load_app_config(_write_json(payload)) + assert False, f"should have raised for {payload}" + except AppConfigError: + pass + + @test("config.py: hardcoded twelve_data_api_key field is gone") def _(): assert not hasattr(PlatformConfig(), "twelve_data_api_key"), \ @@ -3126,6 +3169,55 @@ def _(): tb._autoscan_chats.clear() +@test("telegram_bot: _get_streaming_dp resolves streaming name, else shared") +def _(): + import telegram_bot as tb + from providers import TiingoDataProvider as _Tii + set_app_config(AppConfig( + active_provider="twelve_data", + streaming_provider="tiingo", + provider_settings={"twelve_data": {"api_key": "k"}, + "tiingo": {"api_key": "k"}})) + tb._streaming_dp = None + tb._shared_dp = None + try: + dp = tb._get_streaming_dp() + assert isinstance(dp, _Tii) + assert dp is not tb._get_shared_dp(), \ + "streaming provider must be a separate instance" + # unset → falls back to the shared active provider + set_app_config(_TEST_APP) + tb._streaming_dp = None + tb._shared_dp = None + assert tb._get_streaming_dp() is tb._get_shared_dp() + finally: + set_app_config(_TEST_APP) + tb._streaming_dp = None + tb._shared_dp = None + + +@test("telegram_bot: ws monitor auto-disables on non-streaming or keyless") +def _(): + import asyncio + import telegram_bot as tb + + class NoStream: + name = "stub" + supports_streaming = False + + class KeylessStream: + name = "stub" + supports_streaming = True + _api_key = None + + try: + for stub in (NoStream(), KeylessStream()): + tb._streaming_dp = stub + asyncio.run(tb._ws_monitor(None)) # must return, not connect + finally: + tb._streaming_dp = None + + # ══════════════════════════════════════════════════════════════════════ # MODULE 15: Oanda provider / executor / daemon # ══════════════════════════════════════════════════════════════════════ @@ -4062,7 +4154,7 @@ def _(): def _(): assert TwelveDataProvider.supports_streaming is True assert _POanda.supports_streaming is False - assert TiingoDataProvider.supports_streaming is False + assert TiingoDataProvider.supports_streaming is True class Minimal(BaseDataProvider): name = "minimal" @@ -4092,6 +4184,49 @@ def _(): assert parse('{"event":"price","symbol":"XAU/USD","price":"abc"}') is None +@test("providers: Tiingo WS quote-message parsing") +def _(): + parse = TiingoDataProvider._parse_ws_message + # Field order matches a live-captured frame (2026-08-03): + # ["Q", ticker, ts, bidSize, bid, mid, askSize, ask] + quote = ('{"service":"fx","messageType":"A","data":' + '["Q","xauusd","2026-08-03T17:09:46.084000+00:00",' + '1000000.0,4035.05,4035.1,1000000.0,4035.15]}') + assert parse(quote) == ("xauusd", 4035.1) + # mid null → (bid+ask)/2, ask read from index 7 (index 6 is askSize) + mid_null = ('{"messageType":"A","service":"fx","data":' + '["Q","xauusd","2026-08-03T10:00:00+00:00",' + '1000000,3344.0,null,1000000,3346.0]}') + assert parse(mid_null) == ("xauusd", 3345.0) + # mid and bid null → unusable, skipped + all_null = ('{"messageType":"A","service":"fx","data":' + '["Q","xauusd","2026-08-03T10:00:00+00:00",' + '1000000,null,null,1000000,3346.0]}') + assert parse(all_null) is None + assert parse('{"messageType":"H"}') is None # heartbeat + assert parse('{"messageType":"I","response":{"code":200}}') is None + assert parse('not json{{') is None + assert parse('{"messageType":"A","data":["X","xauusd"]}') is None + # server error (e.g. bad token) must RAISE so the caller's backoff + # applies — Tiingo auths inside the socket, not at the HTTP upgrade + try: + parse('{"messageType":"E","response":' + '{"code":401,"message":"bad token"}}') + assert False, "server error message must raise" + except TiingoDataError: + pass + + +@test("providers: Tiingo WS ticker mapping round-trips the scanned pairs") +def _(): + tp = TiingoDataProvider.__new__(TiingoDataProvider) + for pair in PAIR_INFO: + ticker = tp._to_tiingo_ticker(pair) + assert ticker.islower() and "/" not in ticker + inverse = {tp._to_tiingo_ticker(s): s for s in PAIR_INFO} + assert inverse[ticker] == pair + + @test("providers: Oanda date-window fetch omits count and sends from/to") def _(): captured = {}