Merge branch 'centralized-config': config.json deployment overlay

Centralized deployment settings (Telegram credentials, data-provider
keys/selection, scan pairs/timeframes, per-pair base blocked hours,
alert quality, log level) into a gitignored config.json with env-var
precedence. Hardcoded credentials removed from source. Zero signal-
logic change — parity pinned by tests (200/200).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b

.gitignore

@@ -26,3 +26,5 @@ Thumbs.db
 backtest/cache/
 backtest/*.csv
 backtest/__pycache__/
+.env.*
+logs/

CLAUDE.md

@@ -13,6 +13,8 @@ This file is the authoritative developer/agent reference. The human-facing docs
 | Path | Lines | Purpose |
 |------|-------|---------|
 | `config.py` | ~374 | All config dataclasses, `PAIR_INFO` registry, per-pair strategy configs. **`PlatformConfig` defaults ARE the tuned XAU config.** |
+| `app_config.py` | ~244 | config.json loader/validation, env-precedence resolvers, `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 |
 | `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) |
@@ -22,7 +24,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` | ~2100 | QA suite: **184 tests in 12 modules**, no live API needed. Must stay green. |
+| `test_all.py` | ~2100 | QA suite: **200 tests in 13 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 |
@@ -108,7 +110,9 @@ Legacy fields kept for config-shape compatibility: `rsi_entry_threshold`, `min_e
 `session_filter=True`, window 03:00–23:00 UTC, `blocked_hours_utc` mirrors XAU's list (non-XAU pairs read this copy), `allowed_weekdays` Mon–Fri+Sun. `regime_filter=False` (optional H4-ADX≥25 gate, off).
 
 ### PlatformConfig (top level)
-`symbol="XAU/USD"`, primary TF H1, confirmations [M15, H4], `data_lookback_days=60`, `scan_pairs=[XAUUSD]` (**the single switch for what the bot scans**), `scan_timeframes=[M5,M15,H1,H4]`, `candle_buffer_seconds=10`, `ws_cooldown_seconds=300`, hardcoded `twelve_data_api_key` fallback.
+`symbol="XAU/USD"`, primary TF H1, confirmations [M15, H4], `data_lookback_days=60`, `scan_pairs=[XAUUSD]` (mirrors config.json `scanning.pairs` — see below), `scan_timeframes=[M5,M15,H1,H4]`, `candle_buffer_seconds=10`, `ws_cooldown_seconds=300`.
+
+Per-pair base blocked hours plus `scan_pairs`/`scan_timeframes` overlay from config.json via `app_config.apply_pair_overrides`, applied in the pair factory BEFORE per-TF deltas. Credentials, alert quality, and log level never touch `PlatformConfig` — `telegram_bot.py` and `data_provider.py` read them directly from `app_config.get_app_config()` and its resolvers. `scanning.pairs` in config.json is the deployment switch for what the bot scans — `PlatformConfig.scan_pairs` just carries the resolved value.
 
 ---
 
@@ -247,6 +251,8 @@ Two entry points:
 
 `PAIR_PRODUCERS: Dict[symbol, factory]`; factory signature `make_config(primary_tf, confirmation_tfs=None) -> PlatformConfig`. `produce_signal(symbol, tf, data_provider)` builds config → aggregator → `analyze_pair`. Registered: XAU/USD.
 
+Inside `make_config`, call order is: `PlatformConfig()` defaults → `app_config.apply_pair_overrides(config)` (stamps config.json's non-credential deployment settings onto the config: base `xau_strategy`/`backtest` `blocked_hours_utc`, `scan_pairs`, `scan_timeframes`) → the validated per-TF deltas below. The overlay runs first so e.g. the M5 hour-9 append lands on top of the JSON base, not the other way around.
+
 **A bare `PlatformConfig()` silently runs any pair with XAU defaults — always go through the factory** (the bot, CLI, and backtester all do).
 
 Default confirmation map (same in the factory, the bot, and backtest usage): M5→[M15,H1], M15→[M5,H1], H1→[M15,H4], H4→[H1,D1]. (Exception: `main.py`'s CLI map uses M5→[M1,M15] — a CLI-only quirk.)
@@ -261,7 +267,7 @@ Default confirmation map (same in the factory, the bot, and backtest usage): M5
 
 ## telegram_bot.py — the bot
 
-Run: `python telegram_bot.py` (env `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`; hardcoded fallbacks exist in the file).
+Run: `python telegram_bot.py` (credentials resolve env `TELEGRAM_TOKEN`/`TELEGRAM_CHAT_ID`/`TELEGRAM_CHANNEL_ID` → config.json `telegram.*` → refuses to boot; see Environment variables).
 
 - **Owner lock**: every command/button is wrapped in `_owner_only` — only `TELEGRAM_CHAT_ID` gets responses; others get a refusal. The bot is the owner's control panel; the product is the channel.
 - **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).
@@ -314,18 +320,20 @@ Data: paginated TD fetches (5000-bar chunks) cached as parquet in `backtest/cach
 
 ## test_all.py — QA suite
 
-`python test_all.py` — **184 tests, 12 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). 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), and a live-vs-GPU engine parity sweep incl. the EMA-ribbon path (Module 11). **Any signal-logic change must keep this green and should add tests.**
+`python test_all.py` — **200 tests, 13 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). 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), and config.json loading/precedence/validation coverage (Module 13). **Any signal-logic change must keep this green and should add tests.**
 
 ---
 
 ## Environment variables
 
+Precedence for every credential/deployment setting: **env var → config.json → none**. Hardcoded fallbacks (previously in `config.py` / `telegram_bot.py`) were **REMOVED 2026-07-21**; the bot now refuses to boot (`telegram_bot.main`) and data fetches fail with a named error (`data_provider`) if a secret resolves to neither source. See `app_config.py` and `config.json.example`.
+
 | Variable | Purpose | Default |
 |---|---|---|
-| `TWELVE_DATA_API_KEY` | market data key | hardcoded fallback in `config.py` |
-| `TELEGRAM_TOKEN` | bot token | hardcoded fallback in `telegram_bot.py` |
-| `TELEGRAM_CHAT_ID` | owner chat id (boot auto-scan + owner lock) | hardcoded fallback |
-| `TELEGRAM_CHANNEL_ID` | subscriber channel (`-100…`); `""` disables the feed | hardcoded fallback |
+| `TWELVE_DATA_API_KEY` | market data key | none — falls through to config.json `data_providers.twelve_data.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` |
 | `FB2_EMA_LO` | EMA200 "near" penalty threshold in % | `0.5` |
 
 Dependencies (`requirements.txt`): numpy, pandas, requests, twelvedata, python-telegram-bot ≥20, websockets. Python 3.12. A `venv/` exists in-repo.

app_config.py

@@ -0,0 +1,250 @@
+"""
+app_config.py — deployment-settings overlay (config.json)
+=========================================================
+
+Loads the gitignored config.json at the repo root ONCE per process.
+Deployment surface ONLY: Telegram credentials, data-provider selection
+and API keys, scanned pairs/timeframes, per-pair base blocked hours,
+alert-quality boot default, log level. Signal logic, weights,
+thresholds and gates stay in config.py / pair_signals.
+
+Precedence for every setting: env var → config.json → code default.
+Secrets have NO code default — with neither env nor config.json the
+bot refuses to start (telegram_bot.main) and data fetches fail with a
+named error (data_provider).
+
+Parity contract (test-enforced): with no config.json, or one holding
+the config.json.example defaults, every produced PlatformConfig is
+byte-identical to the pure-code defaults. apply_pair_overrides() runs
+in the pair factory BEFORE the validated per-timeframe deltas, so the
+bot, CLI and backtester all see the same overlay (live == backtest).
+
+Spec: docs/superpowers/specs/2026-07-21-centralized-config-design.md
+"""
+
+import json
+import logging
+import os
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD
+
+logger = logging.getLogger(__name__)
+
+CONFIG_PATH = os.path.join(
+    os.path.dirname(os.path.abspath(__file__)), "config.json")
+
+_VALID_QUALITIES = ("A+", "A", "B", "C")
+_VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
+# Timeframes the bot's candle-boundary scheduler supports (must have
+# AUTOSCAN_CONFIRMATIONS entries in telegram_bot.py).
+_SCANNABLE_TFS = ("M5", "M15", "H1", "H4")
+
+_KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level"}
+_KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"}
+_KNOWN_SCANNING = {"pairs", "timeframes"}
+_KNOWN_PAIR_KEYS = {"blocked_hours_utc"}
+
+
+class AppConfigError(Exception):
+    """config.json is present but invalid — fail fast at boot."""
+
+
+@dataclass
+class AppConfig:
+    telegram_token: str = ""
+    telegram_chat_id: str = ""
+    telegram_channel_id: str = ""          # "" = channel feed disabled
+    alert_min_quality: str = "C"
+    active_provider: str = "twelve_data"
+    provider_settings: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+    scan_pairs: List[str] = field(default_factory=lambda: [XAUUSD])
+    scan_timeframes: List[Timeframe] = field(default_factory=lambda: [
+        Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4])
+    pair_overrides: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+    log_level: str = "INFO"
+
+
+def _check_keys(section: Dict[str, Any], known: set, where: str) -> None:
+    if not isinstance(section, dict):
+        raise AppConfigError(f"config.json: {where} must be a JSON object")
+    unknown = sorted(set(section) - known)
+    if unknown:
+        raise AppConfigError(
+            f"config.json: unknown key(s) {unknown} in {where}; "
+            f"known keys: {sorted(known)}")
+
+
+def load_app_config(path: str = CONFIG_PATH) -> AppConfig:
+    """Parse + validate config.json. Missing file → pure defaults."""
+    app = AppConfig()
+    if not os.path.exists(path):
+        logger.warning(
+            "config.json not found at %s — running on code defaults "
+            "(secrets must come from environment variables)", path)
+        return app
+    try:
+        with open(path, encoding="utf-8") as fh:
+            raw = json.load(fh)
+    except json.JSONDecodeError as e:
+        raise AppConfigError(f"config.json is not valid JSON: {e}") from e
+    if not isinstance(raw, dict):
+        raise AppConfigError("config.json top level must be a JSON object")
+    _check_keys(raw, _KNOWN_TOP, "top level")
+
+    tg = raw.get("telegram", {})
+    _check_keys(tg, _KNOWN_TELEGRAM, '"telegram"')
+    app.telegram_token = str(tg.get("token", app.telegram_token))
+    app.telegram_chat_id = str(tg.get("chat_id", app.telegram_chat_id))
+    app.telegram_channel_id = str(
+        tg.get("channel_id", app.telegram_channel_id))
+    quality = tg.get("alert_min_quality", app.alert_min_quality)
+    if quality not in _VALID_QUALITIES:
+        raise AppConfigError(
+            f'config.json: telegram.alert_min_quality "{quality}" '
+            f"not in {_VALID_QUALITIES}")
+    app.alert_min_quality = quality
+
+    dp = raw.get("data_providers")
+    if dp is not None:
+        if not isinstance(dp, dict):
+            raise AppConfigError('config.json: "data_providers" must be an object')
+        app.active_provider = str(dp.get("active", app.active_provider))
+        app.provider_settings = {}
+        for name, block in dp.items():
+            if name == "active":
+                continue
+            if not isinstance(block, dict):
+                raise AppConfigError(
+                    f'config.json: data_providers.{name} must be an object')
+            app.provider_settings[name] = dict(block)
+        if app.active_provider not in app.provider_settings:
+            raise AppConfigError(
+                f'config.json: data_providers.active "{app.active_provider}" '
+                "has no matching provider block")
+
+    scanning = raw.get("scanning", {})
+    _check_keys(scanning, _KNOWN_SCANNING, '"scanning"')
+    if "pairs" in scanning:
+        pairs = scanning["pairs"]
+        if not isinstance(pairs, list):
+            raise AppConfigError('config.json: scanning.pairs must be a list')
+        bad = [p for p in pairs if p not in PAIR_INFO]
+        if bad:
+            raise AppConfigError(
+                f"config.json: unknown pair(s) {bad} in scanning.pairs; "
+                f"known: {sorted(PAIR_INFO)}")
+        app.scan_pairs = list(pairs)
+    if "timeframes" in scanning:
+        names = scanning["timeframes"]
+        if not isinstance(names, list):
+            raise AppConfigError(
+                'config.json: scanning.timeframes must be a list')
+        bad = [n for n in names if n not in _SCANNABLE_TFS]
+        if bad:
+            raise AppConfigError(
+                f"config.json: unsupported timeframe(s) {bad} in "
+                f"scanning.timeframes; supported: {list(_SCANNABLE_TFS)}")
+        app.scan_timeframes = [Timeframe[n] for n in names]
+
+    pair_section = raw.get("pairs", {})
+    if not isinstance(pair_section, dict):
+        raise AppConfigError('config.json: "pairs" must be a JSON object')
+    for symbol, overrides in pair_section.items():
+        if symbol not in PAIR_INFO:
+            raise AppConfigError(
+                f'config.json: unknown pair "{symbol}" in pairs; '
+                f"known: {sorted(PAIR_INFO)}")
+        _check_keys(overrides, _KNOWN_PAIR_KEYS, f'pairs."{symbol}"')
+        if "blocked_hours_utc" in overrides:
+            hours = overrides["blocked_hours_utc"]
+            valid = isinstance(hours, list) and all(
+                isinstance(h, int) and not isinstance(h, bool)
+                and 0 <= h <= 23 for h in hours)
+            if not valid:
+                raise AppConfigError(
+                    f'config.json: pairs."{symbol}".blocked_hours_utc must '
+                    "be a list of integers in 0-23")
+    app.pair_overrides = {s: dict(o) for s, o in pair_section.items()}
+
+    level = raw.get("log_level", app.log_level)
+    if level not in _VALID_LOG_LEVELS:
+        raise AppConfigError(
+            f'config.json: log_level "{level}" not in {_VALID_LOG_LEVELS}')
+    app.log_level = level
+    return app
+
+
+# ── Process-wide singleton ────────────────────────────────────────────
+
+_cached: Optional[AppConfig] = None
+
+
+def get_app_config() -> AppConfig:
+    global _cached
+    if _cached is None:
+        _cached = load_app_config()
+    return _cached
+
+
+def set_app_config(app: Optional[AppConfig]) -> None:
+    """Override the process-wide AppConfig (tests/experiments).
+    Pass None to re-enable lazy loading from CONFIG_PATH."""
+    global _cached
+    _cached = app
+
+
+# ── Resolution helpers: env var → config.json → default ───────────────
+
+def telegram_token(app: Optional[AppConfig] = None) -> str:
+    app = app or get_app_config()
+    return os.environ.get("TELEGRAM_TOKEN") or app.telegram_token
+
+
+def telegram_chat_id(app: Optional[AppConfig] = None) -> str:
+    app = app or get_app_config()
+    return os.environ.get("TELEGRAM_CHAT_ID") or app.telegram_chat_id
+
+
+def telegram_channel_id(app: Optional[AppConfig] = None) -> str:
+    """Env set-but-empty ("") means "disable the feed" and beats JSON,
+    so this checks None-ness, not truthiness."""
+    app = app or get_app_config()
+    env = os.environ.get("TELEGRAM_CHANNEL_ID")
+    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)."""
+    app = app or get_app_config()
+    if app.active_provider == "twelve_data":
+        env = os.environ.get("TWELVE_DATA_API_KEY")
+        if env:
+            return env
+    block = app.provider_settings.get(app.active_provider, {})
+    return str(block.get("api_key", ""))
+
+
+# ── PlatformConfig overlay ────────────────────────────────────────────
+
+def apply_pair_overrides(
+    cfg: PlatformConfig, app: Optional[AppConfig] = None
+) -> PlatformConfig:
+    """Stamp deployment settings from config.json onto a PlatformConfig.
+
+    Called by the pair factories BEFORE their validated per-timeframe
+    deltas, so e.g. the M5 hour-9 append lands on top of the JSON base.
+    The pair's base blocked_hours_utc is set on BOTH mirrored copies
+    (xau_strategy + backtest) so they cannot drift.
+    """
+    app = app or get_app_config()
+    overrides = app.pair_overrides.get(cfg.symbol, {})
+    if "blocked_hours_utc" in overrides:
+        hours = list(overrides["blocked_hours_utc"])
+        cfg.xau_strategy.blocked_hours_utc = hours
+        cfg.backtest.blocked_hours_utc = list(hours)
+    cfg.scan_pairs = list(app.scan_pairs)
+    cfg.scan_timeframes = list(app.scan_timeframes)
+    return cfg

backtest/gpu_backtest.py

@@ -55,7 +55,7 @@ from config import (  # noqa: E402
     Timeframe,
     XAUUSD,
 )
-from data_provider import ForexDataProvider  # noqa: E402
+from data_provider import ForexDataProvider, make_data_provider  # noqa: E402
 from engine import IndicatorSignal, TechnicalAnalysis, TechnicalEngine  # noqa: E402
 from signal_aggregator import SignalAggregator, SignalReport, TradeSetup  # noqa: E402
 
@@ -1679,7 +1679,7 @@ def run_backtest(
             f"vs PlatformConfig={platform_cfg.primary_timeframe.value}"
         )
 
-    provider = ForexDataProvider(platform_cfg)
+    provider = make_data_provider(platform_cfg)
 
     # ── Load all required timeframes ──
     timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs)

config.py

@@ -343,8 +343,6 @@ class PlatformConfig:
 
     data_lookback_days: int = 60
 
-    twelve_data_api_key: str = "f563b22ffdfd4924955decbe1a8c8701"
-
     refresh_interval: int = 300
 
     scan_pairs: List[str] = field(

data_provider.py

@@ -5,7 +5,6 @@ Fetches and validates OHLCV via the official `twelvedata` Python SDK.
 """
 
 import logging
-import os
 import threading
 import time
 from collections import deque
@@ -14,6 +13,7 @@ from typing import Dict, Optional
 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__)
@@ -70,10 +70,7 @@ class ForexDataProvider:
 
     def __init__(self, config: PlatformConfig):
         self.config = config
-        self._api_key = (
-            os.environ.get("TWELVE_DATA_API_KEY")
-            or config.twelve_data_api_key
-        )
+        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
@@ -264,8 +261,8 @@ class ForexDataProvider:
     ) -> pd.DataFrame:
         if self._td is None:
             raise DataIntegrityError(
-                "No Twelve Data API key. Set TWELVE_DATA_API_KEY env var "
-                "or config.twelve_data_api_key."
+                "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)
@@ -384,3 +381,20 @@ class ForexDataProvider:
             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)

docs/superpowers/plans/2026-07-21-centralized-config.md

@@ -644,7 +644,7 @@ Expected: `✗ config.py: hardcoded twelve_data_api_key field is gone` (plus the
 
 - [ ] **Step 3: Implement**
 
-`config.py`: delete line 346 (`twelve_data_api_key: str = "f563b22ffdfd4924955decbe1a8c8701"`).
+`config.py`: delete line 346 (`twelve_data_api_key: str = "<TWELVE_DATA_API_KEY>"`).
 
 `data_provider.py` — add to the local imports:
 
@@ -910,14 +910,14 @@ These are the exact values being deleted from source (from pre-change `telegram_
 cat > /home/harwinder/Development/ForexBot2/config.json <<'EOF'
 {
   "telegram": {
-    "token": "8228251480:AAFW2OAyQ2idMicql2t_CIXk1r9-qfRTfNE",
-    "chat_id": "6025899022",
-    "channel_id": "-1003942265371",
+    "token": "<TELEGRAM_BOT_TOKEN>",
+    "chat_id": "<OWNER_CHAT_ID>",
+    "channel_id": "<CHANNEL_ID>",
     "alert_min_quality": "C"
   },
   "data_providers": {
     "active": "twelve_data",
-    "twelve_data": { "api_key": "f563b22ffdfd4924955decbe1a8c8701" }
+    "twelve_data": { "api_key": "<TWELVE_DATA_API_KEY>" }
   },
   "scanning": {
     "pairs": ["XAU/USD"],
@@ -932,6 +932,8 @@ EOF
 git status --short   # config.json must NOT appear (gitignored)
 ```
 
+(Real values were scrubbed from this doc 2026-07-21 — take them from the previously deployed telegram_bot.py/config.py in git history, and rotate both credentials after merge.)
+
 - [ ] **Step 2: Boot smoke test (no network needed)**
 
 ```bash

pair_signals/xauusd.py

@@ -69,6 +69,7 @@ M5-only, same per-timeframe pattern as the M15 overrides.
 
 from typing import Optional, List
 from config import PlatformConfig, Timeframe
+from app_config import apply_pair_overrides
 
 _CONFIRMATIONS = {
     Timeframe.M5:  [Timeframe.M15, Timeframe.H1],
@@ -83,6 +84,9 @@ def make_config(
     confirmation_tfs: Optional[List[Timeframe]] = None,
 ) -> PlatformConfig:
     config = PlatformConfig()
+    # Deployment overlay (config.json) BEFORE the validated per-TF
+    # deltas below, so e.g. the M5 hour-9 append lands on the JSON base.
+    apply_pair_overrides(config)
     config.primary_timeframe = primary_tf
     config.confirmation_timeframes = (
         confirmation_tfs or _CONFIRMATIONS.get(primary_tf, [Timeframe.M15, Timeframe.H4])

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
+from data_provider import ForexDataProvider, make_data_provider
 from engine import TechnicalEngine, TechnicalAnalysis
 from news_engine import NewsEngine, FundamentalContext
 
@@ -104,7 +104,7 @@ class SignalAggregator:
 
     def __init__(self, config: PlatformConfig):
         self.config = config
-        self.data_provider = ForexDataProvider(config)
+        self.data_provider = make_data_provider(config)
         self.tech_engine = TechnicalEngine(config.technical)
         self._news_engine: Optional[NewsEngine] = None  # lazy
 

telegram_bot.py

@@ -65,10 +65,16 @@ from config import (
     PlatformConfig, SignalStrength, Timeframe,
     XAUUSD, PAIR_INFO,
 )
-from data_provider import ForexDataProvider
+from app_config import (
+    get_app_config, provider_api_key, telegram_token, telegram_chat_id,
+    telegram_channel_id,
+)
+from data_provider import ForexDataProvider, make_data_provider
 from pair_signals import PAIR_PRODUCERS
 from signal_aggregator import SignalAggregator, SignalReport
 
+_APP = get_app_config()
+
 # ── Force UTF-8 on Windows ────────────────────────────────────────────
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
@@ -85,13 +91,14 @@ _file_handler = logging.handlers.TimedRotatingFileHandler(
     when="midnight", backupCount=30, encoding="utf-8",
 )
 _file_handler.setFormatter(_log_formatter)
-logging.basicConfig(level=logging.INFO, handlers=[_console_handler, _file_handler])
+logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
+                    handlers=[_console_handler, _file_handler])
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
 logger = logging.getLogger("telegram_bot")
 
-TOKEN = os.environ.get("TELEGRAM_TOKEN", "8228251480:AAFW2OAyQ2idMicql2t_CIXk1r9-qfRTfNE")
-CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022")
+TOKEN = telegram_token()
+CHAT_ID = telegram_chat_id()
 
 # Owner lock: only this user/chat may drive the bot (commands + buttons).
 # The bot is the product's signal source — without the lock, anyone who
@@ -102,15 +109,15 @@ OWNER_ID: Optional[int] = int(CHAT_ID) if CHAT_ID else None
 # Telegram). When set, auto-scan signal alerts and level-breach alerts
 # are also posted there. Manual scans, status, cache and boot messages
 # stay owner-DM only. Set TELEGRAM_CHANNEL_ID="" to disable the feed.
-_CHANNEL_ID_RAW = os.environ.get("TELEGRAM_CHANNEL_ID", "-1003942265371")
+_CHANNEL_ID_RAW = telegram_channel_id()
 CHANNEL_ID: Optional[int] = int(_CHANNEL_ID_RAW) if _CHANNEL_ID_RAW else None
 
 # ── Pairs & buttons ───────────────────────────────────────────────────
-# The scanned pair list is config-driven: PlatformConfig.scan_pairs is
-# the single switch for single- vs multi-pair operation. Buttons, WS
-# subscriptions, and scan loops all derive from it.
+# The scanned pair list is config.json-driven (`scanning.pairs`, read via
+# app_config) and is the single switch for single- vs multi-pair operation.
+# Buttons, WS subscriptions, and scan loops all derive from it.
 
-SCAN_PAIRS = list(PlatformConfig().scan_pairs)
+SCAN_PAIRS = list(_APP.scan_pairs)
 
 _PAIRS_LABEL = " + ".join(PAIR_INFO[p]["display"] for p in SCAN_PAIRS)
 
@@ -192,7 +199,7 @@ _shared_dp: Optional[ForexDataProvider] = None
 def _get_shared_dp() -> ForexDataProvider:
     global _shared_dp
     if _shared_dp is None:
-        _shared_dp = ForexDataProvider(PlatformConfig())
+        _shared_dp = make_data_provider(PlatformConfig())
     return _shared_dp
 
 
@@ -475,7 +482,7 @@ async def run_signal(
 
 _QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1}
 _ALERT_QUALITY_CYCLE = ["C", "B", "A"]   # C+ → B+ → A-only
-_alert_min_quality = "C"
+_alert_min_quality = _APP.alert_min_quality
 
 # Fingerprint of the last alert per (symbol, timeframe) so a persisting
 # signal doesn't re-send an identical message every candle.
@@ -609,7 +616,7 @@ _last_price_ts: Dict[str, float] = {}
 _scan_cooldowns: Dict[str, float] = {}
 
 _cfg = PlatformConfig()
-_WS_API_KEY = os.environ.get("TWELVE_DATA_API_KEY") or _cfg.twelve_data_api_key
+_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
 
@@ -794,11 +801,12 @@ async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
         logger.info(f"{label} auto-scan loop stopped")
 
 
+_TF_INTERVAL_MINUTES = {
+    Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30,
+    Timeframe.H1: 60, Timeframe.H4: 240, Timeframe.D1: 1440,
+}
 _AUTOSCAN_LOOPS = [
-    (Timeframe.M5,  5),
-    (Timeframe.M15, 15),
-    (Timeframe.H1,  60),
-    (Timeframe.H4,  240),
+    (tf, _TF_INTERVAL_MINUTES[tf]) for tf in _APP.scan_timeframes
 ]
 
 
@@ -814,7 +822,8 @@ def _start_scheduler(bot) -> None:
             _autoscan_tasks.append(
                 asyncio.create_task(_candle_loop(bot, tf, interval))
             )
-        logger.info("Candle-boundary scheduler started (M5/M15/H1/H4)")
+        logger.info("Candle-boundary scheduler started (%s)",
+                    "/".join(tf.name for tf, _ in _AUTOSCAN_LOOPS))
 
     if _ws_task is None or _ws_task.done():
         _ws_task = asyncio.create_task(_ws_monitor(bot))
@@ -1259,8 +1268,11 @@ async def _post_shutdown(app: Application) -> None:
 # ── Entry point ───────────────────────────────────────────────────────
 
 def main() -> None:
-    if not TOKEN:
-        sys.exit("TELEGRAM_TOKEN not set")
+    if not TOKEN or not CHAT_ID:
+        raise SystemExit(
+            "telegram_bot: missing Telegram credentials. Set telegram.token "
+            "and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and "
+            "TELEGRAM_CHAT_ID environment variables.")
 
     logger.info("Starting ForexBot2 Telegram bot…")
 

test_all.py

@@ -18,6 +18,20 @@ import pandas as pd
 
 sys.path.insert(0, os.path.dirname(__file__))
 
+# ── Hermetic app-config: the suite must not depend on the developer's
+#    local config.json or TELEGRAM_*/TWELVE_DATA_API_KEY env vars. The
+#    dummy TD key mirrors the pre-config.json behavior where a key was
+#    always present (TDClient built, no network at construction).
+import json
+import app_config
+for _k in ("TELEGRAM_TOKEN", "TELEGRAM_CHAT_ID", "TELEGRAM_CHANNEL_ID",
+           "TWELVE_DATA_API_KEY"):
+    os.environ.pop(_k, None)
+_TEST_APP = app_config.AppConfig(
+    provider_settings={"twelve_data": {"api_key": "test-key"}}
+)
+app_config.set_app_config(_TEST_APP)
+
 PASS = 0
 FAIL = 0
 ERRORS = []
