sevenhill/TradeBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: production paper trading platform

Browse Source
This commit is contained in:
Курнат Андрей
2026-07-14 22:52:47 +03:00
parent 7186acb9a1
commit 5c4aecfe5f
29 changed files with 396 additions and 564 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_spot_bot/training_coordination.py
+83 -13
View File
@@ -2,9 +2,11 @@ from __future__ import annotations
import base64
import hashlib
import hmac
import json
import os
import re
import secrets
import shutil
import uuid
from datetime import UTC
@@ -20,11 +22,11 @@ ALLOWED_TRAINING_ARTIFACTS = {
"torch_retrain_guard.json",
"torch_threshold_calibration.json",
}
RUNNING_TIMEOUT = timedelta(hours=12)
RUNNING_LEASE_TIMEOUT = timedelta(minutes=10)
ONLINE_WINDOW = timedelta(minutes=3)
MAX_JOB_ATTEMPTS = 3
MAX_ARTIFACT_CHUNK_BYTES = 1024 * 1024
# Independent per-symbol ensembles are intentionally larger than pooled models.
# Keep a bounded limit, but leave enough room for the supported 12-symbol bundle.
# Keep uploads bounded while leaving room for explicitly requested per-symbol bundles.
MAX_ARTIFACT_BYTES = 256 * 1024 * 1024
MAX_ARTIFACT_CHUNKS = 1024
REQUIRED_MODEL_BUNDLE = set(ALLOWED_TRAINING_ARTIFACTS)
@@ -52,7 +54,12 @@ class TrainingCoordinator:
existing = self._active_job(state)
if existing is not None:
self._save_state(state)
return {"queued": False, "reason": "active_job_exists", "job": existing, "status": self._public_status(state)}
return {
"queued": False,
"reason": "active_job_exists",
"job": self._public_job(existing),
"status": self._public_status(state),
}
now = _now()
job = {
@@ -63,11 +70,16 @@ class TrainingCoordinator:
"parameters": _safe_parameters(payload.get("parameters")),
"message": "",
"artifacts": [],
"attempts": 0,
}
state.setdefault("jobs", []).append(job)
self._trim_jobs(state)
self._save_state(state)
return {"queued": True, "job": job, "status": self._public_status(state)}
return {
"queued": True,
"job": self._public_job(job),
"status": self._public_status(state),
}
def heartbeat(self, payload: dict[str, Any] | None = None) -> dict[str, Any]:
payload = payload or {}
@@ -91,12 +103,21 @@ class TrainingCoordinator:
return {"claimed": False, "job": None, "status": self._public_status(state)}
now = _now()
lease_token = secrets.token_urlsafe(32)
job["status"] = "running"
job["claimed_at"] = now
job["updated_at"] = now
job["claimed_by"] = worker["id"]
job["worker"] = worker
job["lease_token"] = lease_token
job["attempts"] = int(job.get("attempts", 0)) + 1
self._save_state(state)
return {"claimed": True, "job": job, "status": self._public_status(state)}
return {
"claimed": True,
"job": self._public_job(job),
"lease_token": lease_token,
"status": self._public_status(state),
}
def save_artifact_chunk(self, job_id: str, payload: dict[str, Any]) -> dict[str, Any]:
job_id = _valid_job_id(job_id)
@@ -126,6 +147,7 @@ class TrainingCoordinator:
raise ValueError(f"training job not found: {job_id}")
if job.get("status") != "running" or not job.get("claimed_by"):
raise ValueError("training job is not claimed and running")
self._require_lease(job, payload)
uploads = job.setdefault("uploads", {})
upload = uploads.setdefault(name, {"sha256": sha256, "total": total})
if upload.get("sha256") != sha256 or int(upload.get("total", 0)) != total:
@@ -138,6 +160,7 @@ class TrainingCoordinator:
received = sum(1 for part in range(total) if (chunk_dir / f"{part:06d}.part").is_file())
if received < total:
upload["received"] = received
job["updated_at"] = _now()
self._save_state(state)
return {"complete": False, "received": received, "total": total}
@@ -169,6 +192,7 @@ class TrainingCoordinator:
{"name": name, "sha256": sha256, "size": size, "staged_at": _now()}
)
job["artifacts"] = artifacts
job["updated_at"] = _now()
upload["received"] = total
upload["complete"] = True
self._save_state(state)
@@ -184,6 +208,7 @@ class TrainingCoordinator:
raise ValueError(f"training job not found: {job_id}")
if job.get("status") != "running" or not job.get("claimed_by"):
raise ValueError("training job is not claimed and running")
self._require_lease(job, payload)
if isinstance(payload.get("worker"), dict):
state["worker"] = self._worker_from_payload(payload["worker"])
job["status"] = "running"
@@ -194,7 +219,11 @@ class TrainingCoordinator:
if isinstance(payload.get("details"), dict):
job["details"] = payload["details"]
self._save_state(state)
return {"ok": True, "job": job, "status": self._public_status(state)}
return {
"ok": True,
"job": self._public_job(job),
"status": self._public_status(state),
}
def complete(self, job_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
payload = payload or {}
@@ -206,6 +235,7 @@ class TrainingCoordinator:
raise ValueError(f"training job not found: {job_id}")
if job.get("status") != "running" or not job.get("claimed_by"):
raise ValueError("training job is not claimed and running")
self._require_lease(job, payload)
success = bool(payload.get("success", payload.get("status") == "completed"))
if success and job.get("artifacts"):
promoted = self._validate_and_promote(job_id, job)
@@ -221,8 +251,13 @@ class TrainingCoordinator:
job["model_decision"] = (
"accepted" if payload["summary"]["accepted"] else "rejected"
)
job.pop("lease_token", None)
self._save_state(state)
return {"ok": True, "job": job, "status": self._public_status(state)}
return {
"ok": True,
"job": self._public_job(job),
"status": self._public_status(state),
}
def _validate_and_promote(self, job_id: str, job: dict[str, Any]) -> list[dict[str, Any]]:
ready_dir = self.upload_root / job_id / "ready"
@@ -327,11 +362,27 @@ class TrainingCoordinator:
"agent_recently_seen": recently_seen,
"agent_busy": agent_busy,
"worker": worker,
"active_job": active,
"latest_job": latest,
"active_job": self._public_job(active),
"latest_job": self._public_job(latest),
"pending_jobs": sum(1 for job in state.get("jobs", []) if job.get("status") == "pending"),
}
@staticmethod
def _public_job(job: dict[str, Any] | None) -> dict[str, Any] | None:
if job is None:
return None
public = dict(job)
public.pop("lease_token", None)
public.pop("uploads", None)
return public
@staticmethod
def _require_lease(job: dict[str, Any], payload: dict[str, Any]) -> None:
expected = str(job.get("lease_token") or "")
supplied = str(payload.get("lease_token") or "")
if not expected or not supplied or not hmac.compare_digest(expected, supplied):
raise ValueError("training job lease is invalid or expired")
def _active_job(self, state: dict[str, Any]) -> dict[str, Any] | None:
for job in reversed(state.get("jobs", [])):
if job.get("status") in {"pending", "running"}:
@@ -355,11 +406,30 @@ class TrainingCoordinator:
for job in state.get("jobs", []):
if job.get("status") != "running":
continue
claimed_at = _parse_time(str(job.get("claimed_at") or ""))
if claimed_at and now - claimed_at > RUNNING_TIMEOUT:
lease_updated_at = _parse_time(
str(job.get("updated_at") or job.get("claimed_at") or "")
)
if not lease_updated_at or now - lease_updated_at <= RUNNING_LEASE_TIMEOUT:
continue
job_id = str(job.get("id") or "")
if job_id:
_remove_tree(self.upload_root / job_id)
job.pop("lease_token", None)
job.pop("uploads", None)
attempts = int(job.get("attempts", 0))
if attempts < MAX_JOB_ATTEMPTS:
job["status"] = "pending"
job["phase"] = "queued"
job["progress_percent"] = 0
job["message"] = "training worker lease expired; queued for retry"
job["retry_queued_at"] = _now()
for key in ("claimed_at", "claimed_by", "worker", "updated_at"):
job.pop(key, None)
else:
job["status"] = "failed"
job["phase"] = "failed"
job["completed_at"] = _now()
job["message"] = "training worker timeout"
job["message"] = "training worker lease expired after maximum retries"
def _trim_jobs(self, state: dict[str, Any]) -> None:
jobs = state.get("jobs", [])