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).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Nsd8HcrV67GbCZXxZEn1Ma
This commit is contained in:
Harwinder Sekhon
2026-08-03 13:11:36 -04:00
parent b936d443b8
commit f0d76021d1
7 changed files with 308 additions and 19 deletions
+4 -4
View File
@@ -11,7 +11,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest
## Commands ## Commands
```bash ```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 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 telegram_bot.py # the bot (refuses to boot without credentials)
python oanda_daemon.py # auto-execution daemon on Oanda practice account 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 ## 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 ## Architecture
@@ -49,12 +49,12 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz
### Layers ### 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. - **`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. - **`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]. - **`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`). - **`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. - **`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. - **`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.
+14 -1
View File
@@ -106,6 +106,7 @@ class AppConfig:
telegram_channel_id: str = "" # "" = channel feed disabled telegram_channel_id: str = "" # "" = channel feed disabled
alert_min_quality: str = "C" alert_min_quality: str = "C"
active_provider: str = "twelve_data" active_provider: str = "twelve_data"
streaming_provider: Optional[str] = None # None = active provider
provider_settings: Dict[str, Dict[str, Any]] = field(default_factory=dict) provider_settings: Dict[str, Dict[str, Any]] = field(default_factory=dict)
scan_pairs: List[str] = field(default_factory=lambda: [XAUUSD]) scan_pairs: List[str] = field(default_factory=lambda: [XAUUSD])
scan_timeframes: List[Timeframe] = field(default_factory=lambda: [ 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.active_provider = str(dp.get("active", app.active_provider))
app.provider_settings = {} app.provider_settings = {}
for name, block in dp.items(): for name, block in dp.items():
if name == "active": # "active" and "streaming" are reserved scalars, not blocks.
if name in ("active", "streaming"):
continue continue
if not isinstance(block, dict): if not isinstance(block, dict):
raise AppConfigError( raise AppConfigError(
@@ -172,6 +174,17 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig:
raise AppConfigError( raise AppConfigError(
f'config.json: data_providers.active "{app.active_provider}" ' f'config.json: data_providers.active "{app.active_provider}" '
"has no matching provider block") "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", {}) scanning = raw.get("scanning", {})
_check_keys(scanning, _KNOWN_SCANNING, '"scanning"') _check_keys(scanning, _KNOWN_SCANNING, '"scanning"')
+1
View File
@@ -7,6 +7,7 @@
}, },
"data_providers": { "data_providers": {
"active": "oanda", "active": "oanda",
"streaming": "tiingo",
"twelve_data": { "twelve_data": {
"api_key": "REPLACE_WITH_YOUR_TWELVE_DATA_API_KEY" "api_key": "REPLACE_WITH_YOUR_TWELVE_DATA_API_KEY"
}, },
+9 -6
View File
@@ -6,12 +6,13 @@ caching, validation, boundary guarantees, and range pagination — the
behavior the signal pipeline depends on). behavior the signal pipeline depends on).
Adding a provider = one module implementing `_fetch_raw` + one registry Adding a provider = one module implementing `_fetch_raw` + one registry
entry + one config.json `data_providers` block. The live feed is selected entry + one config.json `data_providers` block. The live candle feed is
by config.json `data_providers.active`; registration alone changes selected by config.json `data_providers.active`; the optional
nothing. `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 app_config import AppConfigError, get_app_config
from config import PlatformConfig from config import PlatformConfig
@@ -29,8 +30,10 @@ PROVIDER_REGISTRY: Dict[str, type] = {
} }
def make_data_provider(config: PlatformConfig) -> BaseDataProvider: def make_data_provider(config: PlatformConfig,
"""Build the ACTIVE data provider (config.json: data_providers.active).""" 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 name = get_app_config().active_provider
cls = PROVIDER_REGISTRY.get(name) cls = PROVIDER_REGISTRY.get(name)
if cls is None: if cls is None:
+109 -3
View File
@@ -10,15 +10,18 @@ Tiingo's JSON into the normalized frame.
Every failure mode raises `TiingoDataError` (a `DataIntegrityError`) Every failure mode raises `TiingoDataError` (a `DataIntegrityError`)
with a specific message — the scan loops log it and carry on; nothing 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 `/tiingo/fx/top` quote endpoint publishes indicative two-tier spreads
and must not be used for spread display (removed 2026-07-23) — 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 import logging
from datetime import datetime from datetime import datetime
from typing import Optional from typing import AsyncIterator, Callable, Optional, Sequence, Tuple
import pandas as pd import pandas as pd
import requests import requests
@@ -198,3 +201,106 @@ class TiingoDataProvider(BaseDataProvider):
"Fetched %d bars for %s @ %s from Tiingo", "Fetched %d bars for %s @ %s from Tiingo",
len(df), ticker, resample) len(df), ticker, resample)
return df 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]
+34 -3
View File
@@ -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. M15 fires at :00/:15/:30/:45 + 10s buffer.
H1 fires at :00 + 10s buffer. H1 fires at :00 + 10s buffer.
H4 fires at :00 every 4 hours + 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, When price crosses a key structure level (swing S/R, BB,
EMA200) cached from the last boundary scan, triggers an EMA200) cached from the last boundary scan, triggers an
immediate M15 scan and sends a Level Break Alert. immediate M15 scan and sends a Level Break Alert.
@@ -203,6 +205,27 @@ def _get_shared_dp() -> BaseDataProvider:
return _shared_dp 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 formatting ─────────────────────────────────────────────────
_SIGNAL_EMOJI = { _SIGNAL_EMOJI = {
@@ -755,12 +778,19 @@ async def _ws_monitor(bot) -> None:
"""Level-breach tick monitor. The provider owns the connection and """Level-breach tick monitor. The provider owns the connection and
protocol (`iter_ticks`); this loop owns reconnect/backoff and breach protocol (`iter_ticks`); this loop owns reconnect/backoff and breach
dispatch. Providers without streaming quotes simply disable it.""" dispatch. Providers without streaming quotes simply disable it."""
dp = _get_shared_dp() dp = _get_streaming_dp()
if not dp.supports_streaming: if not dp.supports_streaming:
logger.info( logger.info(
"Data provider %r has no streaming quotes — " "Data provider %r has no streaming quotes — "
"level-breach monitor disabled", dp.name) "level-breach monitor disabled", dp.name)
return 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 backoff = 1
while _feed_active(): while _feed_active():
@@ -951,7 +981,8 @@ def _status_text() -> str:
lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)") lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)")
if _ws_task and not _ws_task.done(): 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: else:
lines.append("❌ WebSocket: *disconnected*") lines.append("❌ WebSocket: *disconnected*")
+136 -1
View File
@@ -3001,6 +3001,49 @@ def _():
set_app_config(_TEST_APP) 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") @test("config.py: hardcoded twelve_data_api_key field is gone")
def _(): def _():
assert not hasattr(PlatformConfig(), "twelve_data_api_key"), \ assert not hasattr(PlatformConfig(), "twelve_data_api_key"), \
@@ -3126,6 +3169,55 @@ def _():
tb._autoscan_chats.clear() 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 # MODULE 15: Oanda provider / executor / daemon
# ══════════════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════════════
@@ -4062,7 +4154,7 @@ def _():
def _(): def _():
assert TwelveDataProvider.supports_streaming is True assert TwelveDataProvider.supports_streaming is True
assert _POanda.supports_streaming is False assert _POanda.supports_streaming is False
assert TiingoDataProvider.supports_streaming is False assert TiingoDataProvider.supports_streaming is True
class Minimal(BaseDataProvider): class Minimal(BaseDataProvider):
name = "minimal" name = "minimal"
@@ -4092,6 +4184,49 @@ def _():
assert parse('{"event":"price","symbol":"XAU/USD","price":"abc"}') is None 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") @test("providers: Oanda date-window fetch omits count and sends from/to")
def _(): def _():
captured = {} captured = {}