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

Centralized config.json Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Move deployment settings (Telegram credentials, data-provider API keys/selection, scanned pairs, scan timeframes, per-pair base blocked hours, alert quality boot default, log level) into a gitignored config.json loaded once at startup, and delete the hardcoded credentials from source.

Architecture: A new app_config.py module loads and validates config.json into an AppConfig dataclass (cached singleton), with resolver functions applying the precedence env var → config.json → code default. The pair factory calls apply_pair_overrides() on every fresh PlatformConfig before its validated per-timeframe deltas, so the bot, CLI, and backtester all pick up the overlay identically (live == backtest parity by construction). Spec: docs/superpowers/specs/2026-07-21-centralized-config-design.md.

Tech Stack: Python 3.12 stdlib only (json, os, dataclasses). No new dependencies.

Global Constraints

• Zero signal-logic change. No indicator weight, threshold, gate, or geometry value may change. With no config.json, or one holding the example defaults, every produced PlatformConfig must be byte-identical (dataclasses.asdict equality) to today's factory output — pinned by a test.
• The test suite is python test_all.py — a custom harness (@test("name") decorator, tests execute at import time, exit code non-zero on failure). It is NOT pytest; there is no single-test selection. "Run the test" always means python test_all.py and checking the named line in the output. Baseline before this work: 184 passed, 0 failed.
• The suite must be hermetic: it must pass identically whether or not the developer's machine has a config.json, and regardless of TELEGRAM_*/TWELVE_DATA_API_KEY env vars.
• Secrets get no code default. After Task 4 the only places credentials exist are config.json (gitignored) and env vars.
• config.json.example (already committed at repo root) is the schema reference. config.json is already in .gitignore — never commit it.
• Repo root: /home/harwinder/Development/ForexBot2. Use venv/bin/python if bare python lacks pandas/numpy.
• Every commit message ends with:

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

• Work on branch centralized-config (create from main at start of Task 1: git checkout -b centralized-config).

---

Task 1: app_config.py — loader, validation, resolvers

Files:

• Create: app_config.py
• Modify: test_all.py (new Module 13 before the SUMMARY block, ~line 2495; hermetic header after line 19)
• Test: test_all.py (same file — this repo keeps all tests in one suite)

Interfaces:

• Consumes: config.PAIR_INFO, config.Timeframe, config.XAUUSD, config.PlatformConfig (all existing).

• Produces (later tasks rely on these exact names):

  • class AppConfigError(Exception)
  • @dataclass class AppConfig — fields: telegram_token: str = "", telegram_chat_id: str = "", telegram_channel_id: str = "", alert_min_quality: str = "C", active_provider: str = "twelve_data", provider_settings: Dict[str, Dict[str, Any]] = {}, scan_pairs: List[str] = [XAUUSD], scan_timeframes: List[Timeframe] = [M5, M15, H1, H4], pair_overrides: Dict[str, Dict[str, Any]] = {}, log_level: str = "INFO"
  • load_app_config(path: str = CONFIG_PATH) -> AppConfig
  • get_app_config() -> AppConfig (cached), set_app_config(app: Optional[AppConfig]) -> None (test hook; None re-enables lazy load)
  • telegram_token(app=None) -> str, telegram_chat_id(app=None) -> str, telegram_channel_id(app=None) -> str, provider_api_key(app=None) -> str
  • apply_pair_overrides(cfg: PlatformConfig, app: Optional[AppConfig] = None) -> PlatformConfig (implemented here, wired in Task 2)

• Step 1: Create branch

cd /home/harwinder/Development/ForexBot2 && git checkout -b centralized-config

• Step 2: Add the hermetic test header and Module 13 failing tests to test_all.py

In test_all.py, directly after sys.path.insert(0, os.path.dirname(__file__)) (line 19), add:

# ── 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)

Then insert Module 13 immediately BEFORE the # SUMMARY comment block near the end of the file (search for SUMMARY):

