momentum devandClaude Fable 5
8c62804808
eod_reconcile, backfill_journal and the admin health block were hardwired to AlpacaPaperClient — on a BROKER=tradier stack they would silently reconcile/report against the wrong broker. make_trading_client() honors BROKER env with no silent cross-broker fallback (missing Tradier keys => skip with reason, never fall back to Alpaca). Admin UI shows the active broker label; JSON key stays 'alpaca' for compatibility. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
438 lines
19 KiB
Python
438 lines
19 KiB
Python
"""Momentum ADMIN panel (:8767) — configure, validate, and control the stack.
|
|
|
|
LAN-only by intent; every mutating endpoint (and the page itself) requires the
|
|
ADMIN_TOKEN (env). Capabilities:
|
|
- service health: docker containers (via /var/run/docker.sock), DB, Alpaca,
|
|
Ollama, loop heartbeat, market clock, data feed
|
|
- config: read/edit the whitelisted strategy knobs via data/overrides.env
|
|
(beats compose env on next boot; see runtime/env_overrides.py); secrets are
|
|
shown as present/absent only, never values
|
|
- profiles: named knob presets (data/profiles/*.json) applied in one click
|
|
- restart: bounce individual containers so config changes take effect
|
|
- paper/live switch: guarded — refuses unless ALPACA_LIVE_* keys are set
|
|
|
|
Run: python admin_api.py (ADMIN_HOST/ADMIN_PORT, default 0.0.0.0:8767)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import http.client
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import urllib.request
|
|
from datetime import datetime
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
if __name__ == "__main__":
|
|
# Load the repo .env ONLY when this file is run as the entrypoint
|
|
# (`python admin_api.py`). A bare module-level load_dotenv() here used to
|
|
# fire on every import too -- including pytest collection -- leaking the
|
|
# real repo .env (secrets included) into any process that merely imports
|
|
# this module. This runs before the os.environ.get(...) constants below,
|
|
# so `python admin_api.py` still behaves identically.
|
|
load_dotenv()
|
|
from runtime.env_overrides import load as _load_overrides # noqa: E402
|
|
_load_overrides()
|
|
from runtime import env_overrides # noqa: E402
|
|
|
|
DOCKER_SOCK = os.environ.get("DOCKER_SOCK", "/var/run/docker.sock")
|
|
# Regime backtest (plan 011) on-demand trigger — logs to the app-data volume so
|
|
# GET /api/regime/status can tail it (the panel had no way to fire this).
|
|
REGIME_LOG = os.environ.get(
|
|
"REGIME_RUN_LOG", os.path.join(os.path.dirname(__file__), "data", "regime_run.log"))
|
|
_REGIME_DONE = "REGIME PIPELINE DONE"
|
|
|
|
|
|
def run_regime_pipeline() -> int:
|
|
"""Fire the full regime shadow pipeline DETACHED (it takes a few min): rebuild
|
|
session labels -> walk-forward score both models -> replay + adjudicate the
|
|
regime_gate experiments. Returns the child pid. Shadow only — never touches a
|
|
live order. stdout+stderr -> REGIME_LOG (truncated each run so status shows
|
|
the CURRENT run); a trailing marker lets status report completion."""
|
|
py = sys.executable
|
|
steps = " && ".join([
|
|
f"{py} -m research.regime build-features",
|
|
f"{py} -m research.regime score --model threshold_v1",
|
|
f"{py} -m research.regime score --model meta_v1",
|
|
f"{py} -m research.experiments run",
|
|
f"{py} -m research.experiments adjudicate",
|
|
f"echo {_REGIME_DONE}",
|
|
])
|
|
os.makedirs(os.path.dirname(REGIME_LOG), exist_ok=True)
|
|
log = open(REGIME_LOG, "wb") # noqa: SIM115 — handed to the detached child
|
|
proc = subprocess.Popen(["/bin/sh", "-c", steps], stdout=log,
|
|
stderr=subprocess.STDOUT, start_new_session=True)
|
|
return proc.pid
|
|
|
|
|
|
def regime_status() -> dict:
|
|
"""Tail REGIME_LOG + whether the last run finished (the marker) — no DB query
|
|
needed; the adjudicate step prints each experiment's verdict into the log."""
|
|
try:
|
|
with open(REGIME_LOG, encoding="utf-8", errors="replace") as fh:
|
|
lines = fh.read().splitlines()
|
|
except FileNotFoundError:
|
|
return {"log_tail": [], "running": False, "ran": False}
|
|
return {"log_tail": lines[-40:], "ran": True,
|
|
"running": _REGIME_DONE not in lines[-3:] if lines else False}
|
|
PROFILES_DIR = os.environ.get(
|
|
"PROFILES_DIR", os.path.join(os.path.dirname(__file__), "data", "profiles"))
|
|
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "")
|
|
|
|
# knobs the UI may read+write — everything else is off-limits by construction
|
|
EDITABLE_PREFIXES = ("TRADING_", "TRIGGER_", "WATCHER_", "LIVE_BARS_", "NEWS_",
|
|
"OLLAMA_", "DISCOVER_", "SYMBOL_WEIGHTS")
|
|
EDITABLE_EXACT = {"ALPACA_DATA_FEED"}
|
|
SECRET_MARKERS = ("KEY", "SECRET", "TOKEN", "PASSWORD", "WEBHOOK", "URL")
|
|
|
|
SEED_PROFILES = {
|
|
"validated-core": {
|
|
"_note": "The year-validated population config (2026-07-09 state).",
|
|
"TRADING_MIN_QUALITY_SCORE": "0.60", "TRADING_MAX_FRESH_ENTRIES_PER_DAY": "10",
|
|
"TRADING_MIN_ENTRY_GAP_PCT": "0.10", "TRADING_MIN_DAY_VOLUME": "300000",
|
|
"TRADING_PRICE_MIN": "2.0", "TRADING_PRICE_MAX": "20.0",
|
|
"TRIGGER_GAP_MIN": "10.0", "TRIGGER_GAP_MAX": "35",
|
|
"TRIGGER_GAP_SQUEEZE_MIN": "150", "TRIGGER_GAP_SQUEEZE_RVOL": "5.0",
|
|
},
|
|
"conservative": {
|
|
"_note": "Fewer, stricter entries: B-grade floor, half the daily cap, no squeeze lottery.",
|
|
"TRADING_MIN_QUALITY_SCORE": "0.75", "TRADING_MAX_FRESH_ENTRIES_PER_DAY": "5",
|
|
"TRADING_MIN_ENTRY_GAP_PCT": "0.10", "TRADING_MIN_DAY_VOLUME": "300000",
|
|
"TRADING_PRICE_MIN": "2.0", "TRADING_PRICE_MAX": "20.0",
|
|
"TRIGGER_GAP_MIN": "10.0", "TRIGGER_GAP_MAX": "35",
|
|
"TRIGGER_GAP_SQUEEZE_MIN": "0",
|
|
},
|
|
"squeeze-hunter": {
|
|
"_note": "Experiment: lean into the >150% volume-verified squeeze band + looser quality.",
|
|
"TRADING_MIN_QUALITY_SCORE": "0.50", "TRADING_MAX_FRESH_ENTRIES_PER_DAY": "10",
|
|
"TRADING_MIN_ENTRY_GAP_PCT": "0.10", "TRADING_MIN_DAY_VOLUME": "300000",
|
|
"TRADING_PRICE_MIN": "2.0", "TRADING_PRICE_MAX": "20.0",
|
|
"TRIGGER_GAP_MIN": "10.0", "TRIGGER_GAP_MAX": "35",
|
|
"TRIGGER_GAP_SQUEEZE_MIN": "150", "TRIGGER_GAP_SQUEEZE_RVOL": "3.0",
|
|
},
|
|
}
|
|
|
|
|
|
def _safe_overrides() -> dict:
|
|
"""Overrides for API responses with secret VALUES redacted — a /api/mode live
|
|
switch writes ALPACA_LIVE_* into overrides.env; without this they'd echo back
|
|
in every /api/status and /api/config response."""
|
|
merged = {**env_overrides.read_overrides(), **env_overrides.read_db_overrides()}
|
|
return {k: ("***" if any(m in k for m in SECRET_MARKERS) else v)
|
|
for k, v in merged.items()}
|
|
|
|
|
|
def _editable(key: str) -> bool:
|
|
if key in EDITABLE_EXACT:
|
|
return True
|
|
return (key.startswith(EDITABLE_PREFIXES)
|
|
and not any(m in key for m in SECRET_MARKERS))
|
|
|
|
|
|
def _mask_value(key: str, value: str) -> str:
|
|
"""Mask secret VALUES (*** + last 4). Uses the broader SECRET_MARKERS set so
|
|
DATABASE_URL / *_WEBHOOK_URL are masked too — narrowing to KEY|SECRET|TOKEN|
|
|
PASSWORD would leak the DB password in the config table."""
|
|
if any(m in key.upper() for m in SECRET_MARKERS):
|
|
tail = value[-4:] if value and len(value) >= 4 else ""
|
|
return "***" + tail
|
|
return value
|
|
|
|
|
|
def _env_example_keys() -> set[str]:
|
|
"""Keys documented in .env.example — the 'required' config surface used for
|
|
drift detection (a documented key absent from the effective env is drift)."""
|
|
path = os.path.join(os.path.dirname(__file__), ".env.example")
|
|
keys: set[str] = set()
|
|
try:
|
|
with open(path) as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
keys.add(line.split("=", 1)[0].strip())
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return keys
|
|
|
|
|
|
def effective_config() -> dict:
|
|
"""Per-key effective value + source for the admin config table.
|
|
|
|
Merge order baked < compose < overrides.env (< learned). Sources we can prove
|
|
at runtime: ``overrides.env`` (from data/overrides.env) vs ``compose`` (the
|
|
process env — baked and compose are indistinguishable inside the container, so
|
|
both surface as ``compose``); a documented .env.example key that is absent
|
|
surfaces as ``missing`` (drift). Secret values are masked."""
|
|
overrides = env_overrides.read_overrides()
|
|
db_overrides = env_overrides.read_db_overrides()
|
|
example = _env_example_keys()
|
|
keys = (set(example) | set(overrides) | set(db_overrides)
|
|
| {k for k in os.environ if _editable(k)})
|
|
out: dict[str, dict] = {}
|
|
for key in sorted(keys):
|
|
value = os.environ.get(key)
|
|
if value is None:
|
|
out[key] = {"value_masked": None,
|
|
"source": "missing" if key in example else "unset",
|
|
"drift": key in example}
|
|
continue
|
|
source = ("database" if key in db_overrides
|
|
else "overrides.env" if key in overrides else "compose")
|
|
out[key] = {"value_masked": _mask_value(key, str(value)), "source": source}
|
|
return out
|
|
|
|
|
|
# ---------- docker over the unix socket (stdlib only) ----------
|
|
class _SockConn(http.client.HTTPConnection):
|
|
def __init__(self):
|
|
super().__init__("localhost")
|
|
|
|
def connect(self):
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
s.connect(DOCKER_SOCK)
|
|
self.sock = s
|
|
|
|
|
|
def docker(method: str, path: str) -> tuple[int, object]:
|
|
try:
|
|
c = _SockConn()
|
|
c.request(method, path)
|
|
r = c.getresponse()
|
|
body = r.read()
|
|
return r.status, (json.loads(body) if body else None)
|
|
except Exception as exc: # noqa: BLE001
|
|
return 0, {"error": str(exc)}
|
|
|
|
|
|
def list_containers() -> list[dict]:
|
|
st, data = docker("GET", "/containers/json?all=1")
|
|
if st != 200 or not isinstance(data, list):
|
|
return []
|
|
out = []
|
|
for c in data:
|
|
name = (c.get("Names") or ["?"])[0].lstrip("/")
|
|
if "momentum" not in name and "postgres" not in name:
|
|
continue
|
|
out.append({
|
|
"name": name,
|
|
"state": c.get("State"),
|
|
"status": c.get("Status"),
|
|
"health": ("healthy" if "(healthy)" in (c.get("Status") or "")
|
|
else "unhealthy" if "(unhealthy)" in (c.get("Status") or "")
|
|
else None),
|
|
"image": c.get("Image"),
|
|
"id": c.get("Id", "")[:12],
|
|
})
|
|
return sorted(out, key=lambda x: x["name"])
|
|
|
|
|
|
# ---------- health checks ----------
|
|
def checks() -> dict:
|
|
out: dict = {}
|
|
try:
|
|
from research.multi_schema import open_research_db
|
|
con = open_research_db("market")
|
|
r = con.execute("SELECT max(timestamp) FROM events").fetchall()[0][0]
|
|
age = (datetime.now() - r).total_seconds() if r else None
|
|
out["db"] = {"ok": True, "newest_event_age_s": round(age) if age else None}
|
|
except Exception as exc: # noqa: BLE001
|
|
out["db"] = {"ok": False, "error": str(exc)[:120]}
|
|
try:
|
|
# active execution broker (BROKER env) — JSON key stays "alpaca" for
|
|
# UI compatibility; "name" carries the real broker label
|
|
from tradier_broker.factory import make_trading_client
|
|
c, label = make_trading_client()
|
|
if c is None:
|
|
out["alpaca"] = {"ok": False, "error": label}
|
|
else:
|
|
clock = c.get_clock()
|
|
acct = c.get_account()
|
|
out["alpaca"] = {"ok": True, "market_open": clock.get("is_open"),
|
|
"equity": acct.get("equity"), "name": label,
|
|
"mode": "LIVE" if label.endswith("LIVE") else "paper"}
|
|
except Exception as exc: # noqa: BLE001
|
|
out["alpaca"] = {"ok": False, "error": str(exc)[:120]}
|
|
try:
|
|
host = os.environ.get("OLLAMA_HOST", "http://192.168.1.9:11434")
|
|
with urllib.request.urlopen(host + "/api/tags", timeout=4) as r:
|
|
out["ollama"] = {"ok": r.status == 200}
|
|
except Exception: # noqa: BLE001
|
|
out["ollama"] = {"ok": False, "note": "enrichment fails safe; trading unaffected"}
|
|
out["feed"] = os.environ.get("ALPACA_DATA_FEED", "iex")
|
|
return out
|
|
|
|
|
|
# ---------- profiles ----------
|
|
def seed_profiles() -> None:
|
|
try:
|
|
os.makedirs(PROFILES_DIR, exist_ok=True)
|
|
for name, body in SEED_PROFILES.items():
|
|
p = os.path.join(PROFILES_DIR, f"{name}.json")
|
|
if not os.path.exists(p):
|
|
json.dump(body, open(p, "w"), indent=2)
|
|
except PermissionError:
|
|
# host volume not writable by the container user (uid 10001) — degrade
|
|
# instead of crash-looping; profiles/config editing will 500 until the
|
|
# host dir is chown'd (see deploy/docker/README.md)
|
|
print(f"[admin] WARNING: {PROFILES_DIR} not writable (chown -R 10001:10001 "
|
|
"the data dir on the host); profiles/config editing disabled")
|
|
|
|
|
|
def profiles() -> dict:
|
|
seed_profiles()
|
|
out = {}
|
|
for f in sorted(os.listdir(PROFILES_DIR)):
|
|
if f.endswith(".json"):
|
|
try:
|
|
out[f[:-5]] = json.load(open(os.path.join(PROFILES_DIR, f)))
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return out
|
|
|
|
|
|
# ---------- HTTP ----------
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def _send(self, code: int, body, ctype="application/json"):
|
|
raw = body if isinstance(body, bytes) else json.dumps(body, default=str).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
def _authed(self) -> bool:
|
|
# FAIL CLOSED: no token configured => deny (this panel fronts the docker
|
|
# socket + the paper/live switch; an unset env var must not open it to the
|
|
# LAN). Header only — a ?token= query would leak into proxy/access logs.
|
|
if not ADMIN_TOKEN:
|
|
return False
|
|
return hmac.compare_digest(self.headers.get("X-Admin-Token", ""), ADMIN_TOKEN)
|
|
|
|
def do_GET(self): # noqa: N802
|
|
path = self.path.split("?")[0]
|
|
if path == "/":
|
|
page = os.path.join(os.path.dirname(__file__), "api", "static", "admin.html")
|
|
return self._send(200, open(page, "rb").read(), "text/html")
|
|
if not self._authed():
|
|
return self._send(401, {"error": "bad or missing X-Admin-Token"})
|
|
if path == "/api/status":
|
|
return self._send(200, {"containers": list_containers(), "checks": checks(),
|
|
"overrides": _safe_overrides(),
|
|
"version": os.environ.get("APP_BUILD_TAG", "dev"),
|
|
"git_sha": os.environ.get("APP_GIT_SHA", "dev")[:12]})
|
|
if path == "/api/config":
|
|
eff = {k: v for k, v in sorted(os.environ.items()) if _editable(k)}
|
|
secrets = {k: bool(os.environ.get(k)) for k in
|
|
("ALPACA_API_KEY", "ALPACA_SECRET_KEY", "ALPACA_LIVE_API_KEY",
|
|
"ALPACA_LIVE_SECRET_KEY", "TELEGRAM_BOT_TOKEN", "YOUTUBE_API_KEY",
|
|
"FMP_API_KEY")}
|
|
return self._send(200, {"effective": eff, "secrets_present": secrets,
|
|
"overrides": _safe_overrides()})
|
|
if path == "/api/profiles":
|
|
return self._send(200, profiles())
|
|
if path == "/api/config/effective":
|
|
return self._send(200, {"config": effective_config()})
|
|
if path == "/api/regime/status":
|
|
return self._send(200, regime_status())
|
|
return self._send(404, {"error": "unknown route"})
|
|
|
|
def do_POST(self): # noqa: N802
|
|
if not self._authed():
|
|
return self._send(401, {"error": "bad or missing X-Admin-Token"})
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
try:
|
|
body = json.loads(self.rfile.read(n) or b"{}")
|
|
except Exception: # noqa: BLE001
|
|
return self._send(400, {"error": "bad json"})
|
|
path = self.path.split("?")[0]
|
|
|
|
if path == "/api/config":
|
|
updates = body.get("set") or {}
|
|
deletes = body.get("delete") or []
|
|
bad = [k for k in list(updates) + list(deletes) if not _editable(k)]
|
|
if bad:
|
|
return self._send(400, {"error": f"not editable: {bad}"})
|
|
# knobs go to the DATABASE (owner-facing store; whitelist above
|
|
# already refused secrets). overrides.env keeps secrets/bootstrap.
|
|
try:
|
|
merged = env_overrides.write_db_overrides(
|
|
{**{k: str(v) for k, v in updates.items()},
|
|
**{k: None for k in deletes}})
|
|
except Exception as exc: # noqa: BLE001
|
|
return self._send(502, {"error": f"db write failed: {exc}"})
|
|
return self._send(200, {"ok": True, "overrides": merged,
|
|
"note": "restart affected services to apply"})
|
|
|
|
if path == "/api/profiles/apply":
|
|
name = body.get("name", "")
|
|
prof = profiles().get(name)
|
|
if not prof:
|
|
return self._send(404, {"error": f"no profile {name!r}"})
|
|
knobs = {k: str(v) for k, v in prof.items() if not k.startswith("_")}
|
|
merged = env_overrides.write_db_overrides(knobs)
|
|
return self._send(200, {"ok": True, "applied": name, "overrides": merged,
|
|
"note": "restart the loop to apply"})
|
|
|
|
if path == "/api/restart":
|
|
name = body.get("service", "")
|
|
target = next((c for c in list_containers() if c["name"] == name), None)
|
|
if not target:
|
|
return self._send(404, {"error": f"no container {name!r}"})
|
|
st, _ = docker("POST", f"/containers/{target['id']}/restart?t=10")
|
|
return self._send(200 if st in (200, 204) else 502,
|
|
{"ok": st in (200, 204), "docker_status": st})
|
|
|
|
if path == "/api/mode":
|
|
# paper/live switch — guarded: live requires ALPACA_LIVE_* keys present
|
|
mode = body.get("mode")
|
|
if mode == "live":
|
|
if not (os.environ.get("ALPACA_LIVE_API_KEY")
|
|
and os.environ.get("ALPACA_LIVE_SECRET_KEY")):
|
|
return self._send(400, {"error": "ALPACA_LIVE_API_KEY/SECRET not set — "
|
|
"refusing to switch to live"})
|
|
merged = env_overrides.write_overrides({
|
|
"ALPACA_BASE_URL": "https://api.alpaca.markets",
|
|
"ALPACA_API_KEY": os.environ["ALPACA_LIVE_API_KEY"],
|
|
"ALPACA_SECRET_KEY": os.environ["ALPACA_LIVE_SECRET_KEY"]})
|
|
elif mode == "paper":
|
|
merged = env_overrides.write_overrides({
|
|
"ALPACA_BASE_URL": None, "ALPACA_API_KEY": None,
|
|
"ALPACA_SECRET_KEY": None})
|
|
else:
|
|
return self._send(400, {"error": "mode must be paper|live"})
|
|
return self._send(200, {"ok": True, "mode": mode, "overrides": merged,
|
|
"note": "restart the loop to apply"})
|
|
|
|
if path == "/api/regime/run":
|
|
try:
|
|
pid = run_regime_pipeline()
|
|
except Exception as e: # noqa: BLE001
|
|
return self._send(500, {"ok": False, "error": str(e)[:200]})
|
|
return self._send(200, {"ok": True, "pid": pid, "log": REGIME_LOG,
|
|
"note": "regime shadow pipeline started (~a few min); "
|
|
"poll GET /api/regime/status"})
|
|
return self._send(404, {"error": "unknown route"})
|
|
|
|
def log_message(self, *a): # quiet
|
|
pass
|
|
|
|
|
|
def main() -> None:
|
|
host = os.environ.get("ADMIN_HOST", "0.0.0.0")
|
|
port = int(os.environ.get("ADMIN_PORT", "8767"))
|
|
seed_profiles()
|
|
print(f"[admin] serving http://{host}:{port} "
|
|
f"(token {'SET' if ADMIN_TOKEN else 'NOT SET — open access'}; "
|
|
f"docker sock {'ok' if os.path.exists(DOCKER_SOCK) else 'MISSING'})")
|
|
ThreadingHTTPServer((host, port), Handler).serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|