@@ -1865,6 +1879,7 @@ def _():
             primary_timeframe=tf,
             confirmation_timeframes=ctfs,
         )
+        app_config.apply_pair_overrides(inline_cfg)
         if tf == Timeframe.M15:
             # Documented per-TF overrides (see xauusd.py docstring):
             # 1R minimum targets + mixed-"trending"-regime block +
@@ -2525,6 +2540,282 @@ def _():
     assert args2.cp is False
 
 
+# ══════════════════════════════════════════════════════════════════════
+#  MODULE 13: app_config.py (config.json overlay)
+# ══════════════════════════════════════════════════════════════════════
+print("\n═══ Module 13: app_config.py ═══")
+
+import tempfile
+from app_config import (
+    AppConfig, AppConfigError, load_app_config, set_app_config,
+    telegram_token, telegram_chat_id, telegram_channel_id,
+    provider_api_key, apply_pair_overrides,
+)
+
+
+def _write_json(payload) -> str:
+    fh = tempfile.NamedTemporaryFile(
+        "w", suffix=".json", delete=False, encoding="utf-8")
+    if isinstance(payload, str):
+        fh.write(payload)
+    else:
+        json.dump(payload, fh)
+    fh.close()
+    return fh.name
+
+
+_FULL_CFG = {
+    "telegram": {"token": "t0k", "chat_id": "111", "channel_id": "-100222",
+                 "alert_min_quality": "B"},
+    "data_providers": {"active": "twelve_data",
+                       "twelve_data": {"api_key": "json-key"}},
+    "scanning": {"pairs": ["XAU/USD"], "timeframes": ["M15", "H1"]},
+    "pairs": {"XAU/USD": {"blocked_hours_utc": [3, 4]}},
+    "log_level": "WARNING",
+}
+
+
+@test("app_config: valid config.json maps every field")
+def _():
+    app = load_app_config(_write_json(_FULL_CFG))
+    assert app.telegram_token == "t0k"
+    assert app.telegram_chat_id == "111"
+    assert app.telegram_channel_id == "-100222"
+    assert app.alert_min_quality == "B"
+    assert app.active_provider == "twelve_data"
+    assert app.provider_settings["twelve_data"]["api_key"] == "json-key"
+    assert app.scan_pairs == [XAUUSD]
+    assert app.scan_timeframes == [Timeframe.M15, Timeframe.H1]
+    assert app.pair_overrides[XAUUSD]["blocked_hours_utc"] == [3, 4]
+    assert app.log_level == "WARNING"
+
+
+@test("app_config: missing file returns pure defaults")
+def _():
+    app = load_app_config("/nonexistent/config.json")
+    assert app == AppConfig(), "absent config.json must yield pure defaults"
+
+
+@test("app_config: malformed JSON raises AppConfigError")
+def _():
+    path = _write_json("{not json")
+    try:
+        load_app_config(path)
+        assert False, "should have raised"
+    except AppConfigError as e:
+        assert "not valid JSON" in str(e)
+
+
+@test("app_config: unknown keys raise AppConfigError")
+def _():
+    for payload in (
+        {"telgram": {}},                                   # top-level typo
+        {"telegram": {"tokn": "x"}},                       # section typo
+        {"pairs": {"XAU/USD": {"blocked_hours": [1]}}},    # pair-block typo
+        {"scanning": {"pair": ["XAU/USD"]}},               # scanning typo
+    ):
+        try:
+            load_app_config(_write_json(payload))
+            assert False, f"should have raised for {payload}"
+        except AppConfigError as e:
+            assert "unknown key" in str(e).lower()
+
+
+@test("app_config: invalid values raise AppConfigError")
+def _():
+    bad_payloads = [
+        {"pairs": {"XAU/USD": {"blocked_hours_utc": [24]}}},      # hour range
+        {"pairs": {"XAU/USD": {"blocked_hours_utc": "3"}}},       # not a list
+        {"pairs": {"EUR/USD": {"blocked_hours_utc": [3]}}},       # unknown pair
+        {"scanning": {"pairs": ["EUR/USD"]}},                     # unknown pair
+        {"scanning": {"timeframes": ["M7"]}},                     # bad TF name
+        {"scanning": {"timeframes": ["W1"]}},                     # unscannable TF
+        {"scanning": {"pairs": 5}},                               # pairs not a list
+        {"scanning": {"timeframes": 5}},                          # timeframes not a list
+        {"data_providers": {"active": "polygon",
+                            "twelve_data": {"api_key": "k"}}},    # no block
+        {"telegram": {"alert_min_quality": "Z"}},                 # bad quality
+        {"log_level": "CHATTY"},                                  # bad level
+    ]
+    for payload in bad_payloads:
+        try:
+            load_app_config(_write_json(payload))
+            assert False, f"should have raised for {payload}"
+        except AppConfigError:
+            pass
+
+
+@test("app_config: env precedence env → config.json → default")
+def _():
+    app = load_app_config(_write_json(_FULL_CFG))
+    with patch.dict(os.environ, {}, clear=True):
+        assert telegram_token(app) == "t0k"
+        assert telegram_chat_id(app) == "111"
+        assert telegram_channel_id(app) == "-100222"
+        assert provider_api_key(app) == "json-key"
+    with patch.dict(os.environ, {"TELEGRAM_TOKEN": "env-tok",
+                                 "TELEGRAM_CHAT_ID": "999",
+                                 "TWELVE_DATA_API_KEY": "env-key"}):
+        assert telegram_token(app) == "env-tok"
+        assert telegram_chat_id(app) == "999"
+        assert provider_api_key(app) == "env-key"
+    # channel: env set-but-empty means "disable feed" and must beat JSON
+    with patch.dict(os.environ, {"TELEGRAM_CHANNEL_ID": ""}):
+        assert telegram_channel_id(app) == ""
+
+
+@test("app_config: non-object sections raise AppConfigError")
+def _():
+    for payload in (
+        {"telegram": None},
+        {"telegram": "oops"},
+        {"scanning": 5},
+        {"pairs": ["oops"]},
+        {"pairs": {"XAU/USD": ["blocked_hours_utc"]}},
+    ):
+        try:
+            load_app_config(_write_json(payload))
+            assert False, f"should have raised for {payload}"
+        except AppConfigError:
+            pass
+
+
+@test("app_config: apply_pair_overrides stamps both blocked-hours mirrors")
+def _():
+    app = AppConfig(pair_overrides={XAUUSD: {"blocked_hours_utc": [3, 4]}},
+                    scan_pairs=[XAUUSD],
+                    scan_timeframes=[Timeframe.M15])
+    cfg = apply_pair_overrides(PlatformConfig(), app)
+    assert cfg.xau_strategy.blocked_hours_utc == [3, 4]
+    assert cfg.backtest.blocked_hours_utc == [3, 4]
+    assert cfg.scan_timeframes == [Timeframe.M15]
+    # the two mirrors must be independent lists, not aliases
+    cfg.xau_strategy.blocked_hours_utc.append(5)
+    assert cfg.backtest.blocked_hours_utc == [3, 4]
+
+
+@test("app_config: M5 factory appends hour 9 on top of a JSON base")
+def _():
+    from pair_signals import PAIR_PRODUCERS
+    app = AppConfig(pair_overrides={XAUUSD: {"blocked_hours_utc": [3, 4]}},
+                    provider_settings={"twelve_data": {"api_key": "test-key"}})
+    set_app_config(app)
+    try:
+        cfg = PAIR_PRODUCERS[XAUUSD](Timeframe.M5, None)
+    finally:
+        set_app_config(_TEST_APP)
+    assert cfg.xau_strategy.blocked_hours_utc == [3, 4, 9], \
+        "validated M5 hour-9 delta must land on top of the JSON base"
+    assert cfg.backtest.blocked_hours_utc == [3, 4], \
+        "mirror carries the base only (per-TF delta is XAU-strategy-side)"
+
+
+@test("app_config: config.json.example is valid and behavior-neutral")
+def _():
+    import dataclasses
+    from pair_signals import PAIR_PRODUCERS
+    example = load_app_config(
+        os.path.join(os.path.dirname(__file__), "config.json.example"))
+    for tf in (Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4):
+        baseline = PAIR_PRODUCERS[XAUUSD](tf, None)   # under _TEST_APP
+        set_app_config(example)
+        try:
+            with_example = PAIR_PRODUCERS[XAUUSD](tf, None)
+        finally:
+            set_app_config(_TEST_APP)
+        assert dataclasses.asdict(baseline) == dataclasses.asdict(with_example), \
+            f"example config.json must be byte-identical to defaults on {tf}"
+
+
+@test("data_provider: api key resolves env → config.json (no code default)")
+def _():
+    from data_provider import ForexDataProvider
+    app = AppConfig(provider_settings={"twelve_data": {"api_key": "json-key"}})
+    set_app_config(app)
+    try:
+        with patch.dict(os.environ, {}, clear=True):
+            dp = ForexDataProvider(PlatformConfig())
+            assert dp._api_key == "json-key"
+        with patch.dict(os.environ, {"TWELVE_DATA_API_KEY": "env-key"}):
+            dp = ForexDataProvider(PlatformConfig())
+            assert dp._api_key == "env-key"
+        set_app_config(AppConfig())        # no key anywhere
+        with patch.dict(os.environ, {}, clear=True):
+            dp = ForexDataProvider(PlatformConfig())
+            assert dp._api_key is None and dp._td is None
+    finally:
+        set_app_config(_TEST_APP)
+
+
+@test("data_provider: make_data_provider dispatches on active provider")
+def _():
+    from data_provider import ForexDataProvider, make_data_provider
+    assert isinstance(make_data_provider(PlatformConfig()), ForexDataProvider)
+    set_app_config(AppConfig(active_provider="polygon",
+                             provider_settings={"polygon": {"api_key": "k"}}))
+    try:
+        make_data_provider(PlatformConfig())
+        assert False, "unregistered provider must raise"
+    except AppConfigError as e:
+        assert "polygon" in str(e)
+    finally:
+        set_app_config(_TEST_APP)
+
+
+@test("config.py: hardcoded twelve_data_api_key field is gone")
+def _():
+    assert not hasattr(PlatformConfig(), "twelve_data_api_key"), \
+        "credentials must not live in committed source"
+
+
+@test("telegram_bot: credentials, quality, pairs, loops from app_config")
+def _():
+    import importlib
+    import telegram_bot
+    app = AppConfig(telegram_token="json-tok", telegram_chat_id="42",
+                    telegram_channel_id="", alert_min_quality="B",
+                    scan_timeframes=[Timeframe.M15, Timeframe.H1],
+                    provider_settings={"twelve_data": {"api_key": "k"}})
+    set_app_config(app)
+    try:
+        with patch.dict(os.environ, {}, clear=True):
+            tb = importlib.reload(telegram_bot)
+            assert tb.TOKEN == "json-tok"
+            assert tb.OWNER_ID == 42
+            assert tb.CHANNEL_ID is None, '"" channel id must disable the feed'
+            assert tb._alert_min_quality == "B"
+            assert [tf for tf, _ in tb._AUTOSCAN_LOOPS] == \
+                [Timeframe.M15, Timeframe.H1]
+            assert tb._AUTOSCAN_LOOPS == [(Timeframe.M15, 15),
+                                          (Timeframe.H1, 60)]
+    finally:
+        set_app_config(_TEST_APP)
+        importlib.reload(telegram_bot)
+
+
+@test("telegram_bot: main() refuses to start without credentials")
+def _():
+    import telegram_bot
+    if telegram_bot.TOKEN:      # only meaningful in the hermetic no-cred env
+        return
+    try:
+        telegram_bot.main()
+        assert False, "main() must refuse to start without token/chat_id"
+    except SystemExit as e:
+        assert "credentials" in str(e)
+
+
+@test("app_config: absolute parity — default overlay leaves tuned values untouched")
+def _():
+    from pair_signals import PAIR_PRODUCERS
+    cfg = PAIR_PRODUCERS[XAUUSD](Timeframe.H1, None)
+    assert cfg.xau_strategy.blocked_hours_utc == [0, 1, 2, 10, 11, 13, 16, 21]
+    assert cfg.backtest.blocked_hours_utc == [0, 1, 2, 10, 11, 13, 16, 21]
+    assert cfg.scan_pairs == [XAUUSD]
+    assert cfg.scan_timeframes == [Timeframe.M5, Timeframe.M15,
+                                   Timeframe.H1, Timeframe.H4]
+
+
 # ══════════════════════════════════════════════════════════════════════
 #  SUMMARY
 # ══════════════════════════════════════════════════════════════════════

test_ws_dual.py

@@ -10,7 +10,9 @@ import time
 
 from twelvedata import TDClient
 
-API_KEY = "f563b22ffdfd4924955decbe1a8c8701"
+from app_config import provider_api_key
+
+API_KEY = provider_api_key()
 SYMBOL = "XAU/USD"
 DURATION = 30