New admin service — LAN-only, token-authed (ADMIN_TOKEN), card-based UI matching the owner's preferred dashboard style. Capabilities: - health: docker containers via /var/run/docker.sock (stdlib unix-socket HTTP client), DB heartbeat, Alpaca account/clock, Ollama, feed - config: whitelisted strategy knobs (TRADING_/TRIGGER_/WATCHER_/...) edited via data/overrides.env, which every entrypoint now loads with override=True at boot (runtime/env_overrides.py) — so changes apply on container restart without touching the deployment YAML; secrets are presence-only and rejected server-side if edited - profiles: named knob presets (seeded: validated-core, conservative, squeeze-hunter) applied in one click - restart buttons per container; paper/live switch guarded behind ALPACA_LIVE_* keys existing Verified live in docker: auth 401s, status/config/profiles/write paths, secret-edit rejection, page serves. Added to both compose files. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
"""Runtime env overrides: /app/data/overrides.env (shared volume) beats the
|
|
compose-baked environment. The admin panel writes this file and restarts the
|
|
service; every entrypoint calls load() first so changes take effect on boot
|
|
without touching the deployment YAML."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
OVERRIDES_FILE = os.environ.get("ENV_OVERRIDES_FILE") or os.path.join(
|
|
os.path.dirname(__file__), "..", "data", "overrides.env")
|
|
|
|
|
|
def load() -> int:
|
|
"""Apply overrides (override=True). Returns count of applied keys."""
|
|
try:
|
|
from dotenv import load_dotenv
|
|
if os.path.exists(OVERRIDES_FILE):
|
|
before = dict(os.environ)
|
|
load_dotenv(OVERRIDES_FILE, override=True)
|
|
return sum(1 for k, v in os.environ.items() if before.get(k) != v)
|
|
except Exception: # noqa: BLE001 — a broken overrides file must not stop boot
|
|
pass
|
|
return 0
|
|
|
|
|
|
def read_overrides() -> dict[str, str]:
|
|
out: dict[str, str] = {}
|
|
if not os.path.exists(OVERRIDES_FILE):
|
|
return out
|
|
for line in open(OVERRIDES_FILE):
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, _, v = line.partition("=")
|
|
out[k.strip()] = v.strip()
|
|
return out
|
|
|
|
|
|
def write_overrides(updates: dict[str, str | None]) -> dict[str, str]:
|
|
"""Merge updates into the overrides file (None deletes a key)."""
|
|
cur = read_overrides()
|
|
for k, v in updates.items():
|
|
if v is None:
|
|
cur.pop(k, None)
|
|
else:
|
|
cur[k] = str(v)
|
|
os.makedirs(os.path.dirname(os.path.abspath(OVERRIDES_FILE)), exist_ok=True)
|
|
with open(OVERRIDES_FILE, "w") as f:
|
|
f.write("# Written by the momentum admin panel — beats compose env at boot.\n")
|
|
for k in sorted(cur):
|
|
f.write(f"{k}={cur[k]}\n")
|
|
return cur
|