""" APEX V18 — TradingView market-data provider. Concrete MarketDataProvider impl backed by tvdatafeed-enhanced (StreamAlpha fork), used as primary feed for indicators when env USE_TV_FEED=1. Why TV instead of BrokerMarketDataProvider: - TopstepX bars on 4hour timeframe are sparse/gappy; TV has clean H4. - TV history depth covers calibration windows; TopstepX is intraday-only. - Decouples indicator pipeline from broker connectivity (broker can be down for orders without losing technical reads). Wiring (main.py): opt-in via env USE_TV_FEED=1 + TV_USERNAME / TV_PASSWORD. Auth optional — anonymous fallback works for shorter histories. Failure modes (network down, auth expired, symbol unmapped, TV returns None / empty) -> empty DataFrame matching the V16 contract, warning logged. Caller skips the iteration (same contract as Fake/Broker). DataFrame contract (consumer-side, see analysis.market_data docstring): columns [open, high, low, close, volume] + 'time' column (UTC), sorted ascending, last row = most recent bar. """ from __future__ import annotations import asyncio import json import logging import os from pathlib import Path from typing import Any, Callable import pandas as pd logger = logging.getLogger(__name__) # ============================================================ # Token cache file (used to inject TV_TOKEN into the lib's auth path) # ============================================================ # tvdatafeed-enhanced reads a cached token via _load_token() before # attempting username/password login. We pre-write the token from # TV_TOKEN env var so the lib skips the CAPTCHA-prone signin flow # entirely. Default location is APEX-scoped (not the lib's # ~/.tv_token.json) so we don't clobber other tools' caches. DEFAULT_TOKEN_CACHE = Path("~/.apex_v18/tv_token.json").expanduser() # ============================================================ # Symbol + timeframe mappings # ============================================================ # V16 root symbol -> (TV ticker, TV exchange). # fut_contract=1 (front-month continuous) is passed to get_hist, so # passing "MES" + "CME_MINI" resolves to MES1! on TradingView. TV_SYMBOL_MAP: dict[str, tuple[str, str]] = { "MES": ("MES", "CME_MINI"), "MNQ": ("MNQ", "CME_MINI"), "MYM": ("MYM", "CBOT_MINI"), "MGC": ("MGC", "COMEX"), "MCL": ("MCL", "NYMEX"), "6E": ("6E", "CME"), "6B": ("6B", "CME"), "6A": ("6A", "CME"), "6J": ("6J", "CME"), "6C": ("6C", "CME"), } # V16 timeframe string -> tvDatafeed.Interval. Built lazily so the module # imports cleanly even if tvdatafeed-enhanced isn't installed (tests can # run with a fake client_factory). _TV_INTERVAL_KEYS = ("5min", "1hour", "4hour") def _build_interval_map() -> dict[str, Any]: from tvDatafeed import Interval return { "5min": Interval.in_5_minute, "1hour": Interval.in_1_hour, "4hour": Interval.in_4_hour, } # ============================================================ # Provider # ============================================================ class TVDataFeedProvider: """ MarketDataProvider impl that fetches bars from TradingView via tvdatafeed-enhanced. Auth modes (preference order): 1. token: TV auth_token extracted from browser. Pre-written to token_cache_file; lib's _load_token() picks it up and skips the username/password login flow. Survives CAPTCHA. 2. username + password: lib attempts POST /signin. Subject to TradingView CAPTCHA challenges on datacenter IPs (the lib falls back to interactive browser-token paste — unsuitable for unattended runtime). Use only on residential IPs. 3. None: anonymous mode. ~5000 bars/req cap, stricter rate limits, but sufficient for the V16 200-bar fetches. Args: username: TV account username, or None for anonymous access. password: TV account password, or None for anonymous access. token: TV auth_token (JWT). Pre-loaded into cache so the lib uses it instead of attempting interactive login. token_cache_file: where to pre-write the token. Defaults to ~/.apex_v18/tv_token.json (NOT the lib's ~/.tv_token.json, to avoid clobbering caches from other TV tools). symbol_map: override TV_SYMBOL_MAP (used by tests to inject unmapped/edge symbols). client_factory: callable (**kwargs) -> TV client. Receives username, password, token, token_cache_file. Tests inject a fake. interval_map_builder: callable () -> {timeframe: Interval}. Default: lazy import from tvdatafeed. Tests bypass the tvdatafeed import by passing a fake builder. The TV client is constructed lazily on the first get_bars call and cached for the provider lifetime (TvDatafeed maintains a websocket session internally; one instance per provider is correct). """ def __init__( self, username: str | None = None, password: str | None = None, *, token: str | None = None, token_cache_file: str | Path | None = None, symbol_map: dict[str, tuple[str, str]] | None = None, client_factory: Callable[..., Any] | None = None, interval_map_builder: Callable[[], dict[str, Any]] | None = None, ) -> None: self.username = username self.password = password self.token = token self.token_cache_file = ( Path(token_cache_file).expanduser() if token_cache_file else DEFAULT_TOKEN_CACHE ) self.symbol_map = symbol_map if symbol_map is not None else TV_SYMBOL_MAP self._client: Any = None self._client_factory = client_factory self._interval_map_builder = interval_map_builder or _build_interval_map self._interval_map: dict[str, Any] | None = None def _get_client(self) -> Any: if self._client is not None: return self._client # If TV_TOKEN provided, pre-write to cache file so the lib's # _load_token() picks it up. Done unconditionally (also when # client_factory is set) so tests can verify the cache write. if self.token: self._write_token_cache() if self._client_factory is not None: self._client = self._client_factory( username=self.username, password=self.password, token=self.token, token_cache_file=str(self.token_cache_file), ) else: from tvDatafeed import TvDatafeed if self.token: # Anonymous constructor + cache-loaded token. Lib reads # the cache via _load_token() and validates JWT exp; if # expired/invalid, lib silently falls back to anonymous. self._client = TvDatafeed( token_cache_file=str(self.token_cache_file), ) if self._client.token != self.token: logger.warning( "TVDataFeed: TV_TOKEN rejected by validator " "(likely expired) — falling back to anonymous" ) elif self.username and self.password: # Will hit CAPTCHA on most server IPs; the lib falls # back to interactive browser-token paste, unsuitable # for unattended runtime. Logged at WARNING. logger.warning( "TVDataFeed: using username/password login. " "TradingView CAPTCHA may force interactive prompts; " "prefer TV_TOKEN env var on server deployments." ) self._client = TvDatafeed(self.username, self.password) else: # Anonymous mode (default). ~5000 bars/req cap. self._client = TvDatafeed() return self._client def _write_token_cache(self) -> None: """Write self.token to self.token_cache_file in the format the lib's _load_token() expects: {"token": "..."}.""" try: self.token_cache_file.parent.mkdir(parents=True, exist_ok=True) self.token_cache_file.write_text( json.dumps({"token": self.token}) ) except OSError as exc: logger.warning( "TVDataFeed: failed to write token cache to %s: %s", self.token_cache_file, exc, ) def _get_interval(self, timeframe: str) -> Any | None: if self._interval_map is None: try: self._interval_map = self._interval_map_builder() except ImportError as exc: logger.warning( "TVDataFeed: tvdatafeed-enhanced not installed (%s); " "set USE_TV_FEED=0 or install the package", exc, ) self._interval_map = {} return self._interval_map.get(timeframe) async def get_bars( self, symbol: str, timeframe: str, n: int, ) -> pd.DataFrame: empty = pd.DataFrame( columns=["time", "open", "high", "low", "close", "volume"], ) if symbol not in self.symbol_map: logger.warning( "TVDataFeed: symbol %r not in TV_SYMBOL_MAP; returning empty", symbol, ) return empty tv_interval = self._get_interval(timeframe) if tv_interval is None: logger.warning( "TVDataFeed: timeframe %r unsupported (expected one of %s)", timeframe, _TV_INTERVAL_KEYS, ) return empty tv_symbol, tv_exchange = self.symbol_map[symbol] try: df = await asyncio.to_thread( self._fetch_sync, tv_symbol, tv_exchange, tv_interval, n, ) except Exception as exc: logger.warning( "TVDataFeed.get_bars(%s, %s, %d) failed: %s", symbol, timeframe, n, exc, ) return empty if df is None or df.empty: logger.warning( "TVDataFeed: no bars for %s %s (TV returned empty)", symbol, timeframe, ) return empty return self._normalize(df) def _fetch_sync( self, tv_symbol: str, tv_exchange: str, tv_interval: Any, n: int, ) -> pd.DataFrame | None: client = self._get_client() return client.get_hist( symbol=tv_symbol, exchange=tv_exchange, interval=tv_interval, n_bars=n, fut_contract=1, # front-month continuous (resolves MES -> MES1!) ) @staticmethod def _normalize(df: pd.DataFrame) -> pd.DataFrame: """ Convert tvdatafeed output to V16 contract. TV format: DatetimeIndex (named 'datetime' or unnamed) + columns ['symbol', 'open', 'high', 'low', 'close', 'volume']. V16 format: columns ['time', 'open', 'high', 'low', 'close', 'volume'], sorted ascending, time as UTC datetime. """ out = df.copy() if "symbol" in out.columns: out = out.drop(columns=["symbol"]) # Index -> 'time' column. tvdatafeed names the index 'datetime'. out = out.reset_index() if "datetime" in out.columns: out = out.rename(columns={"datetime": "time"}) elif "index" in out.columns: out = out.rename(columns={"index": "time"}) out["time"] = pd.to_datetime(out["time"], utc=False) out = out.sort_values("time").reset_index(drop=True) return out[["time", "open", "high", "low", "close", "volume"]]