""" runtime_config.py Small key-value store for the subset of settings that are safe to change live from the dashboard, without editing .env or restarting the process. SQLite is primary storage (logs/runtime_config.db); every write is also mirrored to a JSON file (logs/runtime_config.json) so there's a human-readable backup and a fallback read path if the DB is ever unreadable (corrupted, locked, whatever) — reads try the DB first, then fall back to JSON automatically. This module knows nothing about which keys are "allowed" — that allowlist lives in config.py's TUNABLE_REGISTRY, and webapp/server.py enforces it before ever calling set_raw() here. This module is just storage. """ import json import logging import os import sqlite3 import time from contextlib import closing from datetime import datetime, timezone log = logging.getLogger("wickbot.runtime_config") _DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs", "runtime_config.db") _JSON_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs", "runtime_config.json") SCHEMA = "CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT, updated_at TEXT)" _CACHE_TTL_SECONDS = 5 # avoid hitting SQLite on every single Config.X property access _cache = {} # key -> (value, fetched_at) def _connect(): os.makedirs(os.path.dirname(_DB_PATH), exist_ok=True) conn = sqlite3.connect(_DB_PATH) conn.execute(SCHEMA) return conn def _load_json() -> dict: try: with open(_JSON_PATH) as f: return json.load(f) except Exception: return {} def _save_json(data: dict): try: os.makedirs(os.path.dirname(_JSON_PATH), exist_ok=True) with open(_JSON_PATH, "w") as f: json.dump(data, f, indent=2, sort_keys=True) except Exception: log.exception("Failed to write JSON fallback config file") def get_raw(key: str): """Returns the raw string value for `key` if an override exists, else None (meaning: use the .env/factory default). Cached briefly to avoid a SQLite round-trip on every single Config.X access in hot loops (strategy evaluation, position management, etc.).""" cached = _cache.get(key) if cached and (time.time() - cached[1]) < _CACHE_TTL_SECONDS: return cached[0] value = None try: with closing(_connect()) as conn: row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone() if row is not None: value = row[0] except Exception: log.warning("runtime_config DB read failed for %s, falling back to JSON", key, exc_info=True) value = _load_json().get(key) _cache[key] = (value, time.time()) return value def set_raw(key: str, value) -> None: """Write-through: saves to the DB AND mirrors to the JSON file, so the JSON file is always a faithful, current backup — not just a last-resort path that might be stale.""" value_str = str(value) ts = datetime.now(timezone.utc).isoformat() try: with closing(_connect()) as conn: conn.execute( "INSERT INTO settings (key, value, updated_at) VALUES (?, ?, ?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at", (key, value_str, ts), ) conn.commit() except Exception: log.exception("runtime_config DB write failed for %s — saved to JSON fallback only", key) json_data = _load_json() json_data[key] = value_str _save_json(json_data) _cache[key] = (value_str, time.time()) def delete(key: str) -> None: """Removes an override entirely, reverting that setting back to its .env/factory default on the next read.""" try: with closing(_connect()) as conn: conn.execute("DELETE FROM settings WHERE key=?", (key,)) conn.commit() except Exception: log.exception("runtime_config DB delete failed for %s", key) json_data = _load_json() json_data.pop(key, None) _save_json(json_data) _cache.pop(key, None) def get_all_raw() -> dict: """Merges DB + JSON (DB wins on conflict) — used to populate the dashboard's settings page. Always hits storage directly (not the short-lived cache above), since this is called far less often.""" merged = dict(_load_json()) try: with closing(_connect()) as conn: rows = conn.execute("SELECT key, value FROM settings").fetchall() merged.update({k: v for k, v in rows}) except Exception: log.warning("runtime_config DB read failed for get_all_raw(), using JSON only", exc_info=True) return merged def parse_symbols_list(raw) -> list: """Robustly coerce a SYMBOLS setting value into a plain list of strings. Handles every shape the value can take as it round-trips through .env, the dashboard, and the JSON/DB storage layers: - already a list: ['EURUSD'] -> ['EURUSD'] - a JSON string of a list: '["EURUSD"]' -> ['EURUSD'] - a doubly-encoded string: "['[\"EURUSD\"]']" -> ['EURUSD'] - a comma-separated string: 'BTCUSDTm,BTCUSDm' -> ['BTCUSDTm','BTCUSDm'] - a single bare string: 'EURUSD' -> ['EURUSD'] Strips surrounding quotes/brackets defensively and never raises - on anything it can't parse it returns an empty list rather than crashing a live trading loop over a corrupted setting value. """ if raw is None: return [] if isinstance(raw, (list, tuple)): return [str(s).strip() for s in raw if str(s).strip()] text = str(raw).strip() if not text: return [] candidate = text for _ in range(5): try: parsed = json.loads(candidate) except (ValueError, TypeError): break if isinstance(parsed, list): result = [] for item in parsed: if isinstance(item, (list, tuple)): result.extend(str(s).strip() for s in item if str(s).strip()) else: item_str = str(item).strip() if item_str: result.append(item_str) return result if isinstance(parsed, str): candidate = parsed continue return [str(parsed).strip()] if parsed is not None else [] cleaned = text if cleaned.startswith("[") and cleaned.endswith("]"): cleaned = cleaned[1:-1].strip() cleaned = cleaned.strip().strip("'\"").strip() if cleaned.startswith("[") or cleaned.startswith("'") or cleaned.startswith('"'): cleaned = cleaned.strip("[]'\"").strip() if not cleaned: return [] parts = [p.strip().strip("'\"").strip() for p in cleaned.split(",")] return [p for p in parts if p]