diff --git a/app_config.py b/app_config.py new file mode 100644 index 0000000..cd8ebf9 --- /dev/null +++ b/app_config.py @@ -0,0 +1,240 @@ +""" +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 diff --git a/test_all.py b/test_all.py index 2705def..2f169a8 100644 --- a/test_all.py +++ b/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 = [] @@ -2525,6 +2539,128 @@ 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 + {"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) == "" + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════