Deploy GitHub main 0d2efcf4a46b

cloud/acceptance.py

@@ -0,0 +1,101 @@
+"""Run an optional real model acceptance, isolated from production scores/state."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import sys
+import tempfile
+import time
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path[:0] = [str(ROOT), str(ROOT / "src")]
+BRANCH = "acceptance"
+
+
+def publish(api, repo, output, report):
+    from huggingface_hub import CommitOperationAdd
+    from tsfm_bench.remote_state import pack_files
+    if not api.dataset_info(repo).private:
+        raise ValueError("Acceptance artifacts require a private Dataset")
+    api.create_branch(repo, branch=BRANCH, repo_type="dataset", exist_ok=True)
+    parent = api.dataset_info(repo, revision=BRANCH).sha
+    prefix = "acceptance/" + output.name
+    with tempfile.TemporaryDirectory(prefix="livehouse-acceptance-") as temp:
+        archive = Path(temp) / "artifacts.tar.gz"
+        manifest = pack_files(ROOT, sorted(p for p in output.rglob("*") if p.is_file()), archive)
+        return api.create_commit(repo, repo_type="dataset", revision=BRANCH, parent_commit=parent,
+            commit_message=f"Cloud model acceptance {output.name}: {report['status']}", operations=[
+                CommitOperationAdd(path_in_repo=prefix + "/artifacts.tar.gz", path_or_fileobj=archive),
+                CommitOperationAdd(path_in_repo=prefix + "/manifest.json", path_or_fileobj=json.dumps(manifest).encode()),
+                CommitOperationAdd(path_in_repo=prefix + "/report.json", path_or_fileobj=json.dumps(report, indent=2).encode()),
+            ])
+
+
+def main():
+    from huggingface_hub import HfApi, hf_hub_download
+    from scripts.run_online_daemon import run_managed_subprocess
+    from tsfm_bench.remote_state import restore_archive
+    identifier = os.environ["LIVEHOUSE_ACCEPTANCE_ID"]
+    if not re.fullmatch(r"[a-z0-9][a-z0-9-]{1,60}", identifier):
+        raise ValueError("Acceptance ID must contain only lowercase letters, digits and hyphens")
+    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
+    repo = os.environ["HF_STATE_REPO"]
+    token = os.environ["HF_TOKEN"]
+    api = HfApi(token=token)
+    if not api.dataset_info(repo).private:
+        raise ValueError("Acceptance artifacts require a private Dataset")
+    output = ROOT / "outputs/cloud-acceptance" / identifier
+    output.mkdir(parents=True, exist_ok=True)
+    prefix = "acceptance/" + identifier
+    api.create_branch(repo, branch=BRANCH, repo_type="dataset", exist_ok=True)
+    revision = api.dataset_info(repo, revision=BRANCH).sha
+    def download(name):
+        return Path(hf_hub_download(repo, prefix + "/" + name, repo_type="dataset",
+                                   revision=revision, token=token))
+    report = {"id":identifier, "isolated":True, "space":os.getenv("SPACE_ID"),
+              "source_revision":os.environ["LIVEHOUSE_SOURCE_REVISION"],
+              "started_at":datetime.now(timezone.utc).isoformat()}
+    if api.file_exists(repo, prefix + "/report.json", repo_type="dataset", revision=revision):
+        report = json.loads(download("report.json").read_text())
+        if report["status"] in {"complete", "failed"}:
+            print(f"Acceptance {identifier} already {report['status']}; no inference repeated", flush=True)
+            if report["status"] == "failed":
+                raise RuntimeError("This acceptance already failed; inspect its archived report")
+            return
+        manifest = json.loads(download("manifest.json").read_text())
+        restore_archive(ROOT, download("artifacts.tar.gz"), manifest, allowed_roots=(output.relative_to(ROOT),))
+    def run(phase):
+        return run_managed_subprocess([sys.executable, "scripts/live_acceptance.py", phase,
+            "--output-root", str(output)], timeout=1200, timeout_message="Cloud acceptance timed out")
+    try:
+        if report.get("status") != "issued":
+            if run("issue") != 0:
+                raise RuntimeError("At least one real model forecast failed; inspect archived acceptance-issue.json")
+            report.update(status="issued", issue=json.loads((output / "acceptance-issue.json").read_text()))
+            publish(api, repo, output, report)
+        # The target bar starts at the forecast deadline and closes five minutes later.
+        deadline = datetime.fromisoformat(report["issue"]["forecast_deadline"]).timestamp() + 310
+        while time.time() < deadline:
+            time.sleep(min(15, deadline - time.time()))
+        for attempt in range(4):
+            if run("resolve") == 0:
+                break
+            if attempt == 3:
+                raise RuntimeError("Real future observations did not resolve every model")
+            time.sleep(30)
+        report.update(status="complete", resolve=json.loads((output / "acceptance-resolve.json").read_text()))
+    except Exception as exc:
+        report.update(status="failed", error_type=type(exc).__name__)
+    report["finished_at"] = datetime.now(timezone.utc).isoformat()
+    publish(api, repo, output, report)
+    print(f"Cloud acceptance {identifier}: {report['status']}; artifacts on private Dataset branch {BRANCH}", flush=True)
+    if report["status"] != "complete":
+        raise RuntimeError("Cloud model acceptance failed; inspect its private artifacts")
+
+
+if __name__ == "__main__":
+    main()

cloud/entrypoint.py

@@ -44,6 +44,9 @@ def main():
     commands = {"web": [sys.executable, "cloud/web.py"]}
     if enabled:
         commands["worker"] = [sys.executable, "cloud/worker.py"]
+    if os.getenv("LIVEHOUSE_ACCEPTANCE_ID"):
+        commands["acceptance"] = [sys.executable, "cloud/acceptance.py"]
+    completed = set()
     stopping = False
 
     def shutdown(*_):
@@ -61,10 +64,15 @@ def main():
             for name, command in commands.items():
                 proc = processes.get(name)
                 if proc is not None and proc.poll() is not None:
+                    if name == "acceptance":
+                        print(f"Acceptance process finished with {proc.returncode}; no automatic rerun", flush=True)
+                        completed.add(name)
+                        del processes[name]
+                        continue
                     print(f"{name} exited with {proc.returncode}; restart after 60 seconds", flush=True)
                     restart_at[name] = time.monotonic() + 60
                     del processes[name]
-                if name not in processes and time.monotonic() >= restart_at.get(name, 0):
+                if name not in processes and name not in completed and time.monotonic() >= restart_at.get(name, 0):
                     processes[name] = subprocess.Popen(command, cwd=ROOT)
             status = {"updated_at": time.time(), "source_revision": revision,
                       "evaluator_enabled": enabled,

deployment.json

@@ -1,5 +1,5 @@
 {
   "source_repository": "https://github.com/Thinkcat-Lab/LiveHouse-TS",
   "source_branch": "main",
-  "source_revision": "3feb6cda151162e151892022bd39aa5615f0256e"
+  "source_revision": "0d2efcf4a46bffe86cd0bf965a279c46cb2ecc27"
 }

scripts/configure_cloud_space.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import re
 from pathlib import Path
 
 
@@ -17,6 +18,9 @@ def configure(api, env):
     enabled = env.get("TSFM_EVALUATOR_ENABLED", "0")
     if enabled not in {"0", "1"}:
         raise ValueError("TSFM_EVALUATOR_ENABLED must be 0 or 1")
+    acceptance = env.get("LIVEHOUSE_ACCEPTANCE_ID", "")
+    if acceptance and not re.fullmatch(r"[a-z0-9][a-z0-9-]{1,60}", acceptance):
+        raise ValueError("Invalid cloud acceptance ID")
     # Validate all destinations before the first secret/variable write.
     api.space_info(repo)
     if not api.dataset_info(state).private:
@@ -28,7 +32,7 @@ def configure(api, env):
     for name in names:
         api.add_space_secret(repo, name, env[name])
     variables = {"HF_SPACE_REPO":repo, "HF_RESULTS_REPO":results, "HF_STATE_REPO":state,
-                 "TSFM_EVALUATOR_ENABLED":enabled}
+                 "TSFM_EVALUATOR_ENABLED":enabled, "LIVEHOUSE_ACCEPTANCE_ID":acceptance}
     for name, value in variables.items():
         api.add_space_variable(repo, name, value)
     return {"space":repo, "secret_names":list(names), "variables":variables}

scripts/live_acceptance.py

@@ -56,7 +56,7 @@ def main():
         if remaining < 180:
             print(f"Waiting {remaining + 3:.0f}s for a full forecast issue window", flush=True)
             time.sleep(remaining + 3)
-    response = requests.get("https://api.binance.com/api/v3/klines", timeout=45,
+    response = requests.get("https://data-api.binance.vision/api/v3/klines", timeout=45,
                             params={"symbol": "BTCUSDT", "interval": "5m", "limit": 130})
     response.raise_for_status()
     rows = response.json()