""" 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. A repo-root .env file (gitignored) is loaded at import via setdefault, so it behaves exactly like real environment variables except that a real shell export always wins. 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). """ 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") ENV_PATH = os.path.join( os.path.dirname(os.path.abspath(__file__)), ".env") def load_dotenv(path: str = ENV_PATH) -> None: """Minimal .env loader (no python-dotenv dependency). KEY=VALUE lines land in os.environ via setdefault — a variable already exported in the shell always wins, keeping the env → config.json precedence intact (.env is just a convenient place to put env vars). Comments, blank lines, ``export `` prefixes and single/double quotes are handled; anything else is skipped. """ try: with open(path, encoding="utf-8") as fh: lines = fh.read().splitlines() except OSError: return for line in lines: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue if line.startswith("export "): line = line[len("export "):] key, _, value = line.partition("=") key, value = key.strip(), value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'": value = value[1:-1] elif value[:1] not in "\"'": # Unquoted value: whitespace-then-# starts an inline comment # (KEY=abc # prod). Quote the value to keep a literal " #". for sep in (" #", "\t#"): cut = value.find(sep) if cut != -1: value = value[:cut].rstrip() if key: os.environ.setdefault(key, value) # Module import == process boot, before any resolver can run. load_dotenv() _VALID_QUALITIES = ("A+", "A", "B", "C") _VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") # Timeframes the bot's candle-boundary scheduler supports (must have # TF_CONFIRMATIONS entries in telegram_bot.py). _SCANNABLE_TFS = ("M5", "M15", "H1", "H4") _KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level", "oanda", "deepseek", "agent_panel", "market_outlook", "violent_watch", "signal_log"} _KNOWN_OANDA = { # "account_id" is TOLERATED but ignored: it belonged to the auto- # execution daemon removed 2026-09-03 (last commit 31002cb). The candle # provider needs only api_token + base_url; older config.json files # still carry the key and must keep loading. "account_id", "api_token", "base_url", } _KNOWN_DEEPSEEK = { "api_key", "model", "enabled", "timeout_seconds", "daily_request_cap", } _KNOWN_AGENT_PANEL = { "enabled", "telegram", "dashboard", "min_quality", "max_panels_per_day", "reuse_ttl_seconds", "candles_enabled", "candle_bars_primary", "candle_bars_confirm", "verdict_source", } _KNOWN_MARKET_OUTLOOK = {"enabled", "telegram", "max_per_day", "candles_bars"} _KNOWN_VIOLENT_WATCH = {"enabled", "alert_line", "headsup", "max_headsup_per_day", "headsup_min_level"} _KNOWN_SIGNAL_LOG = {"enabled", "path", "log_blocked_scans"} _KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"} _KNOWN_SCANNING = {"pairs", "timeframes"} _KNOWN_PAIR_KEYS = {"blocked_hours_utc"} _MARKET_OUTLOOK_DEFAULTS = { "enabled": False, "telegram": False, "max_per_day": 12, "candles_bars": 20, } # Violent-move watch (display-only heads-up) — dark by default. _VIOLENT_WATCH_DEFAULTS = { "enabled": False, "alert_line": True, "headsup": True, "max_headsup_per_day": 12, "headsup_min_level": "hot", } _VALID_WATCH_LEVELS = ("watch", "hot") 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" streaming_provider: Optional[str] = None # None = active provider 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) oanda_settings: Dict[str, Any] = field(default_factory=dict) deepseek_settings: Dict[str, Any] = field(default_factory=dict) agent_panel_settings: Dict[str, Any] = field(default_factory=dict) market_outlook_settings: Dict[str, Any] = field( default_factory=lambda: dict(_MARKET_OUTLOOK_DEFAULTS)) violent_watch_settings: Dict[str, Any] = field( default_factory=lambda: dict(_VIOLENT_WATCH_DEFAULTS)) signal_log_settings: 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(): # "active" and "streaming" are reserved scalars, not blocks. if name in ("active", "streaming"): 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") streaming = dp.get("streaming") if streaming is not None: if not isinstance(streaming, str): raise AppConfigError( 'config.json: data_providers.streaming must be a string ' "(a provider name)") if streaming not in app.provider_settings: raise AppConfigError( f'config.json: data_providers.streaming "{streaming}" ' "has no matching provider block") app.streaming_provider = streaming 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()} oanda = raw.get("oanda", {}) _check_keys(oanda, _KNOWN_OANDA, '"oanda"') app.oanda_settings = dict(oanda) deepseek = raw.get("deepseek", {}) _check_keys(deepseek, _KNOWN_DEEPSEEK, '"deepseek"') if "enabled" in deepseek and not isinstance(deepseek["enabled"], bool): raise AppConfigError( "config.json: deepseek.enabled must be true or false") if "timeout_seconds" in deepseek: ts = deepseek["timeout_seconds"] if isinstance(ts, bool) or not isinstance(ts, (int, float)) or ts <= 0: raise AppConfigError( "config.json: deepseek.timeout_seconds must be a positive " "number") if "daily_request_cap" in deepseek: cap = deepseek["daily_request_cap"] if isinstance(cap, bool) or not isinstance(cap, int) or cap < 1: raise AppConfigError( "config.json: deepseek.daily_request_cap must be a positive " "integer") if "model" in deepseek: if not isinstance(deepseek["model"], str) or not deepseek["model"]: raise AppConfigError( "config.json: deepseek.model must be a non-empty string") app.deepseek_settings = dict(deepseek) panel = raw.get("agent_panel", {}) _check_keys(panel, _KNOWN_AGENT_PANEL, '"agent_panel"') for flag in ("enabled", "telegram", "dashboard"): if flag in panel and not isinstance(panel[flag], bool): raise AppConfigError( f"config.json: agent_panel.{flag} must be true or false") if "min_quality" in panel and panel["min_quality"] not in _VALID_QUALITIES: raise AppConfigError( f'config.json: agent_panel.min_quality "{panel["min_quality"]}" ' f"not in {_VALID_QUALITIES}") if "max_panels_per_day" in panel: mpd = panel["max_panels_per_day"] if isinstance(mpd, bool) or not isinstance(mpd, int) or mpd < 1: raise AppConfigError( "config.json: agent_panel.max_panels_per_day must be a " "positive integer") if "reuse_ttl_seconds" in panel: ttl = panel["reuse_ttl_seconds"] if isinstance(ttl, bool) or not isinstance(ttl, (int, float)) \ or ttl < 0: raise AppConfigError( "config.json: agent_panel.reuse_ttl_seconds must be a " "non-negative number") if "candles_enabled" in panel \ and not isinstance(panel["candles_enabled"], bool): raise AppConfigError( "config.json: agent_panel.candles_enabled must be true or false") for key in ("candle_bars_primary", "candle_bars_confirm"): if key in panel: n = panel[key] if isinstance(n, bool) or not isinstance(n, int) \ or not 1 <= n <= 100: raise AppConfigError( f"config.json: agent_panel.{key} must be an integer " "between 1 and 100") if "verdict_source" in panel and panel["verdict_source"] not in \ ("macro", "chair"): raise AppConfigError( 'config.json: agent_panel.verdict_source must be "macro" or ' '"chair"') app.agent_panel_settings = dict(panel) outlook = raw.get("market_outlook", {}) _check_keys(outlook, _KNOWN_MARKET_OUTLOOK, '"market_outlook"') for flag in ("enabled", "telegram"): if flag in outlook and not isinstance(outlook[flag], bool): raise AppConfigError( f"config.json: market_outlook.{flag} must be true or false") for key in ("max_per_day", "candles_bars"): if key in outlook: value = outlook[key] if isinstance(value, bool) or not isinstance(value, int) or value < 1: raise AppConfigError( f"config.json: market_outlook.{key} must be a positive integer") app.market_outlook_settings.update(outlook) watch = raw.get("violent_watch", {}) _check_keys(watch, _KNOWN_VIOLENT_WATCH, '"violent_watch"') for flag in ("enabled", "alert_line", "headsup"): if flag in watch and not isinstance(watch[flag], bool): raise AppConfigError( f"config.json: violent_watch.{flag} must be true or false") if "max_headsup_per_day" in watch: cap = watch["max_headsup_per_day"] if isinstance(cap, bool) or not isinstance(cap, int) or cap < 0: raise AppConfigError( "config.json: violent_watch.max_headsup_per_day must be a " "non-negative integer") if ("headsup_min_level" in watch and watch["headsup_min_level"] not in _VALID_WATCH_LEVELS): raise AppConfigError( f'config.json: violent_watch.headsup_min_level ' f'"{watch["headsup_min_level"]}" not in {_VALID_WATCH_LEVELS}') app.violent_watch_settings.update(watch) slog = raw.get("signal_log", {}) _check_keys(slog, _KNOWN_SIGNAL_LOG, '"signal_log"') if "enabled" in slog and not isinstance(slog["enabled"], bool): raise AppConfigError( "config.json: signal_log.enabled must be true or false") if "path" in slog and (not isinstance(slog["path"], str) or not slog["path"]): raise AppConfigError( "config.json: signal_log.path must be a non-empty string") if ("log_blocked_scans" in slog and not isinstance(slog["log_blocked_scans"], bool)): raise AppConfigError( "config.json: signal_log.log_blocked_scans must be true or false") app.signal_log_settings = dict(slog) 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 # Per-provider env-var overrides for API keys (env → config.json → ""). _PROVIDER_ENV_KEYS = { "twelve_data": "TWELVE_DATA_API_KEY", "tiingo": "TIINGO_API_KEY", } def provider_api_key( app: Optional[AppConfig] = None, provider: Optional[str] = None ) -> str: """API key for `provider` (default: the ACTIVE provider). The env override is per-provider (TWELVE_DATA_API_KEY applies only to twelve_data, TIINGO_API_KEY only to tiingo).""" app = app or get_app_config() name = provider or app.active_provider env_key = _PROVIDER_ENV_KEYS.get(name) if env_key: env = os.environ.get(env_key) if env: return env block = app.provider_settings.get(name, {}) return str(block.get("api_key", "")) # ── Oanda resolution helpers (env → config.json → default) ─────────── def oanda_api_token(app: Optional[AppConfig] = None) -> str: app = app or get_app_config() return os.environ.get("OANDA_API_TOKEN") or str( app.oanda_settings.get("api_token", "")) def oanda_base_url(app: Optional[AppConfig] = None) -> str: app = app or get_app_config() return os.environ.get("OANDA_BASE_URL") or str( app.oanda_settings.get("base_url", "https://api-fxpractice.oanda.com")) # ── DeepSeek resolution helper (env → config.json → "") ────────────── def deepseek_api_key(app: Optional[AppConfig] = None) -> str: app = app or get_app_config() return os.environ.get("DEEPSEEK_API_KEY") or str( app.deepseek_settings.get("api_key", "")) # ── AI market-outlook resolution helper ────────────────────────────── def market_outlook_enabled(app: Optional[AppConfig] = None) -> bool: """Whether the advisory outlook feature is enabled (env wins).""" env = os.environ.get("MARKET_OUTLOOK_ENABLED") if env: return env.strip().lower() in ("1", "true", "yes", "on") app = app or get_app_config() return bool(app.market_outlook_settings.get("enabled", False)) # ── Violent-move watch resolution helper ───────────────────────────── def violent_watch_enabled(app: Optional[AppConfig] = None) -> bool: """Whether the display-only violent-move watch is on (env wins).""" env = os.environ.get("VIOLENT_WATCH_ENABLED") if env: return env.strip().lower() in ("1", "true", "yes", "on") app = app or get_app_config() return bool(app.violent_watch_settings.get("enabled", False)) # ── Signal-log resolution helpers (env → config.json → default) ────── def signal_log_enabled(app: Optional[AppConfig] = None) -> bool: """Default ON — signal history is the platform's audit trail.""" env = os.environ.get("SIGNAL_LOG_ENABLED") if env: return env.strip().lower() in ("1", "true", "yes", "on") app = app or get_app_config() return bool(app.signal_log_settings.get("enabled", True)) def signal_log_blocked_scans(app: Optional[AppConfig] = None) -> bool: """Whether weekend/day-filter early returns are logged (as minimal rows without news/candles). Default ON — they are liveness evidence; turn off to keep the history table to real scans only.""" env = os.environ.get("SIGNAL_LOG_BLOCKED_SCANS") if env: return env.strip().lower() in ("1", "true", "yes", "on") app = app or get_app_config() return bool(app.signal_log_settings.get("log_blocked_scans", True)) def signal_log_path(app: Optional[AppConfig] = None) -> str: """Absolute path to the SQLite signal-history DB. A relative config value resolves against the repo root so CWD never matters.""" env = os.environ.get("SIGNAL_LOG_PATH") if env: path = env else: app = app or get_app_config() path = str(app.signal_log_settings.get("path", "logs/signals.db")) if not os.path.isabs(path): path = os.path.join(os.path.dirname(os.path.abspath(__file__)), path) return path # ── 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