# ══════════════════════════════════════════════════════════════════════
#  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
        {"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) == ""

• Step 3: Run to verify the new tests fail (import error)

python test_all.py 2>&1 | head -5

Expected: ModuleNotFoundError: No module named 'app_config' (the hermetic header imports it before any test runs).

• Step 4: Write app_config.py

Create app_config.py with exactly this content:

"""
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 can run (W1 has no loop).
_SCANNABLE_TFS = ("M5", "M15", "M30", "H1", "H4", "D1")

_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:
    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"]
        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"]
        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", {})
    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

• Step 5: Run tests, verify Module 13 passes and total is 190

python test_all.py 2>&1 | tail -25

Expected: RESULTS: 190 passed, 0 failed, 190 total (184 baseline + 6 new). All six app_config: lines show ✓.

• Step 6: Commit

git add app_config.py test_all.py
git commit -m "Add app_config.py: config.json loader, validation, env-precedence resolvers"

---

Task 2: Factory overlay wiring + parity pins

Files:

• Modify: pair_signals/xauusd.py:81-90 (make_config), test_all.py (Module 9 equivalence test ~line 1851; Module 13 additions)

Interfaces:

• Consumes: app_config.apply_pair_overrides, app_config.set_app_config, app_config.AppConfig, app_config.load_app_config (Task 1).

• Produces: make_config now applies the overlay — every consumer (bot, CLI, backtester) inherits it with no further change. Test symbol _TEST_APP (Task 1) is the restore target for any test that swaps the singleton.

• Step 1: Add failing tests (Module 13 additions)

Append to Module 13 in test_all.py:

@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}"

• Step 2: Run to verify the new tests fail

python test_all.py 2>&1 | grep "app_config: M5 factory"

Expected: ✗ app_config: M5 factory appends hour 9 on top of a JSON base (the factory doesn't apply the overlay yet, so blocked hours are the defaults [0,1,2,10,11,13,16,21,9], not [3,4,9]).

• Step 3: Wire the overlay into the factory

In pair_signals/xauusd.py, change the import line and the top of make_config:

from typing import Optional, List
from config import PlatformConfig, Timeframe
from app_config import apply_pair_overrides

and in make_config, immediately after config = 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

• Step 4: Update the Module 9 equivalence test

In test_all.py (~line 1851, "xauusd.make_config equals inline PlatformConfig construction"), the inline construction must apply the same overlay in the same order. After the inline_cfg = PlatformConfig(...) construction and BEFORE the per-TF override lines, add:

        app_config.apply_pair_overrides(inline_cfg)

(app_config is already imported at the top of test_all.py from Task 1. Under _TEST_APP this is a no-op on blocked hours, but it keeps the equivalence definition exact: factory ≡ inline + overlay + per-TF deltas, whatever the active AppConfig.)

• Step 5: Run tests, verify 193 passing

python test_all.py 2>&1 | tail -25

Expected: RESULTS: 193 passed, 0 failed, 193 total. Watch specifically for ✓ xauusd.make_config equals inline PlatformConfig construction (Module 9) and the three new Module 13 lines.

• Step 6: Commit

git add pair_signals/xauusd.py test_all.py
git commit -m "Wire config.json overlay into the pair factory (before per-TF deltas)"

---

Task 3: Data provider — key from app_config, registry, delete hardcoded key

Files:

• Modify: config.py:346 (delete field), data_provider.py:72-76 (key resolution), data_provider.py:265-269 (error message), data_provider.py end (registry), signal_aggregator.py:107, backtest/gpu_backtest.py:1682, test_all.py (Module 13 additions)

Interfaces:

• Consumes: app_config.provider_api_key, app_config.get_app_config, app_config.AppConfigError (Task 1).

• Produces: data_provider.PROVIDER_REGISTRY: Dict[str, type] and data_provider.make_data_provider(config: PlatformConfig) -> ForexDataProvider — Task 4 uses make_data_provider in telegram_bot.py.

• Step 1: Add failing tests (Module 13 additions)

@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"

• Step 2: Run to verify they fail

python test_all.py 2>&1 | grep -A1 "hardcoded twelve_data"

Expected: ✗ config.py: hardcoded twelve_data_api_key field is gone (plus the make_data_provider import failure).

• Step 3: Implement

config.py: delete line 346 (twelve_data_api_key: str = "f563b22ffdfd4924955decbe1a8c8701").

data_provider.py — add to the local imports:

from app_config import AppConfigError, get_app_config, provider_api_key

In ForexDataProvider.__init__ replace the key resolution:

        self.config = config
        self._api_key = provider_api_key() or None

In _fetch_twelve_data, update the no-key error:

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

At the bottom of data_provider.py (after the class):

# ── 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)

signal_aggregator.py line 18 + 107 — import and use the registry:

from data_provider import ForexDataProvider, make_data_provider

        self.data_provider = make_data_provider(config)

backtest/gpu_backtest.py line 1682 (and its import of ForexDataProvider):

    provider = make_data_provider(platform_cfg)

• Step 4: Run tests, verify 196 passing

python test_all.py 2>&1 | tail -25

Expected: RESULTS: 196 passed, 0 failed, 196 total. If any Module 2 test fails on a missing key, it means it constructed ForexDataProvider outside _TEST_APP scope — fix the test to run under _TEST_APP (which supplies the dummy key), never by re-adding a code default.

• Step 5: Commit

git add config.py data_provider.py signal_aggregator.py backtest/gpu_backtest.py test_all.py
git commit -m "Data provider: key from config.json/env, provider registry, drop hardcoded key"

---

Task 4: telegram_bot wiring + startup guard

Files:

• Modify: telegram_bot.py (credentials block ~93-105, logging ~88, SCAN_PAIRS ~113, alert quality ~478, WS key ~611-614, _AUTOSCAN_LOOPS ~797-802, scheduler log ~817, _get_shared_dp ~195, main() ~1261), test_all.py (Module 13 addition)

Interfaces:

• Consumes: app_config.get_app_config, resolvers, data_provider.make_data_provider (Tasks 1+3).

• Produces: nothing consumed later; this is the final wiring task.

• Step 1: Add failing test (Module 13 addition)

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

• Step 2: Run to verify failure

python test_all.py 2>&1 | grep "telegram_bot: credentials"

Expected: ✗ telegram_bot: credentials, quality, pairs, loops from app_config (TOKEN still comes from the hardcoded fallback, not "json-tok").

• Step 3: Implement the wiring

telegram_bot.py — add to the local imports (after the config import block):

from app_config import (
    get_app_config, telegram_token, telegram_chat_id, telegram_channel_id,
)
from data_provider import ForexDataProvider, make_data_provider

Immediately after the imports (before the logging block), add:

_APP = get_app_config()

Change the logging.basicConfig line (~88) to honor the configured level:

logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
                    handlers=[_console_handler, _file_handler])

Replace the credentials block (lines ~93-105) — keep the existing comments about the owner lock and subscriber channel, change only the assignments:

TOKEN = telegram_token()
CHAT_ID = telegram_chat_id()

# ... (owner-lock comment unchanged)
OWNER_ID: Optional[int] = int(CHAT_ID) if CHAT_ID else None

# ... (subscriber-channel comment unchanged)
_CHANNEL_ID_RAW = telegram_channel_id()
CHANNEL_ID: Optional[int] = int(_CHANNEL_ID_RAW) if _CHANNEL_ID_RAW else None

SCAN_PAIRS (~113):

SCAN_PAIRS = list(_APP.scan_pairs)

_get_shared_dp (~195):

        _shared_dp = make_data_provider(PlatformConfig())

Alert quality boot default (~478):

_alert_min_quality = _APP.alert_min_quality

WS key (~611-614) — provider_api_key needs adding to the app_config import list:

_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

_AUTOSCAN_LOOPS (~797-802) — derive from configured timeframes:

_TF_INTERVAL_MINUTES = {
    Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30,
    Timeframe.H1: 60, Timeframe.H4: 240, Timeframe.D1: 1440,
}
_AUTOSCAN_LOOPS = [
    (tf, _TF_INTERVAL_MINUTES[tf]) for tf in _APP.scan_timeframes
]

Scheduler log line (~817):

        logger.info("Candle-boundary scheduler started (%s)",
                    "/".join(tf.name for tf, _ in _AUTOSCAN_LOOPS))

(The manual /scan buttons — _TF_BUTTON_LABELS — deliberately stay hardcoded M5/M15/H1/H4: they are the owner's on-demand UI, not the automated feed. Documented decision, not an omission.)

main() (~1261) — first lines of the function body:

def main() -> None:
    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.")

• Step 4: Run tests, verify 198 passing

python test_all.py 2>&1 | tail -25

Expected: RESULTS: 198 passed, 0 failed, 198 total. If Module 7 tests break on OWNER_ID being None, they were relying on the hardcoded chat id — fix by having those tests set telegram_bot.OWNER_ID/CHAT_ID explicitly in their setup, never by re-adding fallbacks.

• Step 5: Commit

git add telegram_bot.py test_all.py
git commit -m "telegram_bot: credentials/quality/scan-loops from config.json, startup guard"

---

Task 5: Migration, docs, final verification

Files:

• Create: config.json (LOCAL ONLY — gitignored, never committed)
• Modify: CLAUDE.md (repo map, env-var table, config section, test count)

Interfaces:

• Consumes: everything above.

• Produces: the deployed machine's real config.json; updated authoritative docs.

• Step 1: Create the local config.json with the current live values

These are the exact values being deleted from source (from pre-change telegram_bot.py:93-105 and config.py:346), so the running bot behaves identically:

cat > /home/harwinder/Development/ForexBot2/config.json <<'EOF'
{
  "telegram": {
    "token": "8228251480:AAFW2OAyQ2idMicql2t_CIXk1r9-qfRTfNE",
    "chat_id": "6025899022",
    "channel_id": "-1003942265371",
    "alert_min_quality": "C"
  },
  "data_providers": {
    "active": "twelve_data",
    "twelve_data": { "api_key": "f563b22ffdfd4924955decbe1a8c8701" }
  },
  "scanning": {
    "pairs": ["XAU/USD"],
    "timeframes": ["M5", "M15", "H1", "H4"]
  },
  "pairs": {
    "XAU/USD": { "blocked_hours_utc": [0, 1, 2, 10, 11, 13, 16, 21] }
  },
  "log_level": "INFO"
}
EOF
git status --short   # config.json must NOT appear (gitignored)

• Step 2: Boot smoke test (no network needed)

python - <<'EOF'
from app_config import get_app_config, telegram_token, provider_api_key, apply_pair_overrides
from pair_signals import PAIR_PRODUCERS
from config import Timeframe, XAUUSD
app = get_app_config()
assert telegram_token(app), "token missing"
assert provider_api_key(app), "api key missing"
cfg = PAIR_PRODUCERS[XAUUSD](Timeframe.M5, None)
assert cfg.xau_strategy.blocked_hours_utc == [0, 1, 2, 10, 11, 13, 16, 21, 9]
print("boot smoke OK — live config.json loads and factory output unchanged")
EOF

Expected: boot smoke OK — live config.json loads and factory output unchanged.

• Step 3: Update CLAUDE.md

• Repository-map table: add rows for app_config.py ("config.json loader/validation, env-precedence resolvers, apply_pair_overrides deployment overlay") and config.json.example ("committed schema template; real gitignored config.json holds secrets + deployment settings").

• Environment-variables table: note precedence "env var → config.json → none"; state that hardcoded fallbacks were REMOVED 2026-07-21 and the bot refuses to boot without credentials.

• PlatformConfig section: delete the twelve_data_api_key mention ("hardcoded fallback"); add one line: "Deployment settings (credentials, scan pairs/TFs, per-pair base blocked hours, alert quality, log level) overlay from config.json via app_config.apply_pair_overrides, applied in the pair factory BEFORE per-TF deltas."

• pair_signals section: mention the overlay call order in make_config.

• Test-suite line: update "184 tests in 12 modules" → "198 tests in 13 modules" (verify against the actual final count from Step 4 before writing).

• Step 4: Full-suite final verification

python test_all.py 2>&1 | tail -15

Expected: RESULTS: 198 passed, 0 failed, 198 total (use the real number if it differs, and make CLAUDE.md match).

• Step 5: Commit docs

git add CLAUDE.md
git commit -m "Docs: config.json deployment overlay (CLAUDE.md repo map, env table, test count)"

• Step 6: Merge/PR decision

Implementation complete → follow superpowers:finishing-a-development-branch (options: merge centralized-config to main, or push the branch to origin/gitea for review). After merge, remind the user: rotate the Telegram token and Twelve Data key — the old values remain in git history on both remotes.