harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add app_config.py: config.json loader, validation, env-precedence resolvers

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b
This commit is contained in:
Harwinder Sekhon
2026-07-21 12:47:48 -04:00
parent 8de77deadd
commit 2c070fb9b3
2 changed files with 376 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app_config.py
+240
View File
@@ -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
test_all.py
+136
View File
@@ -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
# ══════════════════════════════════════════════════════════════════════