Harwinder Sekhon
f0d76021d1
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
307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""
|
||
ForexBot2 — Tiingo data provider
|
||
=================================
|
||
OHLCV candles from Tiingo's FX endpoint
|
||
(`/tiingo/fx/{ticker}/prices`, resampled mid-price bars, no volume).
|
||
All shared behavior (cache, validation, boundary guarantees, range
|
||
pagination) lives in `BaseDataProvider`; this module only maps
|
||
symbols/resample frequencies, sizes the request window, and parses
|
||
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: 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, 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 AsyncIterator, Callable, Optional, Sequence, Tuple
|
||
|
||
import pandas as pd
|
||
import requests
|
||
|
||
from app_config import provider_api_key
|
||
from config import PlatformConfig, Timeframe
|
||
|
||
from providers.base import BaseDataProvider, DataIntegrityError
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class TiingoDataError(DataIntegrityError):
|
||
"""Raised when a Tiingo data fetch fails or returns invalid data."""
|
||
|
||
|
||
class TiingoDataProvider(BaseDataProvider):
|
||
"""Tiingo FX REST candle provider."""
|
||
|
||
name = "tiingo"
|
||
DATA_ERROR = TiingoDataError
|
||
|
||
BASE_URL = "https://api.tiingo.com"
|
||
|
||
TIINGO_RESAMPLE = {
|
||
Timeframe.M1: "1min",
|
||
Timeframe.M5: "5min",
|
||
Timeframe.M15: "15min",
|
||
Timeframe.M30: "30min",
|
||
Timeframe.H1: "1hour",
|
||
Timeframe.H4: "4hour",
|
||
Timeframe.D1: "1day",
|
||
# W1 intentionally unmapped: the FX endpoint has no weekly
|
||
# resample — requesting it raises loudly instead of guessing.
|
||
}
|
||
|
||
TIINGO_TICKER = {
|
||
"XAU/USD": "xauusd",
|
||
}
|
||
|
||
# Window padding over the theoretical bar span: FX is closed ~2/7 of
|
||
# the week (Fri 22:00 → Sun 22:00 UTC), so a calendar window must be
|
||
# wider than `bars × candle_seconds` to yield the requested count.
|
||
WINDOW_PAD_FACTOR = 1.6
|
||
|
||
def __init__(self, config: PlatformConfig, api_key: Optional[str] = None):
|
||
super().__init__(config)
|
||
# Credentials default to env TIINGO_API_KEY → config.json
|
||
# data_providers.tiingo.api_key, so the provider registry can
|
||
# build this class with config only.
|
||
if api_key is None:
|
||
api_key = provider_api_key(provider="tiingo")
|
||
self._api_key = api_key or None
|
||
if self._api_key is None:
|
||
logger.warning(
|
||
"Tiingo provider built WITHOUT an API key — every fetch "
|
||
"will fail with a named error. Set the TIINGO_API_KEY env "
|
||
"var or data_providers.tiingo.api_key in config.json.")
|
||
self._session = requests.Session()
|
||
self._session.headers.update({
|
||
"Content-Type": "application/json",
|
||
# Header auth keeps the key out of URLs (and out of logs).
|
||
"Authorization": f"Token {api_key}" if api_key else "",
|
||
})
|
||
|
||
def _to_tiingo_ticker(self, symbol: str) -> str:
|
||
if symbol in self.TIINGO_TICKER:
|
||
return self.TIINGO_TICKER[symbol]
|
||
return symbol.replace("/", "").replace("=X", "").lower()
|
||
|
||
# ── Raw fetch ──────────────────────────────────────────────────
|
||
|
||
def _fetch_raw(
|
||
self,
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
outputsize: int,
|
||
start: Optional[datetime] = None,
|
||
end: Optional[datetime] = None,
|
||
) -> pd.DataFrame:
|
||
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.")
|
||
|
||
resample = self.TIINGO_RESAMPLE.get(timeframe)
|
||
if resample is None:
|
||
raise TiingoDataError(
|
||
f"Unsupported timeframe {timeframe} for Tiingo "
|
||
f"(no FX resample frequency)")
|
||
|
||
ticker = self._to_tiingo_ticker(symbol)
|
||
|
||
# Tiingo takes date windows, not bar counts: derive the start
|
||
# from the requested count for live fetches.
|
||
if start is None:
|
||
span = self.CANDLE_SECONDS[timeframe] * (outputsize + 10)
|
||
start_ts = (pd.Timestamp.now(tz="UTC")
|
||
- pd.Timedelta(seconds=span * self.WINDOW_PAD_FACTOR))
|
||
else:
|
||
start_ts = pd.Timestamp(start)
|
||
|
||
params = {
|
||
"resampleFreq": resample,
|
||
"startDate": start_ts.strftime("%Y-%m-%dT%H:%M:%S"),
|
||
}
|
||
if end is not None:
|
||
params["endDate"] = pd.Timestamp(end).strftime("%Y-%m-%dT%H:%M:%S")
|
||
|
||
url = f"{self.BASE_URL}/tiingo/fx/{ticker}/prices"
|
||
try:
|
||
resp = self._session.get(url, params=params, timeout=30)
|
||
except requests.RequestException as e:
|
||
raise TiingoDataError(
|
||
f"Tiingo candle fetch failed for {ticker}: {e}") from e
|
||
|
||
if resp.status_code == 429:
|
||
raise TiingoDataError(
|
||
f"Tiingo rate limit hit for {ticker} (HTTP 429) — "
|
||
f"plan quota exhausted; scan will retry next boundary")
|
||
if resp.status_code in (401, 403):
|
||
raise TiingoDataError(
|
||
f"Tiingo auth failed for {ticker} (HTTP {resp.status_code}) "
|
||
f"— check TIINGO_API_KEY / config.json api_key")
|
||
if resp.status_code != 200:
|
||
raise TiingoDataError(
|
||
f"Tiingo candle fetch failed for {ticker}: "
|
||
f"HTTP {resp.status_code}: {resp.text[:300]}")
|
||
|
||
try:
|
||
data = resp.json()
|
||
except ValueError as e:
|
||
raise TiingoDataError(
|
||
f"Tiingo candle response for {ticker} is not valid "
|
||
f"JSON: {e}") from e
|
||
|
||
if not isinstance(data, list) or not data:
|
||
raise TiingoDataError(
|
||
f"Empty candle response from Tiingo for {ticker} "
|
||
f"(window {params['startDate']} → "
|
||
f"{params.get('endDate', 'now')}; check ticker and plan "
|
||
f"history limits)")
|
||
|
||
rows = []
|
||
for item in data:
|
||
if "date" not in item:
|
||
raise TiingoDataError(
|
||
f"Malformed Tiingo candle for {ticker}: missing "
|
||
f"'date' in {str(item)[:120]}")
|
||
rows.append({
|
||
"Datetime": pd.Timestamp(item["date"]),
|
||
"Open": item.get("open"),
|
||
"High": item.get("high"),
|
||
"Low": item.get("low"),
|
||
"Close": item.get("close"),
|
||
# FX endpoint has no volume; 0.0 keeps the engine contract.
|
||
"Volume": 0.0,
|
||
})
|
||
|
||
df = pd.DataFrame(rows)
|
||
df = df.set_index("Datetime")
|
||
df = df.sort_index()
|
||
if df.index.tz is None:
|
||
df.index = df.index.tz_localize("UTC")
|
||
else:
|
||
df.index = df.index.tz_convert("UTC")
|
||
df.index.name = "Datetime"
|
||
|
||
for col in ("Open", "High", "Low", "Close"):
|
||
df[col] = pd.to_numeric(df[col], errors="coerce")
|
||
if df["Close"].isna().all():
|
||
raise TiingoDataError(
|
||
f"Tiingo response for {ticker} has no numeric closes — "
|
||
f"feed format changed?")
|
||
|
||
logger.debug(
|
||
"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]
|