momentum devandClaude Opus 5.5
b3f023f6e2
The live (Tradier) stack has no backup service. Only the paper stack backs up its own DB, so the 80-trade test's whole record lived only on the TrueNAS pool. - deploy/backup_live_db.py, run by Task Scheduler "Momentum\LiveDbBackup" at 20:40 ET nightly: pg_dump -Fc (PostgreSQL 18.6 client) of the live DB to C:\Users\Admin\momentum-backups. It is verified with pg_restore --list (the events and config_overrides data must be present), keeps the newest 7, and writes status.json. A broken dump is deleted rather than kept. - research/daily_report.py alerts when the last backup failed or is more than 36 h old. When there are alerts it sends them to Telegram, using the stack's own sender and the credentials in .env. --no-telegram suppresses that for dry runs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B3PXXn2tHBzdv6rYtGPz86
72 lines
3.3 KiB
Python
72 lines
3.3 KiB
Python
"""Nightly off-box backup of the LIVE (Tradier) Postgres, from the owner's workstation.
|
|
|
|
The live stack on TrueNAS has no db-backup service; the paper stack backs up only its own database. This keeps the
|
|
live test's full record (events, trades, settings) somewhere other than the TrueNAS pool.
|
|
|
|
- pg_dump -Fc (compressed, restorable with pg_restore) of the live database (DATABASE_URL host, port 5434)
|
|
- verified: pg_restore --list must read it and it must contain the events and config_overrides tables
|
|
- keeps the newest KEEP dumps (default 7) in LIVE_BACKUP_DIR (default C:\\Users\\Admin\\momentum-backups)
|
|
- writes <dir>/status.json for research.daily_report to alert on a missed or failed backup
|
|
|
|
usage: python deploy/backup_live_db.py (Task Scheduler "Momentum\\LiveDbBackup", nightly)
|
|
restore: pg_restore -d <empty database> --no-owner <dump>
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
from zoneinfo import ZoneInfo
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
PG_BIN = Path(os.environ.get("PG_BIN", r"C:\Users\Admin\pgsql18\bin"))
|
|
OUT = Path(os.environ.get("LIVE_BACKUP_DIR", r"C:\Users\Admin\momentum-backups"))
|
|
KEEP = int(os.environ.get("LIVE_BACKUP_KEEP", "7"))
|
|
PORT = int(os.environ.get("LIVE_DB_PORT", "5434"))
|
|
ET = ZoneInfo("America/New_York")
|
|
MUST_HAVE = ("TABLE DATA public events", "TABLE DATA public config_overrides")
|
|
|
|
|
|
def main() -> int:
|
|
from dotenv import dotenv_values
|
|
u = urlparse(dotenv_values(ROOT / ".env")["DATABASE_URL"])
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
stamp = datetime.now(ET).strftime("%Y-%m-%d_%H%M")
|
|
dump = OUT / f"live_{stamp}.dump"
|
|
env = dict(os.environ, PGPASSWORD=u.password or "")
|
|
status = dict(at=datetime.now(ET).isoformat(timespec="seconds"), file=str(dump), ok=False)
|
|
t0 = time.time()
|
|
try:
|
|
subprocess.run([str(PG_BIN / "pg_dump.exe"), "-h", u.hostname, "-p", str(PORT), "-U", u.username,
|
|
"-d", u.path[1:], "-Fc", "-Z", "6", "-f", str(dump)],
|
|
env=env, check=True, capture_output=True, text=True, timeout=3600)
|
|
listing = subprocess.run([str(PG_BIN / "pg_restore.exe"), "--list", str(dump)],
|
|
check=True, capture_output=True, text=True, timeout=600).stdout
|
|
missing = [m for m in MUST_HAVE if m not in listing]
|
|
if missing:
|
|
raise RuntimeError(f"dump is missing {missing}")
|
|
status.update(ok=True, size_mb=round(dump.stat().st_size / 1e6, 1), seconds=round(time.time() - t0),
|
|
entries=listing.count("\n"))
|
|
except subprocess.CalledProcessError as e:
|
|
status["error"] = (e.stderr or str(e))[-400:]
|
|
except Exception as e: # noqa: BLE001
|
|
status["error"] = str(e)[-400:]
|
|
if not status["ok"] and dump.exists():
|
|
dump.unlink() # never keep a broken dump around as if it were good
|
|
dumps = sorted(OUT.glob("live_*.dump"))
|
|
for old in dumps[:-KEEP]:
|
|
old.unlink()
|
|
status["kept"] = [p.name for p in sorted(OUT.glob("live_*.dump"))]
|
|
(OUT / "status.json").write_text(json.dumps(status, indent=1), encoding="utf-8")
|
|
print(json.dumps(status, indent=1))
|
|
return 0 if status["ok"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|