diff --git a/CLAUDE.md b/CLAUDE.md index 3b9dd73..6349e12 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,7 +15,12 @@ This file is the authoritative developer/agent reference. The human-facing docs | `config.py` | ~374 | All config dataclasses, `PAIR_INFO` registry, per-pair strategy configs. **`PlatformConfig` defaults ARE the tuned XAU config.** | | `app_config.py` | ~290 | config.json loader/validation, env-precedence resolvers, repo-root `.env` loader (setdefault at import), `apply_pair_overrides` deployment overlay | | `config.json.example` | 24 | committed schema template; real gitignored config.json holds secrets + deployment settings | -| `data_provider.py` | ~347 | Twelve Data SDK wrapper: fetch, rate-limit (55 credits/min), in-memory cache, validation/cleaning, freshness checks | +| `providers/base.py` | ~400 | `BaseDataProvider`: shared cache, validation/cleaning, freshness + boundary guarantees, `fetch_range` pagination, streaming stubs. Subclasses implement only `_fetch_raw` | +| `providers/twelve_data.py` | ~240 | `TwelveDataProvider` (alias `ForexDataProvider`): TD SDK fetch, rate-limit (55 credits/min), WS quote streaming | +| `providers/oanda.py` | ~190 | `OandaDataProvider`: v20 REST MBA candles, real bid/ask spread, date-window fetch | +| `providers/tiingo.py` | ~200 | `TiingoDataProvider`: FX endpoint mid-price candles (no volume, no spread — see Tiingo note), named errors for every failure shape | +| `providers/__init__.py` | ~55 | `PROVIDER_REGISTRY` + `make_data_provider` (config.json `data_providers.active` selects) | +| `data_provider.py` / `oanda_provider.py` | ~25 | back-compat shims re-exporting from `providers/` | | `engine.py` | ~1173 | `TechnicalEngine`: computes all indicators on one DataFrame → `TechnicalAnalysis` (composite score, regime, key levels) | | `chart_patterns.py` | ~280 | Chart-pattern detectors for the XAU_CP archetype (pure functions, ATR-unit geometry, first-close-beyond triggers) | | `signal_aggregator.py` | ~1447 | `SignalAggregator`: MTF analysis, news integration, gates/penalties, quality grade, trade-setup generation → `SignalReport` | @@ -24,7 +29,7 @@ This file is the authoritative developer/agent reference. The human-facing docs | `pair_signals/__init__.py` | 52 | `PAIR_PRODUCERS` registry + `produce_signal()` | | `pair_signals/xauusd.py` | 71 | XAU config factory + per-timeframe overrides (docstring = validation record) | | `main.py` | ~348 | CLI: `python main.py --pair XAUUSD --timeframe H1 --detail/--json` | -| `test_all.py` | ~3100 | QA suite: **210 tests in 14 modules**, no live API needed. Must stay green. | +| `test_all.py` | ~4100 | QA suite: **253 tests in 16 modules**, no live API needed. Must stay green. | | `backtest/gpu_backtest.py` | ~2231 | Bar-replay backtester (CuPy→NumPy fallback) using the real `SignalAggregator` | | `backtest/diagnose_filters.py` | 609 | Filter-diagnosis tooling | | `backtest/research/` | — | Calendar research infra (`ff_calendar_scraper.py`, `calendar_store.py`) + result CSVs | @@ -230,15 +235,25 @@ Two entry points: --- -## data_provider.py — ForexDataProvider +## providers/ — data-provider abstraction (refactored 2026-07-25) -- Twelve Data SDK (`TDClient.time_series(...).as_pandas()`), symbols normalized to `"XAU/USD"` form; renames lowercase columns → capitalized; sorts oldest-first; localizes UTC. -- **Rate limit**: 55 credits/min sliding window (`_throttle`), blocks with sleep when exhausted. -- **In-memory cache**: key `{symbol}_{tf}`, TTL = candle duration (M5 300 s … H4 14400 s); a cached frame is evicted and refetched if it fails the freshness check **or the boundary check** (`_has_latest_closed_bar`: the frame must contain the most recent fully closed bar, weekend-exempt — a wall-clock TTL alone would serve mid-candle snapshots at the next boundary and scans would run one bar behind). Fresh fetches enforce the same boundary guarantee: a response missing the just-closed bar (feed latency at the boundary) gets one grace retry after `FRESH_FETCH_RETRY_SECONDS` (3 s), then proceeds with a loud warning. Cache and throttle are lock-protected (one provider is shared across the bot's thread pool). -- **`_drop_forming_candle`**: Twelve Data includes the current partial candle; the scanner fires seconds after boundary, so the last bar is dropped if its close time is in the future. Without this every indicator would read a seconds-old stub and signals would repaint. -- **Freshness** (`_is_data_fresh`): last bar older than 3× candle duration → stale (warns; live path suppresses signal if the PRIMARY TF is stale). Weekends exempt (Fri 22:00→Sun 22:00 UTC). -- **`_validate_and_clean`**: drop all-NaN rows → ffill (limit 3) → drop NaN closes → sort + dedupe (keep last) → drop future timestamps (>now+5 min) → repair OHLC inconsistencies (High=max, Low=min of O/C/existing) → remove >10σ return outliers (spike bar only — the opposite-direction snap-back bar right after it is kept, else every spike punched a two-bar hole) → warn if < ~210 bars. -- `outputsize = max(300, min(days×bars_per_day+20, 5000))`; TD caps at 5000/call (the backtester paginates by date). +One class per provider, all subclassing **`providers/base.py BaseDataProvider`**, which owns every behavior the signal pipeline depends on — defined ONCE (a test enforces that neither subclass overrides shared methods): + +- **In-memory cache**: key `{symbol}_{tf}`, TTL = candle duration (M5 300 s … H4 14400 s); a cached frame is evicted and refetched if it fails the freshness check **or the boundary check** (`_has_latest_closed_bar`: the frame must contain the most recent fully closed bar, weekend-exempt — a wall-clock TTL alone would serve mid-candle snapshots at the next boundary and scans would run one bar behind). Fresh fetches enforce the same boundary guarantee: a response missing the just-closed bar (feed latency at the boundary) gets one grace retry after `FRESH_FETCH_RETRY_SECONDS` (3 s), then proceeds with a loud warning. Cache is lock-protected (one provider is shared across the bot's thread pool). +- **`_drop_forming_candle`**: some feeds include the current partial candle; the scanner fires seconds after boundary, so the last bar is dropped if its close time is in the future. Without this every indicator would read a seconds-old stub and signals would repaint. (No-op on feeds that only return complete candles, e.g. Oanda.) +- **Freshness** (`is_data_fresh`, public): last bar older than 3× candle duration → stale (warns; live path suppresses signal if the PRIMARY TF is stale). Weekends exempt (Fri 22:00→Sun 22:00 UTC). +- **`_validate_and_clean`** (public wrapper `validate_and_clean`): drop all-NaN rows → ffill (limit 3) → drop NaN closes → sort + dedupe (keep last) → drop future timestamps (>now+5 min) → repair OHLC inconsistencies (High=max, Low=min of O/C/existing) → remove >10σ return outliers (spike bar only — the opposite-direction snap-back bar right after it is kept, else every spike punched a two-bar hole) → warn if < ~210 bars. +- **`fetch_range(symbol, tf, start, end)`**: paginated RAW date-range fetch under `MAX_BARS_PER_REQUEST` (5000) — the backtester's `fetch_history` calls this, so ANY registered provider can feed the backtester. +- `outputsize = max(300, min(days×bars_per_day+20, 5000))`; `CANDLE_SECONDS`/`BARS_PER_DAY` tables live on the base class (the aggregator's bar-close clock reads `BaseDataProvider.CANDLE_SECONDS`). +- **Streaming capability (optional)**: `supports_streaming` flag + async generator `iter_ticks(symbols, on_connect)` yielding `(canonical_symbol, price)` for one connection's lifetime; the caller owns reconnect/backoff. Errors: each provider raises its own `DATA_ERROR`, always a subclass of `DataIntegrityError` — catch the base type provider-agnostically. + +**A subclass implements only `_fetch_raw(symbol, tf, outputsize, start=None, end=None)`** (normalized frame: capitalized OHLCV, UTC tz-aware index named "Datetime", oldest-first) plus symbol/interval maps and optional capabilities. To add a provider: one module in `providers/`, one `PROVIDER_REGISTRY` entry in `providers/__init__.py`, one config.json `data_providers` block. The live feed is selected by config.json `data_providers.active` (registration alone changes nothing). + +- **`providers/twelve_data.py` `TwelveDataProvider`** (alias `ForexDataProvider`): TD SDK (`TDClient.time_series(...).as_pandas()`), symbols normalized to `"XAU/USD"` form; lowercase→capitalized columns, newest-first→oldest-first; **rate limit** 55 credits/min sliding window (`_throttle`, lock-protected, blocks with sleep). Implements streaming (`ws.twelvedata.com` quotes WS, 10 s heartbeat, `_parse_ws_price`). +- **`providers/oanda.py` `OandaDataProvider`**: v20 REST MBA candles (complete-only), real bid/ask spread via `get_spread` override; credentials default to the config.json `oanda` block so the registry can build it config-only (explicit `api_token`/`base_url` args still accepted — the daemon passes them). Date-window fetches send `from`/`to` and omit `count` (v20 rejects the combination). +- **`providers/tiingo.py` `TiingoDataProvider`**: `/tiingo/fx/{ticker}/prices` resampled mid-price candles; no volume (filled 0.0), W1 unmapped (raises). Tiingo takes date windows, not bar counts — live fetches derive `startDate` from the requested count × candle duration × 1.6 weekend pad. Auth in the header (key never in URLs/logs). Every failure shape raises `TiingoDataError` with a named cause (429 quota, 401/403 auth, empty response, malformed candle, missing key). **Candles only** — `get_spread` stays on the base default; the free quote endpoint's indicative spread must not be re-added (see Tiingo spread-monitor removal note). +- **`data_provider.py` / `oanda_provider.py` are back-compat shims** re-exporting from `providers/`; `ForexDataProvider` is the historical name of the TD provider. +- **No silent failures**: provider errors all subclass `DataIntegrityError` and carry a named cause; the aggregator logs per-TF fetch failures and degrades the report (missing primary → NEUTRAL/F "data unavailable"); `make_data_provider` raises on an unknown `active` name; keyless providers warn at construction and raise at fetch. --- @@ -270,7 +285,7 @@ Run: `python telegram_bot.py` (credentials resolve env `TELEGRAM_TOKEN`/`TELEGRA - **Subscriber channel** (`TELEGRAM_CHANNEL_ID`, `-100…`; `""` disables): auto-scan signal alerts + breach alerts also post there (bot needs channel-admin). Monetization = Telegram Stars subscription invite links; Telegram handles membership, the bot has zero payment code. The channel feed keeps running even when the owner toggles DM auto-scan off (`_feed_active()` = DM chats OR channel). - **Auto-scan**: boot-starts if `TELEGRAM_CHAT_ID` set (initial M15 scan populates the level cache first). Candle-boundary loops for M5/M15/H1/H4, firing at boundary + 10 s. Each scan runs all `scan_pairs` concurrently, updates the level cache, then broadcasts. - **Alert gating**: quality threshold (`_alert_min_quality`, default "C", menu cycles C→B→A) + fingerprint dedup per (symbol, TF) on (signal, quality, direction) with 1-hour re-arm — a persisting identical signal doesn't re-send every candle, but a setup whose entry moved ≥ half the last-ALERTED risk distance (new pullback/swing anchor; compared against the stored last-alert entry/risk, a stable reference immune to per-candle ATR drift) re-alerts. The fingerprint is committed only **after** a message reaches ≥1 target (`_commit_alert`), so a failed Telegram send doesn't consume the alert. -- **WebSocket monitor**: subscribes every `scan_pairs` symbol to TD quotes WS; 10 s heartbeat; exponential reconnect (≤60 s). On tick crossing any cached key level (`swing_high/low_1..3`, `ema_trend`, `bb_upper/lower`, `kc_upper/lower`; cache TTL 1 h) → immediate M15 rescan → "LEVEL BREAK" alert if actionable+changed (same gates as auto-scan), 5-min per-pair cooldown. +- **WebSocket monitor**: the provider owns the connection/protocol (`BaseDataProvider.iter_ticks` streaming capability — TD implements it; heartbeat + message parsing live in `providers/twelve_data.py`); the bot owns exponential reconnect (≤60 s) + breach dispatch, and disables the monitor with a log line if the active provider has `supports_streaming=False`. On tick crossing any cached key level (`swing_high/low_1..3`, `ema_trend`, `bb_upper/lower`, `kc_upper/lower`; cache TTL 1 h) → immediate M15 rescan → "LEVEL BREAK" alert if actionable+changed (same gates as auto-scan), 5-min per-pair cooldown. - **Commands**: `/start` (menu), `/scan` (buttons: per pair — M5/M15/H1/H4 + Scan All TFs), `/autoscan` (toggle DM copy), `/status` (loops, WS, live prices, level cache, cooldowns), `/cache` (age per pair/TF), `/clearcache`, `/help`. - **Tiingo spread monitor: REMOVED 2026-07-23** (shipped 2026-07-22, removed next day). Tiingo's free `/tiingo/fx/top` publishes a two-tier *indicative* spread for gold — exactly $0.10 in liquid hours and exactly $0.70 in the 21:00–00:00 UTC rollover window / news moments (verified: 273-sample production log + live probes; only those two values ever occur). Quotes are mid ± a posted increment, not executable depth, so the feature could never validate the $0.30 backtest cost assumption. Do not re-add a spread display from Tiingo's free FX feed. - Signal messages (redesigned 2026-07-24, `format_signal_message`): verdict header first, then trade setup, then a condensed one-line indicator summary (`EMA↑ RSI→ …`, `_indicator_summary`) + warnings. The setup block includes an **informational 100-pip TP line** (`_INFO_TP_PIPS`, display-only — entry ± 100 pips, ≈R vs stop; NOT part of the backtested exit model) and the **trailing-stop management plan** (arm price + trail distance, read from the `trail_arm_r`/`trail_dist_r` stamped on the setup by the producing per-TF config). Risk disclaimer on every outbound message. @@ -314,13 +329,13 @@ Report: summary metrics (total R, PF, Sharpe on R, max DD, MFE/MAE), breakdowns **Comparability note (2026-07-16):** the parity/realism fixes of that date (EMA-ribbon port to the GPU engine, exact-pandas Bollinger precompute, raw-path entry re-anchoring, per-TF D1 warmup, live confirm-map CLI default, outlier snap-back retention) all shift books slightly vs earlier runs. Books produced before these fixes are not directly comparable — re-run the baseline before any before/after tuning comparison. -Data: paginated TD fetches (5000-bar chunks) cached as parquet in `backtest/cache/` keyed by symbol/TF/window; warmup pad ≥60 days before the window for EMA200, extended per-TF so slow confirmation TFs get ≥210 bars at window start (D1 gets ~299 days — 62 days gave only ~61 bars: no EMA200, unlike live's 300-bar outputsize floor). `--confirm-tfs` defaults to the pair factory's live confirmation map (pass it explicitly to override); a hardcoded default here previously made the standard invocation backtest a different MTF setup than the bot scans. +Data: paginated fetches via `provider.fetch_range` (5000-bar chunks; works with any registered provider) cached as parquet in `backtest/cache/` keyed by symbol/TF/window; warmup pad ≥60 days before the window for EMA200, extended per-TF so slow confirmation TFs get ≥210 bars at window start (D1 gets ~299 days — 62 days gave only ~61 bars: no EMA200, unlike live's 300-bar outputsize floor). `--confirm-tfs` defaults to the pair factory's live confirmation map (pass it explicitly to override); a hardcoded default here previously made the standard invocation backtest a different MTF setup than the bot scans. --- ## test_all.py — QA suite -`python test_all.py` — **210 tests, 14 modules** (config, data_provider, engine, news_engine, signal_aggregator, main, telegram_bot, integration, pair_signals registry + engine-bug regressions, 2026-07-16 audit-fix regressions, chart patterns, app_config, .env loader + bot resilience). No network needed (mocked). Uses a custom `@test("name")` decorator that runs immediately (NOT pytest — don't grep for `def test_`). Exit code non-zero on failure. Includes: the xauusd-factory ≡ inline-config equivalence test, engine bug-fix regression tests (Module 9b), a live-vs-GPU engine parity sweep incl. the EMA-ribbon path (Module 11), config.json loading/precedence/validation coverage (Module 13), and `.env` loader + bot-resilience coverage (Module 14). **Any signal-logic change must keep this green and should add tests.** +`python test_all.py` — **253 tests, 16 modules** (config, data_provider, engine, news_engine, signal_aggregator, main, telegram_bot, integration, pair_signals registry + engine-bug regressions, 2026-07-16 audit-fix regressions, chart patterns, app_config, .env loader + bot resilience, Oanda provider/executor/daemon, provider abstraction). No network needed (mocked). Uses a custom `@test("name")` decorator that runs immediately (NOT pytest — don't grep for `def test_`). Exit code non-zero on failure. Includes: the xauusd-factory ≡ inline-config equivalence test, engine bug-fix regression tests (Module 9b), a live-vs-GPU engine parity sweep incl. the EMA-ribbon path (Module 11), config.json loading/precedence/validation coverage (Module 13), and `.env` loader + bot-resilience coverage (Module 14). **Any signal-logic change must keep this green and should add tests.** --- @@ -330,7 +345,8 @@ Precedence for every credential/deployment setting: **env var → config.json | Variable | Purpose | Default | |---|---|---| -| `TWELVE_DATA_API_KEY` | market data key | none — falls through to config.json `data_providers.twelve_data.api_key`, then fails fast | +| `TWELVE_DATA_API_KEY` | market data key (twelve_data only) | none — falls through to config.json `data_providers.twelve_data.api_key`, then fails fast | +| `TIINGO_API_KEY` | market data key (tiingo only) | none — falls through to config.json `data_providers.tiingo.api_key`, then fails fast | | `TELEGRAM_TOKEN` | bot token | none — falls through to config.json `telegram.token`, then refuses to boot | | `TELEGRAM_CHAT_ID` | owner chat id (boot auto-scan + owner lock) | none — falls through to config.json `telegram.chat_id`, then refuses to boot | | `TELEGRAM_CHANNEL_ID` | subscriber channel (`-100…`); `""` disables the feed | none — falls through to config.json `telegram.channel_id` | diff --git a/app_config.py b/app_config.py index e4cb012..1f68f0c 100644 --- a/app_config.py +++ b/app_config.py @@ -282,15 +282,27 @@ def telegram_channel_id(app: Optional[AppConfig] = None) -> str: return env if env is not None else app.telegram_channel_id -def provider_api_key(app: Optional[AppConfig] = None) -> str: - """API key for the ACTIVE provider. The env override is per-provider - (TWELVE_DATA_API_KEY applies only when twelve_data is active).""" +# Per-provider env-var overrides for API keys (env → config.json → ""). +_PROVIDER_ENV_KEYS = { + "twelve_data": "TWELVE_DATA_API_KEY", + "tiingo": "TIINGO_API_KEY", +} + + +def provider_api_key( + app: Optional[AppConfig] = None, provider: Optional[str] = None +) -> str: + """API key for `provider` (default: the ACTIVE provider). The env + override is per-provider (TWELVE_DATA_API_KEY applies only to + twelve_data, TIINGO_API_KEY only to tiingo).""" app = app or get_app_config() - if app.active_provider == "twelve_data": - env = os.environ.get("TWELVE_DATA_API_KEY") + name = provider or app.active_provider + env_key = _PROVIDER_ENV_KEYS.get(name) + if env_key: + env = os.environ.get(env_key) if env: return env - block = app.provider_settings.get(app.active_provider, {}) + block = app.provider_settings.get(name, {}) return str(block.get("api_key", "")) diff --git a/backtest/diagnose_filters.py b/backtest/diagnose_filters.py index 086f278..a2d82a3 100644 --- a/backtest/diagnose_filters.py +++ b/backtest/diagnose_filters.py @@ -51,7 +51,7 @@ from gpu_backtest import ( # noqa: E402 precompute_indicators, _is_xau_blocked_by_calendar, ) -from data_provider import ForexDataProvider # noqa: E402 +from providers import make_data_provider # noqa: E402 logger = logging.getLogger("diagnose_filters") @@ -206,7 +206,7 @@ def run_diagnosis( confirmation_timeframes=list(confirm_tfs), ) - provider = ForexDataProvider(platform_cfg) + provider = make_data_provider(platform_cfg) # ── Load data ── timeframes = [primary_tf] + list(confirm_tfs) diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index 4676d6d..68d634c 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -55,7 +55,7 @@ from config import ( # noqa: E402 Timeframe, XAUUSD, ) -from data_provider import ForexDataProvider, make_data_provider # noqa: E402 +from providers import BaseDataProvider, make_data_provider # noqa: E402 from engine import IndicatorSignal, TechnicalAnalysis, TechnicalEngine # noqa: E402 from signal_aggregator import SignalAggregator, SignalReport, TradeSetup # noqa: E402 @@ -941,7 +941,7 @@ def _cache_path(symbol: str, tf: Timeframe, start: datetime, end: datetime) -> P def fetch_history( - provider: ForexDataProvider, + provider: BaseDataProvider, symbol: str, timeframe: Timeframe, start: datetime, @@ -949,12 +949,13 @@ def fetch_history( use_cache: bool = True, clean: bool = True, ) -> pd.DataFrame: - """Fetch OHLCV between start/end with parquet caching and pagination. + """Fetch OHLCV between start/end with parquet caching. - Twelve Data caps responses at 5000 bars; we paginate by date range so - multi-month M5/M15 backtests work. + Pagination under the provider's per-request bar cap lives in + `BaseDataProvider.fetch_range`, so any registered provider can feed + the backtester. - The parquet cache stores the RAW series; `_validate_and_clean` runs on + The parquet cache stores the RAW series; `validate_and_clean` runs on load. Pass clean=False to get the raw price path — trade resolution must use it, because cleaning drops >10σ spike bars (news candles) that hit stops in reality. (Caches written before this change already contain @@ -967,73 +968,16 @@ def fetch_history( df = pd.read_parquet(path) if df.index.tz is None: df.index = df.index.tz_localize("UTC") - return provider._validate_and_clean(df, symbol, timeframe) if clean else df + return provider.validate_and_clean(df, symbol, timeframe) if clean else df logger.info(f"Fetching {symbol} {timeframe.value} from {start.date()} to {end.date()}") - # Bars-per-call estimate to size each window safely under the 5000 cap. - bars_per_day = ForexDataProvider.BARS_PER_DAY[timeframe] - days_per_chunk = max(1, int(4500 / bars_per_day)) - - chunks: List[pd.DataFrame] = [] - cursor_end = end - empty_streak = 0 - while cursor_end > start: - cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk)) - try: - chunk = provider._fetch_twelve_data( - symbol=symbol, - timeframe=timeframe, - outputsize=5000, - start_date=cursor_start.strftime("%Y-%m-%d %H:%M:%S"), - end_date=cursor_end.strftime("%Y-%m-%d %H:%M:%S"), - ) - except Exception as e: - # Twelve Data raises BadRequest for windows with no data (weekends, - # holidays, asset gaps). Skip the window, step back, retry — but - # bail out after a few consecutive empties to avoid spinning. - logger.warning( - f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}" - ) - empty_streak += 1 - if empty_streak >= 3: - break - cursor_end = cursor_start - timedelta(seconds=1) - continue - - if chunk is None or chunk.empty: - empty_streak += 1 - if empty_streak >= 3: - break - cursor_end = cursor_start - timedelta(seconds=1) - continue - - empty_streak = 0 - chunks.append(chunk) - # Step back; dedup on concat. - cursor_end = chunk.index[0] - timedelta(seconds=1) - if len(chunk) < 2: - break - - if not chunks: - raise RuntimeError(f"No data fetched for {symbol} {timeframe.value}") - - df = pd.concat(chunks).sort_index() - df = df[~df.index.duplicated(keep="last")] - # Fetches ending "now" include the still-forming candle — drop it so the - # cache only ever contains closed bars. - df = provider._drop_forming_candle(df, timeframe) - # Trim to requested window. `start`/`end` are already tz-aware UTC - # datetimes — wrapping with pd.Timestamp(..., tz=...) would raise on - # already-localized inputs. - end_ts = pd.Timestamp(end) - start_ts = pd.Timestamp(start) - df = df[(df.index >= start_ts) & (df.index <= end_ts)] + df = provider.fetch_range(symbol, timeframe, start, end) if use_cache: df.to_parquet(path) logger.info(f"Cached {len(df)} raw bars to {path.name}") - return provider._validate_and_clean(df, symbol, timeframe) if clean else df + return provider.validate_and_clean(df, symbol, timeframe) if clean else df # ═════════════════════════════════════════════════════════════════════ @@ -1703,7 +1647,7 @@ def run_backtest( # ≥210 bars (outputsize floor). Target data_provider's min_required # (210 bars); ×7/5 converts trading days to calendar days. Intraday # TFs stay on the base pad, so their books are unchanged. - bars_per_day = ForexDataProvider.BARS_PER_DAY[tf] + bars_per_day = BaseDataProvider.BARS_PER_DAY[tf] min_pad_days = int(np.ceil(210 / bars_per_day * 7 / 5)) + 5 tf_start = bt_cfg.start - timedelta(days=max(pad_days, min_pad_days)) df = fetch_history( diff --git a/config.json.example b/config.json.example index 261d631..441ae22 100644 --- a/config.json.example +++ b/config.json.example @@ -9,6 +9,9 @@ "active": "twelve_data", "twelve_data": { "api_key": "REPLACE_WITH_YOUR_TWELVE_DATA_API_KEY" + }, + "tiingo": { + "api_key": "REPLACE_WITH_YOUR_TIINGO_API_KEY_OR_EMPTY" } }, "scanning": { diff --git a/data_provider.py b/data_provider.py index 1119cd1..bcbb4db 100755 --- a/data_provider.py +++ b/data_provider.py @@ -1,403 +1,24 @@ """ -ForexBot2 — Data Provider (Twelve Data SDK) -========================================================== -Fetches and validates OHLCV via the official `twelvedata` Python SDK. +ForexBot2 — Data Provider (back-compat shim) +============================================= +The provider implementations live in the `providers/` package: + + providers/base.py BaseDataProvider (shared contract + behavior) + providers/twelve_data.py TwelveDataProvider (SDK fetch, throttle, WS) + providers/oanda.py OandaDataProvider (v20 REST, real spread) + +`ForexDataProvider` is the historical name of the Twelve Data provider and +is kept as an alias so existing imports keep working. """ -import logging -import threading -import time -from collections import deque -from typing import Dict, Optional +from providers import ( # noqa: F401 + BaseDataProvider, + DataIntegrityError, + PROVIDER_REGISTRY, + TwelveDataProvider, + make_data_provider, + _is_forex_weekend, +) -import pandas as pd -from twelvedata import TDClient - -from app_config import AppConfigError, get_app_config, provider_api_key -from config import PlatformConfig, Timeframe, XAUUSD, PAIR_INFO - -logger = logging.getLogger(__name__) - - -def _is_forex_weekend(now=None) -> bool: - """Forex market closed: Friday 22:00 UTC → Sunday 22:00 UTC.""" - if now is None: - now = pd.Timestamp.now(tz="UTC") - wd = now.weekday() - hour = now.hour - if wd == 4 and hour >= 22: - return True - if wd == 5: - return True - if wd == 6 and hour < 22: - return True - return False - - -class DataIntegrityError(Exception): - """Raised when data fails validation checks.""" - pass - - -class ForexDataProvider: - """Twelve Data SDK client + on-disk parquet cache.""" - - TD_INTERVALS = { - Timeframe.M1: "1min", - Timeframe.M5: "5min", - Timeframe.M15: "15min", - Timeframe.M30: "30min", - Timeframe.H1: "1h", - Timeframe.H4: "4h", - Timeframe.D1: "1day", - Timeframe.W1: "1week", - } - - BARS_PER_DAY = { - Timeframe.M1: 1440, Timeframe.M5: 288, Timeframe.M15: 96, - Timeframe.M30: 48, Timeframe.H1: 24, Timeframe.H4: 6, - Timeframe.D1: 1, Timeframe.W1: 0.143, - } - - CANDLE_SECONDS = { - Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900, - Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, - Timeframe.D1: 86400, Timeframe.W1: 604800, - } - - RATE_LIMIT = 55 - RATE_WINDOW = 60.0 - # Grace wait before the one retry when a fresh response at a candle - # boundary is missing the just-closed bar (feed latency). - FRESH_FETCH_RETRY_SECONDS = 3.0 - - def __init__(self, config: PlatformConfig): - self.config = config - self._api_key = provider_api_key() or None - self._cache: Dict[str, pd.DataFrame] = {} - self._cache_timestamps: Dict[str, float] = {} - # The bot shares one provider across a thread pool; cache eviction - # and the throttle deque are check-then-act and need locking. - self._cache_lock = threading.Lock() - self._throttle_lock = threading.Lock() - self._request_times: deque = deque() - self._td: Optional[TDClient] = ( - TDClient(apikey=self._api_key) if self._api_key else None - ) - - @staticmethod - def _to_td_symbol(symbol: str) -> str: - if "/" in symbol: - return symbol - clean = symbol.replace("=X", "") - return f"{clean[:3]}/{clean[3:]}" - - # ── Public API ───────────────────────────────────────────────── - - def fetch_ohlcv( - self, - symbol: str = XAUUSD, - timeframe: Timeframe = Timeframe.H1, - lookback_days: Optional[int] = None, - ) -> pd.DataFrame: - """Fetch OHLCV with in-memory caching for live mode.""" - if lookback_days is None: - lookback_days = self.config.data_lookback_days - - cache_key = f"{symbol}_{timeframe.value}" - now = time.time() - - cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval) - with self._cache_lock: - cached = self._cache.get(cache_key) - cached_at = self._cache_timestamps.get(cache_key) - if ( - cached is not None - and cached_at is not None - and (now - cached_at) < cache_ttl - and not cached.empty - ): - if (self._is_data_fresh(cached, symbol, timeframe) - and self._has_latest_closed_bar(cached, timeframe)): - return cached - logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching") - with self._cache_lock: - self._cache.pop(cache_key, None) - self._cache_timestamps.pop(cache_key, None) - - outputsize = max( - 300, - min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000), - ) - df = self._fetch_twelve_data(symbol, timeframe, outputsize) - df = self._drop_forming_candle(df, timeframe) - df = self._validate_and_clean(df, symbol, timeframe) - - # The closed-bar guarantee must hold for FRESH responses too, not - # just cache hits: the scanner fires seconds after the boundary, - # and a lagging feed can answer without the just-closed bar — the - # scan would silently analyze (and session-check) the previous - # bar. Give the feed one short grace retry, then proceed loudly. - if (not df.empty and not self._has_latest_closed_bar(df, timeframe)): - time.sleep(self.FRESH_FETCH_RETRY_SECONDS) - retry = self._fetch_twelve_data(symbol, timeframe, outputsize) - retry = self._drop_forming_candle(retry, timeframe) - retry = self._validate_and_clean(retry, symbol, timeframe) - if retry is not None and not retry.empty: - df = retry - if not self._has_latest_closed_bar(df, timeframe): - logger.warning( - f"{symbol} {timeframe.value}: feed lagging — just-closed bar " - f"still missing after retry; this scan may run one bar behind" - ) - - if not self._is_data_fresh(df, symbol, timeframe): - logger.warning( - f"STALE DATA from API: {symbol} {timeframe.value} — " - f"last bar {df.index[-1]} is too old. Signal quality may be degraded." - ) - - with self._cache_lock: - self._cache[cache_key] = df - self._cache_timestamps[cache_key] = now - return df - - def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool: - """Cached data must include the most recent fully closed bar. - - The TTL alone is wall-clock based: a snapshot fetched mid-candle is - still within TTL at the next candle boundary but is missing the bar - that just closed, so every boundary scan would analyze — and run the - bar-close session check against — the previous bar. - """ - now = pd.Timestamp.now(tz="UTC") - if _is_forex_weekend(now): - return True - candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) - dur = pd.Timedelta(seconds=candle_seconds) - expected_label = now.floor(dur) - dur - return df.index[-1] >= expected_label - - def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame: - """Drop the still-forming last candle so analysis only sees closed bars. - - Twelve Data's time_series includes the current partial candle; the - scanner fires seconds after the boundary, so without this guard every - indicator reads a seconds-old stub bar and signals repaint. - """ - if df is None or df.empty: - return df - candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) - now = pd.Timestamp.now(tz="UTC") - if df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now: - df = df.iloc[:-1] - return df - - def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool: - """Check if the last candle is within acceptable age for this timeframe.""" - if df is None or df.empty: - return False - - now = pd.Timestamp.now(tz="UTC") - - if _is_forex_weekend(now): - return True - - last_bar = df.index[-1] - candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) - max_age_seconds = candle_seconds * 3 - age_seconds = (now - last_bar).total_seconds() - - if age_seconds > max_age_seconds: - logger.warning( - f"{symbol} {timeframe.value}: last bar {last_bar.strftime('%Y-%m-%d %H:%M')} " - f"is {age_seconds / 60:.0f}m old (limit: {max_age_seconds / 60:.0f}m)" - ) - return False - - return True - - def clear_cache(self) -> int: - """Wipe all cached data. Returns number of entries cleared.""" - with self._cache_lock: - count = len(self._cache) - self._cache.clear() - self._cache_timestamps.clear() - logger.info(f"Cache cleared: {count} entries wiped") - return count - - def get_pair_info(self, symbol: str = XAUUSD) -> Dict: - info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD]) - return {"symbol": symbol, **info} - - # ── Private ──────────────────────────────────────────────────── - - def _throttle(self, credits: int = 1) -> None: - """Pro tier: 55 credits / minute. Block until budget allows.""" - while True: - with self._throttle_lock: - now = time.time() - while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: - self._request_times.popleft() - - available = self.RATE_LIMIT - len(self._request_times) - if available >= credits: - for _ in range(credits): - self._request_times.append(time.time()) - return - - used = len(self._request_times) - wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2 - logger.info( - f"Rate limit: {used}/{self.RATE_LIMIT} credits used " - f"— waiting {wait:.1f}s..." - ) - time.sleep(wait) - - def _fetch_twelve_data( - self, - symbol: str, - timeframe: Timeframe, - outputsize: int, - start_date: Optional[str] = None, - end_date: Optional[str] = None, - ) -> pd.DataFrame: - if self._td is None: - raise DataIntegrityError( - "No data-provider API key. Set the TWELVE_DATA_API_KEY env " - "var or data_providers.twelve_data.api_key in config.json." - ) - - td_symbol = self._to_td_symbol(symbol) - interval = self.TD_INTERVALS[timeframe] - - self._throttle(credits=1) - - kwargs = { - "symbol": td_symbol, - "interval": interval, - "outputsize": min(outputsize, 5000), - "timezone": "UTC", - } - if start_date: - kwargs["start_date"] = start_date - if end_date: - kwargs["end_date"] = end_date - - try: - df = self._td.time_series(**kwargs).as_pandas() - except Exception as e: - raise DataIntegrityError( - f"Twelve Data fetch failed for {td_symbol}: {e}" - ) from e - - if df is None or df.empty: - raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}") - - # SDK returns lowercase columns; engine expects capitalised. - df = df.rename(columns={ - "open": "Open", "high": "High", "low": "Low", - "close": "Close", "volume": "Volume", - }) - for col in ("Open", "High", "Low", "Close", "Volume"): - if col in df.columns: - df[col] = pd.to_numeric(df[col], errors="coerce") - if "Volume" not in df.columns: - df["Volume"] = 0.0 - - # SDK returns newest-first; engine + indicators want oldest-first. - df = df.sort_index() - if df.index.tz is None: - df.index = df.index.tz_localize("UTC") - df.index.name = "Datetime" - - logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") - return df - - def _validate_and_clean( - self, df: pd.DataFrame, symbol: str, timeframe: Timeframe - ) -> pd.DataFrame: - if df is None or df.empty: - raise DataIntegrityError(f"Empty dataset for {symbol}") - - original_len = len(df) - - df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all") - df = df.ffill(limit=3) - df = df.dropna(subset=["Close"]) - - df = df.sort_index() - df = df[~df.index.duplicated(keep="last")] - - now = pd.Timestamp.utcnow() - future_mask = df.index > now + pd.Timedelta(minutes=5) - if future_mask.any(): - logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps") - df = df[~future_mask] - - invalid_ohlc = ( - (df["High"] < df["Low"]) | - (df["High"] < df["Open"]) | - (df["High"] < df["Close"]) | - (df["Low"] > df["Open"]) | - (df["Low"] > df["Close"]) - ) - if invalid_ohlc.any(): - logger.warning(f"{symbol}: Fixing {invalid_ohlc.sum()} OHLC-inconsistent bars") - for idx in df[invalid_ohlc].index: - vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]] - df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]]) - df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) - - returns = df["Close"].pct_change() - std_ret = returns.std() - if std_ret > 0: - dev = returns - returns.mean() - outlier_mask = dev.abs() > 10 * std_ret - # A spike bar makes the NEXT bar's return look extreme too (it is - # measured against the spike). If the previous bar is flagged and - # this bar reverses direction, it is the legitimate snap-back — - # keep it, or every spike would punch a two-bar hole in the data. - recovery = ( - outlier_mask - & outlier_mask.shift(1, fill_value=False) - & (dev * dev.shift(1) < 0) - ) - outlier_mask = outlier_mask & ~recovery - if outlier_mask.any(): - logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars") - df = df[~outlier_mask] - - min_required = max( - self.config.technical.ema_trend, - 200, - ) + 10 - - if len(df) < min_required: - logger.warning( - f"{symbol} {timeframe.value}: Only {len(df)} bars " - f"(need {min_required}). Some indicators will be partial." - ) - - cleaned = original_len - len(df) - if cleaned > 0: - logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars") - - return df - - -# ── Provider registry ───────────────────────────────────────────────── -# config.json data_providers.active selects from here. Adding a provider -# later = one adapter class + one entry + one config block. -PROVIDER_REGISTRY: Dict[str, type] = {"twelve_data": ForexDataProvider} - - -def make_data_provider(config: PlatformConfig) -> ForexDataProvider: - """Build the ACTIVE data provider (config.json: data_providers.active).""" - name = get_app_config().active_provider - cls = PROVIDER_REGISTRY.get(name) - if cls is None: - raise AppConfigError( - f'Unknown data provider "{name}"; ' - f"registered: {sorted(PROVIDER_REGISTRY)}") - return cls(config) +# Historical name for the Twelve Data provider. +ForexDataProvider = TwelveDataProvider diff --git a/oanda_daemon.py b/oanda_daemon.py index 8dd19a9..65b02d1 100644 --- a/oanda_daemon.py +++ b/oanda_daemon.py @@ -46,7 +46,7 @@ from app_config import ( from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD from signal_aggregator import SignalAggregator, SignalReport, SignalStrength from pair_signals import PAIR_PRODUCERS -from oanda_provider import OandaDataProvider +from providers.oanda import OandaDataProvider from oanda_executor import OandaExecutor, OandaAPIError logger = logging.getLogger(__name__) diff --git a/oanda_provider.py b/oanda_provider.py index c421fda..d840019 100644 --- a/oanda_provider.py +++ b/oanda_provider.py @@ -1,361 +1,9 @@ """ -ForexBot2 — Oanda v20 Data Provider -==================================== -Drop-in replacement for ForexDataProvider that fetches OHLCV candles -from the Oanda v20 REST API. Returns the same DataFrame format -(Open/High/Low/Close/Volume, UTC DatetimeIndex, oldest-first, closed -bars only) so the signal pipeline (engine + aggregator) runs unchanged. - -Additionally exposes real bid/ask spread from Oanda's MBA candles. +ForexBot2 — Oanda provider (back-compat shim) +============================================== +The implementation lives in `providers/oanda.py`; shared behavior is in +`providers/base.py`. This module keeps the historical import path working. """ -import logging -import threading -import time -from typing import Dict, Optional - -import pandas as pd -import requests - -from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD - -logger = logging.getLogger(__name__) - - -class OandaDataError(Exception): - """Raised when an Oanda data fetch fails or returns invalid data.""" - - -def _is_forex_weekend(now=None) -> bool: - if now is None: - now = pd.Timestamp.now(tz="UTC") - wd = now.weekday() - hour = now.hour - if wd == 4 and hour >= 22: - return True - if wd == 5: - return True - if wd == 6 and hour < 22: - return True - return False - - -class OandaDataProvider: - """Oanda v20 REST candle provider with in-memory caching.""" - - OANDA_GRANULARITY = { - Timeframe.M1: "M1", - Timeframe.M5: "M5", - Timeframe.M15: "M15", - Timeframe.M30: "M30", - Timeframe.H1: "H1", - Timeframe.H4: "H4", - Timeframe.D1: "D", - Timeframe.W1: "W", - } - - OANDA_INSTRUMENT = { - "XAU/USD": "XAU_USD", - } - - BARS_PER_DAY = { - Timeframe.M1: 1440, Timeframe.M5: 288, Timeframe.M15: 96, - Timeframe.M30: 48, Timeframe.H1: 24, Timeframe.H4: 6, - Timeframe.D1: 1, Timeframe.W1: 0.143, - } - - CANDLE_SECONDS = { - Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900, - Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, - Timeframe.D1: 86400, Timeframe.W1: 604800, - } - - FRESH_FETCH_RETRY_SECONDS = 3.0 - - def __init__(self, config: PlatformConfig, api_token: str, base_url: str): - self.config = config - self._base_url = base_url.rstrip("/") - self._session = requests.Session() - self._session.headers.update({ - "Authorization": f"Bearer {api_token}", - "Accept-Datetime-Format": "RFC3339", - "Content-Type": "application/json", - }) - self._cache: Dict[str, pd.DataFrame] = {} - self._cache_timestamps: Dict[str, float] = {} - self._cache_lock = threading.Lock() - self._spread_cache: Dict[str, float] = {} - - # ── Public API ───────────────────────────────────────────────── - - def fetch_ohlcv( - self, - symbol: str = XAUUSD, - timeframe: Timeframe = Timeframe.H1, - lookback_days: Optional[int] = None, - ) -> pd.DataFrame: - if lookback_days is None: - lookback_days = self.config.data_lookback_days - - cache_key = f"{symbol}_{timeframe.value}" - now = time.time() - - cache_ttl = self.CANDLE_SECONDS.get(timeframe, 3600) - with self._cache_lock: - cached = self._cache.get(cache_key) - cached_at = self._cache_timestamps.get(cache_key) - if ( - cached is not None - and cached_at is not None - and (now - cached_at) < cache_ttl - and not cached.empty - ): - if (self._is_data_fresh(cached, symbol, timeframe) - and self._has_latest_closed_bar(cached, timeframe)): - return cached - logger.warning( - "%s %s: cached data is stale, refetching", - symbol, timeframe.value) - with self._cache_lock: - self._cache.pop(cache_key, None) - self._cache_timestamps.pop(cache_key, None) - - count = max(300, min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000)) - df = self._fetch_oanda_candles(symbol, timeframe, count) - df = self._validate_and_clean(df, symbol, timeframe) - - if not df.empty and not self._has_latest_closed_bar(df, timeframe): - time.sleep(self.FRESH_FETCH_RETRY_SECONDS) - try: - retry = self._fetch_oanda_candles(symbol, timeframe, count) - retry = self._validate_and_clean(retry, symbol, timeframe) - if retry is not None and not retry.empty: - df = retry - except OandaDataError as e: - logger.warning( - "%s %s: boundary retry refetch failed (%s) — keeping " - "original data", symbol, timeframe.value, e) - if not self._has_latest_closed_bar(df, timeframe): - logger.warning( - "%s %s: feed lagging — just-closed bar still missing " - "after retry; this scan may run one bar behind", - symbol, timeframe.value) - - if not self._is_data_fresh(df, symbol, timeframe): - logger.warning( - "STALE DATA from Oanda: %s %s — last bar %s is too old. " - "Signal quality may be degraded.", - symbol, timeframe.value, - df.index[-1] if not df.empty else "N/A") - - with self._cache_lock: - self._cache[cache_key] = df - self._cache_timestamps[cache_key] = now - return df - - def get_pair_info(self, symbol: str = XAUUSD) -> Dict: - info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD]) - return {"symbol": symbol, **info} - - def get_spread(self, symbol: str = XAUUSD) -> float: - return self._spread_cache.get( - symbol, PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"]) - - def clear_cache(self) -> int: - with self._cache_lock: - count = len(self._cache) - self._cache.clear() - self._cache_timestamps.clear() - logger.info("Cache cleared: %d entries wiped", count) - return count - - # ── Private ──────────────────────────────────────────────────── - - def _to_oanda_instrument(self, symbol: str) -> str: - if symbol in self.OANDA_INSTRUMENT: - return self.OANDA_INSTRUMENT[symbol] - return symbol.replace("/", "_") - - def _fetch_oanda_candles( - self, symbol: str, timeframe: Timeframe, count: int - ) -> pd.DataFrame: - instrument = self._to_oanda_instrument(symbol) - granularity = self.OANDA_GRANULARITY.get(timeframe) - if granularity is None: - raise OandaDataError( - f"Unsupported timeframe {timeframe} for Oanda") - - url = f"{self._base_url}/v3/instruments/{instrument}/candles" - params = { - "granularity": granularity, - "count": min(count, 5000), - "price": "MBA", - "dailyAlignment": 0, - "alignmentTimezone": "Etc/UTC", - } - - try: - resp = self._session.get(url, params=params, timeout=30) - except requests.RequestException as e: - raise OandaDataError( - f"Oanda candle fetch failed for {instrument}: {e}") from e - - if resp.status_code != 200: - raise OandaDataError( - f"Oanda candle fetch failed for {instrument}: " - f"HTTP {resp.status_code}: {resp.text[:300]}") - - try: - data = resp.json() - except ValueError as e: - raise OandaDataError( - f"Oanda candle response for {instrument} is not valid " - f"JSON: {e}") from e - candles = data.get("candles", []) - if not candles: - raise OandaDataError( - f"Empty candle response from Oanda for {instrument}") - - rows = [] - last_bid_close = None - last_ask_close = None - for c in candles: - if not c.get("complete", False): - continue - mid = c.get("mid", {}) - bid = c.get("bid", {}) - ask = c.get("ask", {}) - rows.append({ - "Datetime": pd.Timestamp(c["time"]).tz_convert("UTC"), - "Open": float(mid.get("o", 0)), - "High": float(mid.get("h", 0)), - "Low": float(mid.get("l", 0)), - "Close": float(mid.get("c", 0)), - "Volume": int(c.get("volume", 0)), - }) - if bid and ask: - last_bid_close = float(bid.get("c", 0)) - last_ask_close = float(ask.get("c", 0)) - - if not rows: - raise OandaDataError( - f"No complete candles from Oanda for {instrument}") - - if last_bid_close and last_ask_close: - self._spread_cache[symbol] = round( - last_ask_close - last_bid_close, 4) - - 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") - df.index.name = "Datetime" - - for col in ("Open", "High", "Low", "Close"): - df[col] = pd.to_numeric(df[col], errors="coerce") - df["Volume"] = pd.to_numeric(df["Volume"], errors="coerce").fillna(0).astype(float) - - logger.debug( - "Fetched %d bars for %s @ %s from Oanda", - len(df), instrument, granularity) - return df - - def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool: - if df.empty: - return False - now = pd.Timestamp.now(tz="UTC") - if _is_forex_weekend(now): - return True - candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) - dur = pd.Timedelta(seconds=candle_seconds) - expected_label = now.floor(dur) - dur - return df.index[-1] >= expected_label - - def _is_data_fresh( - self, df: pd.DataFrame, symbol: str, timeframe: Timeframe - ) -> bool: - if df is None or df.empty: - return False - now = pd.Timestamp.now(tz="UTC") - if _is_forex_weekend(now): - return True - last_bar = df.index[-1] - candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) - max_age_seconds = candle_seconds * 3 - age_seconds = (now - last_bar).total_seconds() - if age_seconds > max_age_seconds: - logger.warning( - "%s %s: last bar %s is %dm old (limit: %dm)", - symbol, timeframe.value, - last_bar.strftime("%Y-%m-%d %H:%M"), - age_seconds / 60, max_age_seconds / 60) - return False - return True - - def _validate_and_clean( - self, df: pd.DataFrame, symbol: str, timeframe: Timeframe - ) -> pd.DataFrame: - if df is None or df.empty: - raise OandaDataError(f"Empty dataset for {symbol}") - - original_len = len(df) - - df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all") - df = df.ffill(limit=3) - df = df.dropna(subset=["Close"]) - - df = df.sort_index() - df = df[~df.index.duplicated(keep="last")] - - now = pd.Timestamp.utcnow() - future_mask = df.index > now + pd.Timedelta(minutes=5) - if future_mask.any(): - logger.warning( - "%s: Dropping %d future timestamps", symbol, future_mask.sum()) - df = df[~future_mask] - - invalid_ohlc = ( - (df["High"] < df["Low"]) - | (df["High"] < df["Open"]) - | (df["High"] < df["Close"]) - | (df["Low"] > df["Open"]) - | (df["Low"] > df["Close"]) - ) - if invalid_ohlc.any(): - logger.warning( - "%s: Fixing %d OHLC-inconsistent bars", - symbol, invalid_ohlc.sum()) - for idx in df[invalid_ohlc].index: - vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]] - df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]]) - df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) - - returns = df["Close"].pct_change() - std_ret = returns.std() - if std_ret > 0: - dev = returns - returns.mean() - outlier_mask = dev.abs() > 10 * std_ret - recovery = ( - outlier_mask - & outlier_mask.shift(1, fill_value=False) - & (dev * dev.shift(1) < 0) - ) - outlier_mask = outlier_mask & ~recovery - if outlier_mask.any(): - logger.warning( - "%s: Removing %d extreme outlier bars", - symbol, outlier_mask.sum()) - df = df[~outlier_mask] - - min_required = max(self.config.technical.ema_trend, 200) + 10 - if len(df) < min_required: - logger.warning( - "%s %s: Only %d bars (need %d). Some indicators will be partial.", - symbol, timeframe.value, len(df), min_required) - - cleaned = original_len - len(df) - if cleaned > 0: - logger.info("%s: Cleaned %d/%d bars", symbol, cleaned, original_len) - - return df +from providers.base import _is_forex_weekend # noqa: F401 +from providers.oanda import OandaDataError, OandaDataProvider # noqa: F401 diff --git a/pair_signals/__init__.py b/pair_signals/__init__.py index 95440de..5dd873a 100644 --- a/pair_signals/__init__.py +++ b/pair_signals/__init__.py @@ -12,7 +12,7 @@ from typing import Optional, List, Dict, Callable from config import PlatformConfig, Timeframe from signal_aggregator import SignalAggregator, SignalReport -from data_provider import ForexDataProvider +from providers import BaseDataProvider PAIR_PRODUCERS: Dict[str, Callable] = {} @@ -28,7 +28,7 @@ def get_supported_pairs() -> List[str]: def produce_signal( symbol: str, primary_tf: Timeframe, - data_provider: ForexDataProvider, + data_provider: BaseDataProvider, confirmation_tfs: Optional[List[Timeframe]] = None, ) -> SignalReport: config_factory = PAIR_PRODUCERS.get(symbol) diff --git a/providers/__init__.py b/providers/__init__.py new file mode 100644 index 0000000..5a60083 --- /dev/null +++ b/providers/__init__.py @@ -0,0 +1,54 @@ +""" +ForexBot2 — Data-provider package +================================== +One class per provider, all subclassing `BaseDataProvider` (which owns +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. +""" + +from typing import Dict + +from app_config import AppConfigError, get_app_config +from config import PlatformConfig + +from providers.base import BaseDataProvider, DataIntegrityError, _is_forex_weekend +from providers.twelve_data import TwelveDataProvider +from providers.oanda import OandaDataError, OandaDataProvider +from providers.tiingo import TiingoDataError, TiingoDataProvider + +#: config.json data_providers.active selects from here. +PROVIDER_REGISTRY: Dict[str, type] = { + "twelve_data": TwelveDataProvider, + "oanda": OandaDataProvider, + "tiingo": TiingoDataProvider, +} + + +def make_data_provider(config: PlatformConfig) -> BaseDataProvider: + """Build the ACTIVE data provider (config.json: data_providers.active).""" + name = get_app_config().active_provider + cls = PROVIDER_REGISTRY.get(name) + if cls is None: + raise AppConfigError( + f'Unknown data provider "{name}"; ' + f"registered: {sorted(PROVIDER_REGISTRY)}") + return cls(config) + + +__all__ = [ + "BaseDataProvider", + "DataIntegrityError", + "OandaDataError", + "OandaDataProvider", + "TiingoDataError", + "TiingoDataProvider", + "TwelveDataProvider", + "PROVIDER_REGISTRY", + "make_data_provider", + "_is_forex_weekend", +] diff --git a/providers/base.py b/providers/base.py new file mode 100644 index 0000000..aac87a6 --- /dev/null +++ b/providers/base.py @@ -0,0 +1,423 @@ +""" +ForexBot2 — Provider base class +================================ +`BaseDataProvider` owns everything that must behave identically no matter +where the bars come from: in-memory caching with candle-duration TTL, the +closed-bar boundary guarantee, freshness checks, validation/cleaning, the +forming-candle drop, and paginated date-range fetching for the backtester. + +A concrete provider implements only: + _fetch_raw(symbol, timeframe, outputsize, start, end) -> pd.DataFrame +returning a normalized frame (capitalized OHLCV columns, UTC tz-aware +DatetimeIndex named "Datetime", oldest-first), plus optionally +`get_spread` and the streaming-quote capability (`supports_streaming` + +`iter_ticks`). + +The signal pipeline (engine + aggregator + backtester) consumes ONLY the +public contract defined here, so swapping providers never touches signal +code. +""" + +import logging +import threading +import time +from abc import ABC, abstractmethod +from datetime import datetime, timedelta +from typing import AsyncIterator, Callable, Dict, Optional, Sequence, Tuple + +import pandas as pd + +from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD + +logger = logging.getLogger(__name__) + + +def _is_forex_weekend(now=None) -> bool: + """Forex market closed: Friday 22:00 UTC → Sunday 22:00 UTC.""" + if now is None: + now = pd.Timestamp.now(tz="UTC") + wd = now.weekday() + hour = now.hour + if wd == 4 and hour >= 22: + return True + if wd == 5: + return True + if wd == 6 and hour < 22: + return True + return False + + +class DataIntegrityError(Exception): + """Raised when data fails validation checks.""" + pass + + +class BaseDataProvider(ABC): + """Shared OHLCV provider behavior; subclasses supply the raw fetch.""" + + #: Registry name; overridden per provider. + name: str = "base" + + #: Exception type this provider raises for fetch/validation failures. + #: Provider-specific errors subclass DataIntegrityError so consumers + #: can catch one type regardless of the active provider. + DATA_ERROR = DataIntegrityError + + BARS_PER_DAY = { + Timeframe.M1: 1440, Timeframe.M5: 288, Timeframe.M15: 96, + Timeframe.M30: 48, Timeframe.H1: 24, Timeframe.H4: 6, + Timeframe.D1: 1, Timeframe.W1: 0.143, + } + + CANDLE_SECONDS = { + Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900, + Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, + Timeframe.D1: 86400, Timeframe.W1: 604800, + } + + #: Hard per-request bar cap shared by Twelve Data and Oanda. + MAX_BARS_PER_REQUEST = 5000 + + # Grace wait before the one retry when a fresh response at a candle + # boundary is missing the just-closed bar (feed latency). + FRESH_FETCH_RETRY_SECONDS = 3.0 + + def __init__(self, config: PlatformConfig): + self.config = config + self._cache: Dict[str, pd.DataFrame] = {} + self._cache_timestamps: Dict[str, float] = {} + # The bot shares one provider across a thread pool; cache eviction + # is check-then-act and needs locking. + self._cache_lock = threading.Lock() + + # ── Abstract: the only provider-specific piece ───────────────── + + @abstractmethod + def _fetch_raw( + self, + symbol: str, + timeframe: Timeframe, + outputsize: int, + start: Optional[datetime] = None, + end: Optional[datetime] = None, + ) -> pd.DataFrame: + """Fetch up to `outputsize` bars, optionally bounded by start/end + (tz-aware UTC datetimes). Must return a normalized frame: + capitalized OHLCV columns, UTC DatetimeIndex named "Datetime", + oldest-first. Raise self.DATA_ERROR on failure.""" + + # ── Public API ───────────────────────────────────────────────── + + def fetch_ohlcv( + self, + symbol: str = XAUUSD, + timeframe: Timeframe = Timeframe.H1, + lookback_days: Optional[int] = None, + ) -> pd.DataFrame: + """Fetch OHLCV with in-memory caching for live mode.""" + if lookback_days is None: + lookback_days = self.config.data_lookback_days + + cache_key = f"{symbol}_{timeframe.value}" + now = time.time() + + cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval) + with self._cache_lock: + cached = self._cache.get(cache_key) + cached_at = self._cache_timestamps.get(cache_key) + if ( + cached is not None + and cached_at is not None + and (now - cached_at) < cache_ttl + and not cached.empty + ): + if (self._is_data_fresh(cached, symbol, timeframe) + and self._has_latest_closed_bar(cached, timeframe)): + return cached + logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching") + with self._cache_lock: + self._cache.pop(cache_key, None) + self._cache_timestamps.pop(cache_key, None) + + outputsize = max( + 300, + min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, + self.MAX_BARS_PER_REQUEST), + ) + df = self._fetch_raw(symbol, timeframe, outputsize) + df = self._drop_forming_candle(df, timeframe) + df = self._validate_and_clean(df, symbol, timeframe) + + # The closed-bar guarantee must hold for FRESH responses too, not + # just cache hits: the scanner fires seconds after the boundary, + # and a lagging feed can answer without the just-closed bar — the + # scan would silently analyze (and session-check) the previous + # bar. Give the feed one short grace retry, then proceed loudly. + if not df.empty and not self._has_latest_closed_bar(df, timeframe): + time.sleep(self.FRESH_FETCH_RETRY_SECONDS) + try: + retry = self._fetch_raw(symbol, timeframe, outputsize) + retry = self._drop_forming_candle(retry, timeframe) + retry = self._validate_and_clean(retry, symbol, timeframe) + if retry is not None and not retry.empty: + df = retry + except DataIntegrityError as e: + logger.warning( + f"{symbol} {timeframe.value}: boundary retry refetch " + f"failed ({e}) — keeping original data") + if not self._has_latest_closed_bar(df, timeframe): + logger.warning( + f"{symbol} {timeframe.value}: feed lagging — just-closed bar " + f"still missing after retry; this scan may run one bar behind" + ) + + if not self._is_data_fresh(df, symbol, timeframe): + logger.warning( + f"STALE DATA from API: {symbol} {timeframe.value} — " + f"last bar {df.index[-1]} is too old. Signal quality may be degraded." + ) + + with self._cache_lock: + self._cache[cache_key] = df + self._cache_timestamps[cache_key] = now + return df + + def fetch_range( + self, + symbol: str, + timeframe: Timeframe, + start: datetime, + end: datetime, + ) -> pd.DataFrame: + """Fetch the RAW closed-bar series between start/end, paginating + under the per-request bar cap. Returns uncleaned bars (the + backtester resolves trades on the raw price path); callers wanting + the signal-side series run `validate_and_clean` on the result.""" + bars_per_day = self.BARS_PER_DAY[timeframe] + days_per_chunk = max(1, int(4500 / bars_per_day)) + + chunks = [] + cursor_end = end + empty_streak = 0 + while cursor_end > start: + cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk)) + try: + chunk = self._fetch_raw( + symbol, timeframe, self.MAX_BARS_PER_REQUEST, + start=cursor_start, end=cursor_end, + ) + except Exception as e: + # Providers raise for windows with no data (weekends, + # holidays, asset gaps). Skip the window, step back, retry + # — but bail after a few consecutive empties. + logger.warning( + f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}" + ) + empty_streak += 1 + if empty_streak >= 3: + break + cursor_end = cursor_start - timedelta(seconds=1) + continue + + if chunk is None or chunk.empty: + empty_streak += 1 + if empty_streak >= 3: + break + cursor_end = cursor_start - timedelta(seconds=1) + continue + + empty_streak = 0 + chunks.append(chunk) + # Step back; dedup on concat. + cursor_end = chunk.index[0] - timedelta(seconds=1) + if len(chunk) < 2: + break + + if not chunks: + raise RuntimeError(f"No data fetched for {symbol} {timeframe.value}") + + df = pd.concat(chunks).sort_index() + df = df[~df.index.duplicated(keep="last")] + # Fetches ending "now" include the still-forming candle — drop it + # so callers only ever see closed bars. + df = self._drop_forming_candle(df, timeframe) + start_ts = pd.Timestamp(start) + end_ts = pd.Timestamp(end) + return df[(df.index >= start_ts) & (df.index <= end_ts)] + + def is_data_fresh( + self, df: pd.DataFrame, symbol: str, timeframe: Timeframe + ) -> bool: + """Public freshness check (last bar within 3 candle durations).""" + return self._is_data_fresh(df, symbol, timeframe) + + def validate_and_clean( + self, df: pd.DataFrame, symbol: str, timeframe: Timeframe + ) -> pd.DataFrame: + """Public entry to the shared cleaning pipeline.""" + return self._validate_and_clean(df, symbol, timeframe) + + def clear_cache(self) -> int: + """Wipe all cached data. Returns number of entries cleared.""" + with self._cache_lock: + count = len(self._cache) + self._cache.clear() + self._cache_timestamps.clear() + logger.info(f"Cache cleared: {count} entries wiped") + return count + + def get_pair_info(self, symbol: str = XAUUSD) -> Dict: + info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD]) + return {"symbol": symbol, **info} + + def get_spread(self, symbol: str = XAUUSD) -> float: + """Latest observed round-trip spread; providers without live + bid/ask fall back to the pair's assumed default.""" + return PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"] + + # ── Streaming-quote capability (optional) ────────────────────── + + #: True when the provider can stream live ticks via `iter_ticks`. + supports_streaming: bool = False + + def iter_ticks( + self, + symbols: Sequence[str], + on_connect: Optional[Callable[[], None]] = None, + ) -> AsyncIterator[Tuple[str, float]]: + """Async generator yielding (canonical_symbol, price) ticks for one + connection's lifetime; the caller owns reconnect/backoff. Providers + that support it set `supports_streaming = True` and override.""" + raise NotImplementedError( + f"{self.name} provider does not support streaming quotes") + + # ── Shared internals ─────────────────────────────────────────── + + def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool: + """Cached data must include the most recent fully closed bar. + + The TTL alone is wall-clock based: a snapshot fetched mid-candle is + still within TTL at the next candle boundary but is missing the bar + that just closed, so every boundary scan would analyze — and run the + bar-close session check against — the previous bar. + """ + if df is None or df.empty: + return False + now = pd.Timestamp.now(tz="UTC") + if _is_forex_weekend(now): + return True + candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) + dur = pd.Timedelta(seconds=candle_seconds) + expected_label = now.floor(dur) - dur + return df.index[-1] >= expected_label + + def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame: + """Drop the still-forming last candle so analysis only sees closed bars. + + Some feeds include the current partial candle; the scanner fires + seconds after the boundary, so without this guard every indicator + reads a seconds-old stub bar and signals repaint. Feeds that only + return complete candles pass through unchanged. + """ + if df is None or df.empty: + return df + candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) + now = pd.Timestamp.now(tz="UTC") + if df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now: + df = df.iloc[:-1] + return df + + def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool: + """Check if the last candle is within acceptable age for this timeframe.""" + if df is None or df.empty: + return False + + now = pd.Timestamp.now(tz="UTC") + + if _is_forex_weekend(now): + return True + + last_bar = df.index[-1] + candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) + max_age_seconds = candle_seconds * 3 + age_seconds = (now - last_bar).total_seconds() + + if age_seconds > max_age_seconds: + logger.warning( + f"{symbol} {timeframe.value}: last bar {last_bar.strftime('%Y-%m-%d %H:%M')} " + f"is {age_seconds / 60:.0f}m old (limit: {max_age_seconds / 60:.0f}m)" + ) + return False + + return True + + def _validate_and_clean( + self, df: pd.DataFrame, symbol: str, timeframe: Timeframe + ) -> pd.DataFrame: + if df is None or df.empty: + raise self.DATA_ERROR(f"Empty dataset for {symbol}") + + original_len = len(df) + + df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all") + df = df.ffill(limit=3) + df = df.dropna(subset=["Close"]) + + df = df.sort_index() + df = df[~df.index.duplicated(keep="last")] + + now = pd.Timestamp.utcnow() + future_mask = df.index > now + pd.Timedelta(minutes=5) + if future_mask.any(): + logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps") + df = df[~future_mask] + + invalid_ohlc = ( + (df["High"] < df["Low"]) | + (df["High"] < df["Open"]) | + (df["High"] < df["Close"]) | + (df["Low"] > df["Open"]) | + (df["Low"] > df["Close"]) + ) + if invalid_ohlc.any(): + logger.warning(f"{symbol}: Fixing {invalid_ohlc.sum()} OHLC-inconsistent bars") + for idx in df[invalid_ohlc].index: + vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]] + df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]]) + df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) + + returns = df["Close"].pct_change() + std_ret = returns.std() + if std_ret > 0: + dev = returns - returns.mean() + outlier_mask = dev.abs() > 10 * std_ret + # A spike bar makes the NEXT bar's return look extreme too (it is + # measured against the spike). If the previous bar is flagged and + # this bar reverses direction, it is the legitimate snap-back — + # keep it, or every spike would punch a two-bar hole in the data. + recovery = ( + outlier_mask + & outlier_mask.shift(1, fill_value=False) + & (dev * dev.shift(1) < 0) + ) + outlier_mask = outlier_mask & ~recovery + if outlier_mask.any(): + logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars") + df = df[~outlier_mask] + + min_required = max( + self.config.technical.ema_trend, + 200, + ) + 10 + + if len(df) < min_required: + logger.warning( + f"{symbol} {timeframe.value}: Only {len(df)} bars " + f"(need {min_required}). Some indicators will be partial." + ) + + cleaned = original_len - len(df) + if cleaned > 0: + logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars") + + return df diff --git a/providers/oanda.py b/providers/oanda.py new file mode 100644 index 0000000..1805f03 --- /dev/null +++ b/providers/oanda.py @@ -0,0 +1,192 @@ +""" +ForexBot2 — Oanda v20 data provider +==================================== +OHLCV candles from the Oanda v20 REST API. All shared behavior (cache, +validation, boundary guarantees, range pagination) lives in +`BaseDataProvider`; this module only maps symbols/granularities, parses +Oanda's candle JSON, and exposes the real bid/ask spread from MBA candles. +""" + +import logging +from datetime import datetime +from typing import Dict, Optional + +import pandas as pd +import requests + +from app_config import oanda_api_token, oanda_base_url +from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD + +from providers.base import BaseDataProvider, DataIntegrityError + +logger = logging.getLogger(__name__) + + +class OandaDataError(DataIntegrityError): + """Raised when an Oanda data fetch fails or returns invalid data.""" + + +class OandaDataProvider(BaseDataProvider): + """Oanda v20 REST candle provider.""" + + name = "oanda" + DATA_ERROR = OandaDataError + + OANDA_GRANULARITY = { + Timeframe.M1: "M1", + Timeframe.M5: "M5", + Timeframe.M15: "M15", + Timeframe.M30: "M30", + Timeframe.H1: "H1", + Timeframe.H4: "H4", + Timeframe.D1: "D", + Timeframe.W1: "W", + } + + OANDA_INSTRUMENT = { + "XAU/USD": "XAU_USD", + } + + def __init__( + self, + config: PlatformConfig, + api_token: Optional[str] = None, + base_url: Optional[str] = None, + ): + super().__init__(config) + # Credentials default to the config.json `oanda` block / env vars + # so the provider registry can build this class with config only. + if api_token is None: + api_token = oanda_api_token() + if base_url is None: + base_url = oanda_base_url() + self._base_url = base_url.rstrip("/") + self._session = requests.Session() + self._session.headers.update({ + "Authorization": f"Bearer {api_token}", + "Accept-Datetime-Format": "RFC3339", + "Content-Type": "application/json", + }) + self._spread_cache: Dict[str, float] = {} + + # ── Public extras ────────────────────────────────────────────── + + def get_spread(self, symbol: str = XAUUSD) -> float: + return self._spread_cache.get( + symbol, PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"]) + + # ── Raw fetch ────────────────────────────────────────────────── + + def _fetch_raw( + self, + symbol: str, + timeframe: Timeframe, + outputsize: int, + start: Optional[datetime] = None, + end: Optional[datetime] = None, + ) -> pd.DataFrame: + if start is not None or end is not None: + return self._fetch_oanda_candles( + symbol, timeframe, outputsize, start, end) + return self._fetch_oanda_candles(symbol, timeframe, outputsize) + + def _to_oanda_instrument(self, symbol: str) -> str: + if symbol in self.OANDA_INSTRUMENT: + return self.OANDA_INSTRUMENT[symbol] + return symbol.replace("/", "_") + + def _fetch_oanda_candles( + self, + symbol: str, + timeframe: Timeframe, + count: int, + start: Optional[datetime] = None, + end: Optional[datetime] = None, + ) -> pd.DataFrame: + instrument = self._to_oanda_instrument(symbol) + granularity = self.OANDA_GRANULARITY.get(timeframe) + if granularity is None: + raise OandaDataError( + f"Unsupported timeframe {timeframe} for Oanda") + + url = f"{self._base_url}/v3/instruments/{instrument}/candles" + params = { + "granularity": granularity, + "price": "MBA", + "dailyAlignment": 0, + "alignmentTimezone": "Etc/UTC", + } + # v20 rejects `count` combined with both `from` and `to`; a date + # window replaces it (used by fetch_range pagination). + if start is not None and end is not None: + params["from"] = pd.Timestamp(start).isoformat() + params["to"] = pd.Timestamp(end).isoformat() + else: + params["count"] = min(count, self.MAX_BARS_PER_REQUEST) + + try: + resp = self._session.get(url, params=params, timeout=30) + except requests.RequestException as e: + raise OandaDataError( + f"Oanda candle fetch failed for {instrument}: {e}") from e + + if resp.status_code != 200: + raise OandaDataError( + f"Oanda candle fetch failed for {instrument}: " + f"HTTP {resp.status_code}: {resp.text[:300]}") + + try: + data = resp.json() + except ValueError as e: + raise OandaDataError( + f"Oanda candle response for {instrument} is not valid " + f"JSON: {e}") from e + candles = data.get("candles", []) + if not candles: + raise OandaDataError( + f"Empty candle response from Oanda for {instrument}") + + rows = [] + last_bid_close = None + last_ask_close = None + for c in candles: + if not c.get("complete", False): + continue + mid = c.get("mid", {}) + bid = c.get("bid", {}) + ask = c.get("ask", {}) + rows.append({ + "Datetime": pd.Timestamp(c["time"]).tz_convert("UTC"), + "Open": float(mid.get("o", 0)), + "High": float(mid.get("h", 0)), + "Low": float(mid.get("l", 0)), + "Close": float(mid.get("c", 0)), + "Volume": int(c.get("volume", 0)), + }) + if bid and ask: + last_bid_close = float(bid.get("c", 0)) + last_ask_close = float(ask.get("c", 0)) + + if not rows: + raise OandaDataError( + f"No complete candles from Oanda for {instrument}") + + if last_bid_close and last_ask_close: + self._spread_cache[symbol] = round( + last_ask_close - last_bid_close, 4) + + 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") + df.index.name = "Datetime" + + for col in ("Open", "High", "Low", "Close"): + df[col] = pd.to_numeric(df[col], errors="coerce") + df["Volume"] = pd.to_numeric(df["Volume"], errors="coerce").fillna(0).astype(float) + + logger.debug( + "Fetched %d bars for %s @ %s from Oanda", + len(df), instrument, granularity) + return df diff --git a/providers/tiingo.py b/providers/tiingo.py new file mode 100644 index 0000000..14fbcfe --- /dev/null +++ b/providers/tiingo.py @@ -0,0 +1,200 @@ +""" +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: this is the CANDLE feed 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. +""" + +import logging +from datetime import datetime +from typing import Optional + +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 diff --git a/providers/twelve_data.py b/providers/twelve_data.py new file mode 100644 index 0000000..288ff7a --- /dev/null +++ b/providers/twelve_data.py @@ -0,0 +1,238 @@ +""" +ForexBot2 — Twelve Data provider +================================= +OHLCV via the official `twelvedata` SDK plus streaming quotes over the +Twelve Data WebSocket. All shared behavior (cache, validation, boundary +guarantees, range pagination) lives in `BaseDataProvider`. +""" + +import asyncio +import json +import logging +import threading +import time +from collections import deque +from datetime import datetime +from typing import AsyncIterator, Callable, Optional, Sequence, Tuple + +import pandas as pd +from twelvedata import TDClient + +from app_config import provider_api_key +from config import PAIR_INFO, PlatformConfig, Timeframe + +from providers.base import BaseDataProvider, DataIntegrityError + +logger = logging.getLogger(__name__) + + +class TwelveDataProvider(BaseDataProvider): + """Twelve Data SDK client with credit throttling and WS quotes.""" + + name = "twelve_data" + + TD_INTERVALS = { + Timeframe.M1: "1min", + Timeframe.M5: "5min", + Timeframe.M15: "15min", + Timeframe.M30: "30min", + Timeframe.H1: "1h", + Timeframe.H4: "4h", + Timeframe.D1: "1day", + Timeframe.W1: "1week", + } + + RATE_LIMIT = 55 + RATE_WINDOW = 60.0 + + WS_URL = "wss://ws.twelvedata.com/v1/quotes/price?apikey={api_key}" + + def __init__(self, config: PlatformConfig): + super().__init__(config) + self._api_key = provider_api_key(provider="twelve_data") or None + self._throttle_lock = threading.Lock() + self._request_times: deque = deque() + self._td: Optional[TDClient] = ( + TDClient(apikey=self._api_key) if self._api_key else None + ) + + @staticmethod + def _to_td_symbol(symbol: str) -> str: + if "/" in symbol: + return symbol + clean = symbol.replace("=X", "") + return f"{clean[:3]}/{clean[3:]}" + + # ── Raw fetch ────────────────────────────────────────────────── + + def _fetch_raw( + self, + symbol: str, + timeframe: Timeframe, + outputsize: int, + start: Optional[datetime] = None, + end: Optional[datetime] = None, + ) -> pd.DataFrame: + return self._fetch_twelve_data( + symbol, timeframe, outputsize, + start_date=start.strftime("%Y-%m-%d %H:%M:%S") if start else None, + end_date=end.strftime("%Y-%m-%d %H:%M:%S") if end else None, + ) + + def _throttle(self, credits: int = 1) -> None: + """Pro tier: 55 credits / minute. Block until budget allows.""" + while True: + with self._throttle_lock: + now = time.time() + while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: + self._request_times.popleft() + + available = self.RATE_LIMIT - len(self._request_times) + if available >= credits: + for _ in range(credits): + self._request_times.append(time.time()) + return + + used = len(self._request_times) + wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2 + logger.info( + f"Rate limit: {used}/{self.RATE_LIMIT} credits used " + f"— waiting {wait:.1f}s..." + ) + time.sleep(wait) + + def _fetch_twelve_data( + self, + symbol: str, + timeframe: Timeframe, + outputsize: int, + start_date: Optional[str] = None, + end_date: Optional[str] = None, + ) -> pd.DataFrame: + if self._td is None: + raise DataIntegrityError( + "No data-provider API key. Set the TWELVE_DATA_API_KEY env " + "var or data_providers.twelve_data.api_key in config.json." + ) + + td_symbol = self._to_td_symbol(symbol) + interval = self.TD_INTERVALS[timeframe] + + self._throttle(credits=1) + + kwargs = { + "symbol": td_symbol, + "interval": interval, + "outputsize": min(outputsize, self.MAX_BARS_PER_REQUEST), + "timezone": "UTC", + } + if start_date: + kwargs["start_date"] = start_date + if end_date: + kwargs["end_date"] = end_date + + try: + df = self._td.time_series(**kwargs).as_pandas() + except Exception as e: + raise DataIntegrityError( + f"Twelve Data fetch failed for {td_symbol}: {e}" + ) from e + + if df is None or df.empty: + raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}") + + # SDK returns lowercase columns; engine expects capitalised. + df = df.rename(columns={ + "open": "Open", "high": "High", "low": "Low", + "close": "Close", "volume": "Volume", + }) + for col in ("Open", "High", "Low", "Close", "Volume"): + if col in df.columns: + df[col] = pd.to_numeric(df[col], errors="coerce") + if "Volume" not in df.columns: + df["Volume"] = 0.0 + + # SDK returns newest-first; engine + indicators want oldest-first. + df = df.sort_index() + if df.index.tz is None: + df.index = df.index.tz_localize("UTC") + df.index.name = "Datetime" + + logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") + return df + + # ── Streaming quotes ─────────────────────────────────────────── + + supports_streaming = True + + @staticmethod + def _parse_ws_price(raw: str) -> Optional[Tuple[str, float]]: + """Parse one WS message; return (ws_symbol, price) for price + events, None for everything else (logged where noteworthy).""" + try: + msg = json.loads(raw) + except json.JSONDecodeError: + logger.warning("WebSocket: malformed JSON, skipping") + return None + + event = msg.get("event") + if event == "subscribe-status": + fails = msg.get("fails") or [] + if fails: + logger.warning(f"WebSocket: failed to subscribe: {fails}") + return None + if event != "price": + return None + + ws_symbol = msg.get("symbol") + try: + price = float(msg["price"]) + except (KeyError, ValueError, TypeError) as e: + logger.warning(f"WebSocket: bad price data for {ws_symbol}: {e}") + return None + return ws_symbol, price + + @staticmethod + async def _ws_heartbeat(ws) -> None: + """Twelve Data drops idle WS connections; heartbeat every 10s keeps + it alive through quiet periods (weekends, low-liquidity hours).""" + while True: + await asyncio.sleep(10) + await ws.send(json.dumps({"action": "heartbeat"})) + + 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, price) ticks. + + The TD WS speaks display symbols ("XAU/USD"); ticks are mapped back + to the canonical pair symbols passed in. The caller owns the + reconnect/backoff loop around this generator. + """ + import websockets + + ws_to_pair = {PAIR_INFO[s]["display"]: s for s in symbols} + url = self.WS_URL.format(api_key=self._api_key or "") + async with websockets.connect(url) as ws: + subscribe = { + "action": "subscribe", + "params": {"symbols": ",".join(ws_to_pair)}, + } + await ws.send(json.dumps(subscribe)) + logger.info(f"WebSocket connected, subscribed to {len(ws_to_pair)} symbols") + if on_connect is not None: + on_connect() + hb_task = asyncio.create_task(self._ws_heartbeat(ws)) + try: + async for raw in ws: + parsed = self._parse_ws_price(raw) + if parsed is None: + continue + pair_symbol = ws_to_pair.get(parsed[0]) + if pair_symbol is None: + continue + yield pair_symbol, parsed[1] + finally: + hb_task.cancel() diff --git a/signal_aggregator.py b/signal_aggregator.py index 9afb7b5..1a646f0 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -15,7 +15,7 @@ import pandas as pd from chart_patterns import detect_chart_pattern from config import PlatformConfig, SignalStrength, Timeframe -from data_provider import ForexDataProvider, make_data_provider +from providers import BaseDataProvider, make_data_provider from engine import TechnicalEngine, TechnicalAnalysis from news_engine import NewsEngine, FundamentalContext @@ -180,7 +180,7 @@ class SignalAggregator: try: df = self.data_provider.fetch_ohlcv(symbol, tf) dfs[tf] = df - if not self.data_provider._is_data_fresh(df, symbol, tf): + if not self.data_provider.is_data_fresh(df, symbol, tf): stale_tfs.append(tf.value) except Exception as e: logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}") @@ -214,7 +214,7 @@ class SignalAggregator: bar_close = ( pd.Timestamp(report.timestamp) + pd.Timedelta( - seconds=ForexDataProvider.CANDLE_SECONDS[ + seconds=BaseDataProvider.CANDLE_SECONDS[ self.config.primary_timeframe ] ) @@ -637,7 +637,7 @@ class SignalAggregator: close_h = ( pd.Timestamp(report.timestamp) + pd.Timedelta( - seconds=ForexDataProvider.CANDLE_SECONDS[ + seconds=BaseDataProvider.CANDLE_SECONDS[ self.config.primary_timeframe ] ) @@ -939,7 +939,7 @@ class SignalAggregator: close_h = ( pd.Timestamp(report.timestamp) + pd.Timedelta( - seconds=ForexDataProvider.CANDLE_SECONDS[ + seconds=BaseDataProvider.CANDLE_SECONDS[ self.config.primary_timeframe ] ) diff --git a/telegram_bot.py b/telegram_bot.py index 8306bf2..2ba0f06 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -38,7 +38,6 @@ Usage: import asyncio import functools -import json import logging import logging.handlers import os @@ -49,8 +48,6 @@ from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from typing import Dict, List, Optional, Set -import websockets - from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.constants import ParseMode from telegram.error import BadRequest @@ -66,10 +63,10 @@ from config import ( XAUUSD, PAIR_INFO, ) from app_config import ( - get_app_config, provider_api_key, telegram_token, telegram_chat_id, + get_app_config, telegram_token, telegram_chat_id, telegram_channel_id, ) -from data_provider import ForexDataProvider, make_data_provider +from providers import BaseDataProvider, make_data_provider from pair_signals import PAIR_PRODUCERS from signal_aggregator import SignalAggregator, SignalReport @@ -127,9 +124,6 @@ SCAN_PAIRS = list(_APP.scan_pairs) _PAIRS_LABEL = " + ".join(PAIR_INFO[p]["display"] for p in SCAN_PAIRS) -_WS_TO_PAIR = {PAIR_INFO[p]["display"]: p for p in SCAN_PAIRS} -_WS_SYMBOLS = list(_WS_TO_PAIR.keys()) - _LEVEL_KEYS = ( "swing_high_1", "swing_high_2", "swing_high_3", "swing_low_1", "swing_low_2", "swing_low_3", @@ -199,10 +193,10 @@ def _fmt_price(price: float) -> str: # ── Shared data provider (single cache across all scans) ───────────── -_shared_dp: Optional[ForexDataProvider] = None +_shared_dp: Optional[BaseDataProvider] = None -def _get_shared_dp() -> ForexDataProvider: +def _get_shared_dp() -> BaseDataProvider: global _shared_dp if _shared_dp is None: _shared_dp = make_data_provider(PlatformConfig()) @@ -663,8 +657,6 @@ _last_price_ts: Dict[str, float] = {} _scan_cooldowns: Dict[str, float] = {} _cfg = PlatformConfig() -_WS_API_KEY = provider_api_key() -_WS_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={_WS_API_KEY}" _WS_COOLDOWN = _cfg.ws_cooldown_seconds _breach_tasks: Set[asyncio.Task] = set() @@ -710,13 +702,13 @@ def _cleanup_cooldowns() -> None: del _scan_cooldowns[k] -async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, level_name: str) -> None: +async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) -> None: now = time.time() if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN: return _scan_cooldowns[pair_symbol] = now - display = PAIR_INFO.get(pair_symbol, {}).get("display", ws_symbol) + display = PAIR_INFO.get(pair_symbol, {}).get("display", pair_symbol) label = level_name.replace("_", " ").title() logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15") @@ -759,71 +751,41 @@ def _breach_task_done(task: asyncio.Task) -> None: logger.error("Breach alert task failed", exc_info=task.exception()) -async def _ws_heartbeat(ws) -> None: - """Twelve Data drops idle WS connections; heartbeat every 10s keeps it - alive through quiet periods (weekends, low-liquidity hours).""" - while True: - await asyncio.sleep(10) - await ws.send(json.dumps({"action": "heartbeat"})) - - 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() + if not dp.supports_streaming: + logger.info( + "Data provider %r has no streaming quotes — " + "level-breach monitor disabled", dp.name) + return + backoff = 1 while _feed_active(): - hb_task: Optional[asyncio.Task] = None try: - async with websockets.connect(_WS_URL) as ws: - subscribe = { - "action": "subscribe", - "params": {"symbols": ",".join(_WS_SYMBOLS)}, - } - await ws.send(json.dumps(subscribe)) - logger.info(f"WebSocket connected, subscribed to {len(_WS_SYMBOLS)} symbols") + def _on_connect() -> None: + nonlocal backoff backoff = 1 - hb_task = asyncio.create_task(_ws_heartbeat(ws)) - async for raw in ws: - if not _feed_active(): - break + async for pair_symbol, price in dp.iter_ticks( + SCAN_PAIRS, on_connect=_on_connect): + if not _feed_active(): + break - try: - msg = json.loads(raw) - except json.JSONDecodeError: - logger.warning(f"WebSocket: malformed JSON, skipping") - continue + breached = _check_level_breach(pair_symbol, price) + _last_prices[pair_symbol] = price + _last_price_ts[pair_symbol] = time.time() - event = msg.get("event") - if event == "subscribe-status": - fails = msg.get("fails") or [] - if fails: - logger.warning(f"WebSocket: failed to subscribe: {fails}") - continue - if event != "price": - continue + if breached: + task = asyncio.create_task( + _handle_breach(bot, pair_symbol, price, breached) + ) + _breach_tasks.add(task) + task.add_done_callback(_breach_task_done) - ws_symbol = msg.get("symbol") - pair_symbol = _WS_TO_PAIR.get(ws_symbol) - if not pair_symbol: - continue - - try: - price = float(msg["price"]) - except (KeyError, ValueError, TypeError) as e: - logger.warning(f"WebSocket: bad price data for {ws_symbol}: {e}") - continue - - breached = _check_level_breach(pair_symbol, price) - _last_prices[pair_symbol] = price - _last_price_ts[pair_symbol] = time.time() - - if breached: - task = asyncio.create_task( - _handle_breach(bot, ws_symbol, pair_symbol, price, breached) - ) - _breach_tasks.add(task) - task.add_done_callback(_breach_task_done) - - _cleanup_cooldowns() + _cleanup_cooldowns() except asyncio.CancelledError: break @@ -831,9 +793,6 @@ async def _ws_monitor(bot) -> None: logger.exception(f"WebSocket error, reconnecting in {backoff}s") await asyncio.sleep(backoff) backoff = min(backoff * 2, 60) - finally: - if hb_task is not None: - hb_task.cancel() logger.info("WebSocket monitor stopped") @@ -998,9 +957,9 @@ def _status_text() -> str: if _last_prices: lines.append("\n*Live Prices (WebSocket):*") - for ws_sym, pair in _WS_TO_PAIR.items(): + for pair in SCAN_PAIRS: price = _last_prices.get(pair) - display = PAIR_INFO.get(pair, {}).get("display", ws_sym) + display = PAIR_INFO.get(pair, {}).get("display", pair) age = time.time() - _last_price_ts.get(pair, 0) if price is None: lines.append(f" {display}: waiting...") diff --git a/test_all.py b/test_all.py index 438f97b..87d4f1a 100644 --- a/test_all.py +++ b/test_all.py @@ -1561,7 +1561,6 @@ from telegram_bot import ( _seconds_until_boundary, _check_level_breach, scan_keyboard, SCAN_BUTTONS, SCAN_LOOKUP, _update_level_cache, _last_prices, _level_cache, _level_cache_ts, - _WS_TO_PAIR, ) @@ -1836,12 +1835,16 @@ def _(): _level_cache_ts.clear() -@test("_WS_TO_PAIR mapping covers every scanned pair") +@test("streaming symbol mapping covers every scanned pair") def _(): + # iter_ticks maps the provider's WS symbols (display names) back to + # canonical pair symbols — the mapping must be invertible for every + # scanned pair or ticks would be silently dropped. from telegram_bot import SCAN_PAIRS + ws_to_pair = {PAIR_INFO[p]["display"]: p for p in SCAN_PAIRS} for p in SCAN_PAIRS: - assert _WS_TO_PAIR[PAIR_INFO[p]["display"]] == p - assert _WS_TO_PAIR["XAU/USD"] == XAUUSD + assert ws_to_pair[PAIR_INFO[p]["display"]] == p + assert ws_to_pair["XAU/USD"] == XAUUSD # ══════════════════════════════════════════════════════════════════════ @@ -2290,7 +2293,7 @@ def _(): stale = bars_ending(last_closed - pd.Timedelta(minutes=15)) fresh = bars_ending(last_closed) key = "XAU/USD_15min" - with patch("data_provider._is_forex_weekend", return_value=False): + with patch("providers.base._is_forex_weekend", return_value=False): # (a) snapshot cached mid-candle 5 min ago, missing the # just-closed bar: TTL says fresh, boundary check must refetch. provider._cache[key] = stale @@ -2324,7 +2327,7 @@ def _(): last_closed = started - pd.Timedelta(minutes=15) lagging = bars_ending(last_closed - pd.Timedelta(minutes=15)) complete = bars_ending(last_closed) - with patch("data_provider._is_forex_weekend", return_value=False), \ + with patch("providers.base._is_forex_weekend", return_value=False), \ patch.object(ForexDataProvider, "FRESH_FETCH_RETRY_SECONDS", 0.0): # (a) lagging first response → one retry returns the closed bar. with patch.object(provider, "_fetch_twelve_data", @@ -3963,6 +3966,331 @@ def _(): daemon._heartbeat_once() # must not raise +# ══════════════════════════════════════════════════════════════════════ +# MODULE 16: Provider abstraction (providers/ package) +# ══════════════════════════════════════════════════════════════════════ + +print("\n═══ Module 16: Provider abstraction ═══") + +from providers import ( + BaseDataProvider, PROVIDER_REGISTRY, TwelveDataProvider, + make_data_provider, +) +from providers.oanda import OandaDataProvider as _POanda +from providers.tiingo import TiingoDataError, TiingoDataProvider + + +@test("providers: registry holds all providers, active default builds TD") +def _(): + assert PROVIDER_REGISTRY["twelve_data"] is TwelveDataProvider + assert PROVIDER_REGISTRY["oanda"] is _POanda + assert PROVIDER_REGISTRY["tiingo"] is TiingoDataProvider + dp = make_data_provider(PlatformConfig()) + assert isinstance(dp, TwelveDataProvider) + assert isinstance(dp, BaseDataProvider) + + +@test("providers: single knob — data_providers.active selects the class") +def _(): + from app_config import AppConfig + for name, cls in PROVIDER_REGISTRY.items(): + fake = AppConfig() + fake.active_provider = name + fake.provider_settings = {name: {"api_key": ""}} + with patch("providers.get_app_config", return_value=fake): + dp = make_data_provider(PlatformConfig()) + assert type(dp) is cls, f"active={name} built {type(dp).__name__}" + # An unknown name must fail loudly, not fall back silently. + from app_config import AppConfigError + fake = AppConfig() + fake.active_provider = "nonsense" + with patch("providers.get_app_config", return_value=fake): + try: + make_data_provider(PlatformConfig()) + assert False, "unknown provider must raise" + except AppConfigError as e: + assert "nonsense" in str(e) + + +@test("providers: shared behavior is defined ONCE on the base class") +def _(): + # The pre-refactor bug this guards against: Oanda's provider was a + # copy-paste of the TD one, so shared-logic fixes had to be made + # twice. Every shared method must now BE the base implementation. + for meth in ("fetch_ohlcv", "fetch_range", "_validate_and_clean", + "_is_data_fresh", "_has_latest_closed_bar", + "_drop_forming_candle", "clear_cache", "get_pair_info"): + for cls in PROVIDER_REGISTRY.values(): + assert getattr(cls, meth) is getattr(BaseDataProvider, meth), \ + f"{cls.__name__} overrides shared method {meth}" + assert TwelveDataProvider.CANDLE_SECONDS is BaseDataProvider.CANDLE_SECONDS + assert _POanda.BARS_PER_DAY is BaseDataProvider.BARS_PER_DAY + assert TiingoDataProvider.CANDLE_SECONDS is BaseDataProvider.CANDLE_SECONDS + + +@test("providers: back-compat shims re-export the real classes") +def _(): + import data_provider as dp_shim + import oanda_provider as oa_shim + assert dp_shim.ForexDataProvider is TwelveDataProvider + assert dp_shim.make_data_provider is make_data_provider + assert oa_shim.OandaDataProvider is _POanda + from providers.base import DataIntegrityError as base_die + assert dp_shim.DataIntegrityError is base_die + # Provider-specific errors are catchable as the shared type. + assert issubclass(oa_shim.OandaDataError, base_die) + + +@test("providers: fetch_range paginates, dedupes, and trims to the window") +def _(): + class FakeProvider(BaseDataProvider): + name = "fake" + + def __init__(self, config): + super().__init__(config) + idx = pd.date_range("2026-06-01", "2026-06-30 23:00", + freq="1h", tz="UTC") + v = np.linspace(3000.0, 3100.0, len(idx)) + self.series = pd.DataFrame( + {"Open": v, "High": v + 1, "Low": v - 1, + "Close": v, "Volume": 100.0}, index=idx) + self.series.index.name = "Datetime" + self.calls = 0 + + def _fetch_raw(self, symbol, timeframe, outputsize, + start=None, end=None): + self.calls += 1 + df = self.series + if start is not None: + df = df[df.index >= pd.Timestamp(start)] + if end is not None: + df = df[df.index <= pd.Timestamp(end)] + # Overlap by a bar to prove concat dedupes. + return df.tail(min(len(df), 200)) + + # Keep the test date-stable: nothing here is "forming". + def _drop_forming_candle(self, df, timeframe): + return df + + fp = FakeProvider(PlatformConfig()) + start = datetime(2026, 6, 5, tzinfo=timezone.utc) + end = datetime(2026, 6, 25, tzinfo=timezone.utc) + df = fp.fetch_range("XAU/USD", Timeframe.H1, start, end) + + assert fp.calls > 1, "a 20-day H1 window must take multiple chunks" + assert df.index.is_monotonic_increasing + assert not df.index.duplicated().any() + assert df.index[0] >= pd.Timestamp(start) + assert df.index[-1] <= pd.Timestamp(end) + expected = fp.series[(fp.series.index >= pd.Timestamp(start)) + & (fp.series.index <= pd.Timestamp(end))] + assert len(df) == len(expected), \ + f"window must be gapless: {len(df)} vs {len(expected)}" + + +@test("providers: streaming capability flags and base refusal") +def _(): + assert TwelveDataProvider.supports_streaming is True + assert _POanda.supports_streaming is False + assert TiingoDataProvider.supports_streaming is False + + class Minimal(BaseDataProvider): + name = "minimal" + + def _fetch_raw(self, symbol, timeframe, outputsize, + start=None, end=None): + raise NotImplementedError + + m = Minimal(PlatformConfig()) + assert m.supports_streaming is False + try: + m.iter_ticks(["XAU/USD"]) + assert False, "base iter_ticks must refuse" + except NotImplementedError: + pass + + +@test("providers: TD WS price-message parsing") +def _(): + parse = TwelveDataProvider._parse_ws_price + assert parse('{"event":"price","symbol":"XAU/USD","price":"3345.5"}') \ + == ("XAU/USD", 3345.5) + assert parse('{"event":"subscribe-status","fails":[]}') is None + assert parse('{"event":"heartbeat"}') is None + assert parse('not json{{') is None + assert parse('{"event":"price","symbol":"XAU/USD"}') is None # no price + assert parse('{"event":"price","symbol":"XAU/USD","price":"abc"}') is None + + +@test("providers: Oanda date-window fetch omits count and sends from/to") +def _(): + captured = {} + + def mock_get(self, url, params=None, timeout=None): + captured["params"] = params + return type("R", (), { + "status_code": 200, + "json": lambda self: {"candles": [{ + "complete": True, + "time": "2026-06-02T10:00:00.000000000Z", + "mid": {"o": "3000", "h": "3001", "l": "2999", "c": "3000.5"}, + "bid": {"c": "3000.2"}, "ask": {"c": "3000.8"}, + "volume": 10, + }]}, + })() + + provider = _POanda.__new__(_POanda) + provider.config = PlatformConfig() + provider._base_url = "https://test" + provider._spread_cache = {} + provider._session = type("S", (), {"get": mock_get})() + + start = datetime(2026, 6, 1, tzinfo=timezone.utc) + end = datetime(2026, 6, 3, tzinfo=timezone.utc) + df = provider._fetch_raw("XAU/USD", Timeframe.H1, 5000, start, end) + assert "count" not in captured["params"], \ + "v20 rejects count combined with from/to" + assert captured["params"]["from"].startswith("2026-06-01") + assert captured["params"]["to"].startswith("2026-06-03") + assert len(df) == 1 + + # Live path (no window) keeps sending count only. + provider._fetch_raw("XAU/USD", Timeframe.H1, 300) + assert captured["params"]["count"] == 300 + assert "from" not in captured["params"] + + +def _tiingo_provider(response): + """TiingoDataProvider with a canned session response (no network).""" + provider = TiingoDataProvider.__new__(TiingoDataProvider) + provider.config = PlatformConfig() + provider._api_key = "test-key" + provider._spread_cache = {} + captured = {} + + def mock_get(self, url, params=None, timeout=None): + captured["url"] = url + captured["params"] = params + return response + + provider._session = type("S", (), {"get": mock_get})() + return provider, captured + + +def _tiingo_resp(status_code, payload): + return type("R", (), { + "status_code": status_code, + "json": lambda self: payload, + "text": json.dumps(payload) if not isinstance(payload, str) else payload, + })() + + +@test("providers: Tiingo parses FX JSON into the normalized frame") +def _(): + payload = [ + {"date": "2026-07-20T11:00:00.000Z", "ticker": "xauusd", + "open": 2403.0, "high": 2408.0, "low": 2401.0, "close": 2406.0}, + # Deliberately out of order — provider must sort oldest-first. + {"date": "2026-07-20T10:00:00.000Z", "ticker": "xauusd", + "open": 2400.0, "high": 2405.0, "low": 2398.0, "close": 2403.0}, + ] + provider, captured = _tiingo_provider(_tiingo_resp(200, payload)) + df = provider._fetch_raw("XAU/USD", Timeframe.H1, 300) + + assert list(df.columns) == ["Open", "High", "Low", "Close", "Volume"] + assert df.index.name == "Datetime" + assert df.index.tz is not None, "Index must be tz-aware UTC" + assert df.index.is_monotonic_increasing, "must be oldest-first" + assert df.iloc[0]["Open"] == 2400.0 and df.iloc[1]["Close"] == 2406.0 + assert (df["Volume"] == 0.0).all(), "FX endpoint has no volume" + assert captured["url"].endswith("/tiingo/fx/xauusd/prices") + assert captured["params"]["resampleFreq"] == "1hour" + assert "startDate" in captured["params"], \ + "live fetch must derive a start window from the bar count" + assert "token" not in captured["params"], \ + "key must travel in the auth header, never the URL" + + +@test("providers: Tiingo date-window fetch sends startDate AND endDate") +def _(): + payload = [{"date": "2026-06-02T10:00:00.000Z", + "open": 1, "high": 2, "low": 0, "close": 1}] + provider, captured = _tiingo_provider(_tiingo_resp(200, payload)) + start = datetime(2026, 6, 1, tzinfo=timezone.utc) + end = datetime(2026, 6, 3, tzinfo=timezone.utc) + provider._fetch_raw("XAU/USD", Timeframe.H1, 5000, start, end) + assert captured["params"]["startDate"].startswith("2026-06-01") + assert captured["params"]["endDate"].startswith("2026-06-03") + + +@test("providers: Tiingo fails LOUDLY on every bad-response shape") +def _(): + cases = [ + (_tiingo_resp(429, "quota"), "rate limit"), + (_tiingo_resp(401, "denied"), "auth failed"), + (_tiingo_resp(500, "boom"), "HTTP 500"), + (_tiingo_resp(200, []), "Empty candle response"), + (_tiingo_resp(200, [{"open": 1}]), "missing 'date'"), + (_tiingo_resp(200, [{"date": "2026-07-20T10:00:00Z", + "close": "n/a"}]), "no numeric closes"), + ] + for resp, needle in cases: + provider, _cap = _tiingo_provider(resp) + try: + provider._fetch_raw("XAU/USD", Timeframe.H1, 300) + assert False, f"must raise for: {needle}" + except TiingoDataError as e: + assert needle.lower() in str(e).lower(), \ + f"error must name the cause ({needle}): {e}" + + # Missing key and unsupported timeframe are named errors too. + provider, _cap = _tiingo_provider(_tiingo_resp(200, [])) + provider._api_key = None + try: + provider._fetch_raw("XAU/USD", Timeframe.H1, 300) + assert False, "missing key must raise" + except TiingoDataError as e: + assert "TIINGO_API_KEY" in str(e) + provider._api_key = "k" + try: + provider._fetch_raw("XAU/USD", Timeframe.W1, 300) + assert False, "unsupported TF must raise" + except TiingoDataError as e: + assert "Unsupported timeframe" in str(e) + # And everything above is catchable as the shared base error. + assert issubclass(TiingoDataError, DataIntegrityError) + + +@test("providers: per-provider env keys (TIINGO_API_KEY vs TWELVE_DATA_API_KEY)") +def _(): + from app_config import AppConfig, provider_api_key + app = AppConfig() + app.active_provider = "tiingo" + app.provider_settings = {"tiingo": {"api_key": "json-tiingo"}, + "twelve_data": {"api_key": "json-td"}} + with patch.dict(os.environ, {"TIINGO_API_KEY": "env-tiingo", + "TWELVE_DATA_API_KEY": "env-td"}): + assert provider_api_key(app) == "env-tiingo" + assert provider_api_key(app, provider="twelve_data") == "env-td" + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("TIINGO_API_KEY", None) + assert provider_api_key(app) == "json-tiingo" + # The TD env var must never leak into another provider's key. + with patch.dict(os.environ, {"TWELVE_DATA_API_KEY": "env-td"}): + assert provider_api_key(app, provider="tiingo") == "json-tiingo" + + +@test("providers: aggregator bar-close clock uses the base-class table") +def _(): + # signal_aggregator computes bar-CLOSE hours (session filter, SELL + # hour gate) from BaseDataProvider.CANDLE_SECONDS — a provider swap + # must never change the session clock. + import signal_aggregator as sa + assert sa.BaseDataProvider.CANDLE_SECONDS[Timeframe.M15] == 900 + assert not hasattr(sa, "ForexDataProvider"), \ + "aggregator must not depend on a concrete provider class" + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════