Harwinder Sekhon
251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""
|
|
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
|