diff --git a/README.md b/README.md index c938102..8a08101 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,14 @@ powershell -ExecutionPolicy Bypass -File tools\run_torch_retrain.ps1 powershell -ExecutionPolicy Bypass -File tools\install_windows_torch_retrainer.ps1 ``` +Для удалённого запуска с телефона или с бота используется Windows training agent. Бот на `tb.kusoft.xyz` хранит очередь заданий, а Windows-машина сама подключается к интернету, забирает задания, обучает модель и загружает артефакты обратно: + +```powershell +powershell -ExecutionPolicy Bypass -File tools\install_windows_training_agent.ps1 -ApiAuth "login:password" -StartNow +``` + +Установщик регистрирует Scheduled Task `TradeBot Windows Training Agent` при входе в Windows и удаляет старые локальные retrain-задачи, чтобы обучение запускалось через очередь, а не двумя независимыми механизмами. + По умолчанию Windows-расписание переобучает PyTorch `LSTM/GRU` каждые 6 часов с `--limit 3000` на 12 spot-парах из `SYMBOLS`. Параметры можно переопределить через env: `TORCH_RETRAIN_SYMBOLS`, `TORCH_RETRAIN_LIMIT`, `TORCH_RETRAIN_LOOKBACKS`, `TORCH_RETRAIN_ARCHITECTURES`, `TORCH_RETRAIN_HIDDEN_SIZES`, `TORCH_RETRAIN_LAYERS`, `TORCH_RETRAIN_DROPOUTS`, `TORCH_RETRAIN_HORIZON`, `TORCH_RETRAIN_HORIZONS`, `TORCH_RETRAIN_CONTEXT_SYMBOLS`, `TORCH_RETRAIN_FEATURES`, `TORCH_RETRAIN_SEED`, `TORCH_RETRAIN_EPOCHS`, `TORCH_RETRAIN_PATIENCE`, `TORCH_RETRAIN_INTERVAL`, `TORCH_RETRAIN_ENV`. Если retrain запускается с `-DeployToPi`, после успешного guard он синхронизирует `runtime/lstm_forecaster.json`, `runtime/torch_retrain_guard.json` и `runtime/torch_threshold_calibration.json` на Raspberry Pi через SSH-ключ и перезапускает сервис `tradebot`. Отдельный запуск sync: diff --git a/android/TradeBotMonitor/README.md b/android/TradeBotMonitor/README.md index 45550b2..b2c5358 100644 --- a/android/TradeBotMonitor/README.md +++ b/android/TradeBotMonitor/README.md @@ -10,7 +10,7 @@ - Параметры Torch: edge, P(up), confidence, skill, quantiles, gate, причина решения. - Kelly/размер позиции: текущий размер, Kelly-цель, занятая экспозиция, остаток, множители edge/P(up)/skill. - Обзор equity/cash/exposure/PnL и последних решений. -- Удалённый запуск retrain через настраиваемый webhook. +- Удалённый запуск retrain через очередь заданий на боте и закреплённый Windows-компьютер обучения. - Расписание retrain на телефоне: Android отправляет команду по расписанию, но обучение идёт на Windows-машине. - Настройки API, токена команд, тёмной/светлой темы. - Live-чеклист: приложение показывает, готов ли сервер к реальной торговле, и не включает live одной опасной кнопкой. @@ -46,7 +46,7 @@ https://tb.kusoft.xyz ## Переобучение -Телефон не обучает модель локально. Вкладка `Обучение` отправляет POST на указанный webhook Windows-машины. Так телефон становится пультом запуска/расписания, а тяжёлый PyTorch retrain остаётся на нормальном компьютере. +Телефон не обучает модель локально. Вкладка `Обучение` ставит задание в очередь на `tb.kusoft.xyz`, а Windows-agent на закреплённой машине `DESKTOP-TMFDL0H` сам выходит в интернет, забирает задание, обучает модель и отправляет артефакты обратно боту. Так телефон становится пультом запуска/расписания, а тяжёлый PyTorch retrain остаётся на нормальном компьютере даже если он находится в другой сети. ## Live-торговля diff --git a/android/TradeBotMonitor/app/build.gradle.kts b/android/TradeBotMonitor/app/build.gradle.kts index e558e8d..252a746 100644 --- a/android/TradeBotMonitor/app/build.gradle.kts +++ b/android/TradeBotMonitor/app/build.gradle.kts @@ -10,7 +10,7 @@ android { applicationId = "xyz.kusoft.tradebotmonitor" minSdk = 26 targetSdk = 36 - versionCode = 5 - versionName = "0.2.2" + versionCode = 6 + versionName = "0.2.3" } } diff --git a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/AppPrefs.kt b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/AppPrefs.kt index aba2f56..6b4e875 100644 --- a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/AppPrefs.kt +++ b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/AppPrefs.kt @@ -23,10 +23,6 @@ class AppPrefs(context: Context) { get() = prefs.getString("command_token", "") ?: "" set(value) = prefs.edit().putString("command_token", value.trim()).apply() - var retrainWebhookUrl: String - get() = prefs.getString("retrain_webhook_url", "") ?: "" - set(value) = prefs.edit().putString("retrain_webhook_url", value.trim()).apply() - var selectedSymbol: String get() = prefs.getString("selected_symbol", "BTCUSDT") ?: "BTCUSDT" set(value) = prefs.edit().putString("selected_symbol", value).apply() diff --git a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt index 0979cf4..64c3da8 100644 --- a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt +++ b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/MainActivity.kt @@ -565,13 +565,20 @@ class MainActivity : Activity() { return section("Открытые позиции", box) } - private fun trainingComputerPanel(): View = + private fun trainingComputerPanel(retrain: JSONObject): View = LinearLayout(this).apply { + val coordination = retrain.optJSONObject("coordination") ?: JSONObject() + val activeJob = coordination.optJSONObject("active_job") + val agentOnline = coordination.optBoolean("agent_online", false) orientation = LinearLayout.VERTICAL addView(text("Компьютер обучения", 12f, Typeface.NORMAL, palette.muted)) addView(text(prefs.trainingComputerName, 18f, Typeface.BOLD, palette.green).top(dp(5))) addView(text(prefs.trainingComputerPath, 12f, Typeface.NORMAL, palette.muted).top(dp(4))) addView(keyValueLine("Статус", if (prefs.trainingComputerPinned) "закреплен" else "не закреплен", if (prefs.trainingComputerPinned) palette.green else palette.amber).top(dp(8))) + addView(keyValueLine("Связь агента", if (agentOnline) "онлайн" else "нет связи", if (agentOnline) palette.green else palette.amber).top(dp(4))) + if (activeJob != null) { + addView(keyValueLine("Задание", activeJob.optStringClean("status").ifBlank { "активно" }).top(dp(4))) + } addView(text("Переобучение Torch закреплено за этой Windows-машиной. Телефон только управляет запуском и расписанием, Pi только исполняет бота.", 12f, Typeface.NORMAL, palette.muted).top(dp(8))) addView(actionRow( "Закрепить эту машину" to { @@ -583,24 +590,17 @@ class MainActivity : Activity() { } private fun retrainSettingsBlock(retrain: JSONObject, backtest: JSONObject): View { - val webhookInput = input("URL запуска retrain на Windows", prefs.retrainWebhookUrl) return LinearLayout(this).apply { orientation = LinearLayout.VERTICAL addView(keyValueLine("Доступно", yesNo(retrain.optBoolean("available", false)))) addView(keyValueLine("Принята новая модель", yesNo(retrain.optBoolean("accepted", false))).top(dp(4))) - addView(trainingComputerPanel().top(dp(12))) + addView(trainingComputerPanel(retrain).top(dp(12))) addView(thinDivider().top(dp(12))) val replay = backtest.optJSONObject("full_replay") ?: JSONObject() addView(keyValueLine("Replay сделок", replay.optInt("trades", 0).toString()).top(dp(4))) addView(keyValueLine("Replay PnL", signedMoney(replay.optDouble("net_pnl", 0.0)), colorForSigned(replay.optDouble("net_pnl", 0.0))).top(dp(4))) - addView(webhookInput.top(dp(12))) addView(actionRow( - "Сохранить URL" to { - prefs.retrainWebhookUrl = webhookInput.text.toString() - toast("URL retrain сохранен") - }, "Запустить сейчас" to { - prefs.retrainWebhookUrl = webhookInput.text.toString() triggerRetrainNow() }, ).top(dp(10))) @@ -956,7 +956,7 @@ class MainActivity : Activity() { private fun triggerRetrainNow() { executor.execute { try { - val message = RetrainCommandClient(prefs.retrainWebhookUrl, prefs.commandToken).triggerNow() + val message = TradeBotApi(prefs.apiBaseUrl, prefs.commandToken).requestRetrain() mainHandler.post { toast(message.take(180)) } } catch (error: Exception) { mainHandler.post { toast(error.message ?: "retrain не запущен") } diff --git a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/RetrainScheduler.kt b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/RetrainScheduler.kt index 3a96c6e..a9f0fc3 100644 --- a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/RetrainScheduler.kt +++ b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/RetrainScheduler.kt @@ -42,8 +42,8 @@ class RetrainAlarmReceiver : BroadcastReceiver() { Executors.newSingleThreadExecutor().execute { try { val prefs = AppPrefs(context) - if (prefs.retrainScheduleEnabled && prefs.retrainWebhookUrl.isNotBlank()) { - RetrainCommandClient(prefs.retrainWebhookUrl, prefs.commandToken).triggerNow() + if (prefs.retrainScheduleEnabled) { + TradeBotApi(prefs.apiBaseUrl, prefs.commandToken).requestRetrain() } } finally { pending.finish() diff --git a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/TradeBotApi.kt b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/TradeBotApi.kt index 9e9ea76..1f53b88 100644 --- a/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/TradeBotApi.kt +++ b/android/TradeBotMonitor/app/src/main/java/xyz/kusoft/tradebotmonitor/TradeBotApi.kt @@ -52,11 +52,22 @@ class TradeBotApi( postJson("/api/control/stop") } + fun requestRetrain(): String { + val response = postJson("/api/training/retrain", "{\"source\":\"android\"}") + if (response.optBoolean("queued", false)) { + return "Задание переобучения отправлено на закрепленный компьютер" + } + return when (response.optStringClean("reason")) { + "active_job_exists" -> "Задание переобучения уже выполняется или ждет агента" + else -> "Задание переобучения принято ботом" + } + } + private fun getJson(path: String): JSONObject = request(path, "GET") - private fun postJson(path: String): JSONObject = request(path, "POST") + private fun postJson(path: String, body: String = "{}"): JSONObject = request(path, "POST", body) - private fun request(path: String, method: String): JSONObject { + private fun request(path: String, method: String, body: String = "{}"): JSONObject { val root = baseUrl.trim().trimEnd('/') val url = URL(root + path) val connection = (url.openConnection() as HttpURLConnection).apply { @@ -68,7 +79,7 @@ class TradeBotApi( if (method == "POST") { doOutput = true setRequestProperty("Content-Type", "application/json") - outputStream.use { stream -> stream.write("{}".toByteArray(StandardCharsets.UTF_8)) } + outputStream.use { stream -> stream.write(body.toByteArray(StandardCharsets.UTF_8)) } } } val code = connection.responseCode @@ -216,35 +227,6 @@ class TradeBotApi( } } -class RetrainCommandClient( - private val webhookUrl: String, - private val token: String, -) { - fun triggerNow(): String { - if (webhookUrl.isBlank()) { - throw IllegalStateException("URL запуска retrain не задан") - } - val connection = (URL(webhookUrl).openConnection() as HttpURLConnection).apply { - requestMethod = "POST" - connectTimeout = 6000 - readTimeout = 15000 - doOutput = true - setRequestProperty("Content-Type", "application/json") - setRequestProperty("Accept", "application/json,text/plain,*/*") - applyAuthHeaders(this, token) - outputStream.use { stream -> stream.write("{\"source\":\"android\"}".toByteArray(StandardCharsets.UTF_8)) } - } - val code = connection.responseCode - val stream = if (code in 200..299) connection.inputStream else connection.errorStream - val text = stream?.bufferedReader(StandardCharsets.UTF_8)?.use(BufferedReader::readText).orEmpty() - connection.disconnect() - if (code !in 200..299) { - throw IllegalStateException("Retrain HTTP $code: ${text.take(240)}") - } - return text.ifBlank { "Команда retrain отправлена" } - } -} - private fun applyAuthHeaders(connection: HttpURLConnection, token: String) { val value = token.trim() if (value.isBlank()) return diff --git a/crypto_spot_bot/dashboard.py b/crypto_spot_bot/dashboard.py index 8c8c4fa..bafa6f9 100644 --- a/crypto_spot_bot/dashboard.py +++ b/crypto_spot_bot/dashboard.py @@ -4,7 +4,7 @@ import json from contextlib import asynccontextmanager from typing import Any -from fastapi import FastAPI, Response +from fastapi import FastAPI, HTTPException, Response from fastapi.responses import JSONResponse, PlainTextResponse from crypto_spot_bot.analytics import analytics_snapshot @@ -19,6 +19,7 @@ from crypto_spot_bot.reconciliation import reconciliation_snapshot from crypto_spot_bot.storage import Storage from crypto_spot_bot.strategy import SpotStrategy from crypto_spot_bot.time_series import TimeSeriesForecaster +from crypto_spot_bot.training_coordination import TrainingCoordinator WEB_UI_REMOVED_MESSAGE = "Web UI removed. Use the Android TradeBot AI app and /api/* endpoints." @@ -42,6 +43,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: learner = TradeLearner(settings, storage) forecaster = TimeSeriesForecaster(settings) bot = CryptoSpotBot(settings, storage, market, broker, strategy, pattern_analyzer, learner, forecaster) + training = TrainingCoordinator(settings.time_series_lstm_model_path.parent) @asynccontextmanager async def lifespan(_: FastAPI): @@ -56,6 +58,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: app.state.storage = storage app.state.bot = bot app.state.market = market + app.state.training = training @app.get("/", response_class=PlainTextResponse, status_code=410) async def index() -> str: @@ -114,7 +117,39 @@ def create_app(settings: Settings | None = None) -> FastAPI: @app.get("/api/retrain") async def retrain() -> dict[str, Any]: - return _runtime_json(settings, "torch_retrain_guard.json") + data = _runtime_json(settings, "torch_retrain_guard.json") + data["coordination"] = training.status() + return data + + @app.get("/api/training/status") + async def training_status() -> dict[str, Any]: + return training.status() + + @app.post("/api/training/retrain") + async def training_retrain(payload: dict[str, Any] | None = None) -> dict[str, Any]: + return training.request_retrain(payload) + + @app.post("/api/training/heartbeat") + async def training_heartbeat(payload: dict[str, Any] | None = None) -> dict[str, Any]: + return training.heartbeat(payload) + + @app.post("/api/training/claim") + async def training_claim(payload: dict[str, Any] | None = None) -> dict[str, Any]: + return training.claim(payload) + + @app.post("/api/training/jobs/{job_id}/artifacts/chunk") + async def training_artifact_chunk(job_id: str, payload: dict[str, Any]) -> dict[str, Any]: + try: + return training.save_artifact_chunk(job_id, payload) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + @app.post("/api/training/jobs/{job_id}/complete") + async def training_complete(job_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: + try: + return training.complete(job_id, payload) + except ValueError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc @app.get("/api/config") async def config() -> dict[str, Any]: diff --git a/crypto_spot_bot/training_coordination.py b/crypto_spot_bot/training_coordination.py new file mode 100644 index 0000000..d705ab4 --- /dev/null +++ b/crypto_spot_bot/training_coordination.py @@ -0,0 +1,268 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import uuid +from datetime import UTC +from datetime import datetime +from datetime import timedelta +from pathlib import Path +from threading import Lock +from typing import Any + + +ALLOWED_TRAINING_ARTIFACTS = { + "lstm_forecaster.json", + "torch_retrain_guard.json", + "torch_threshold_calibration.json", +} +RUNNING_TIMEOUT = timedelta(hours=12) +ONLINE_WINDOW = timedelta(minutes=3) + + +class TrainingCoordinator: + def __init__(self, runtime_dir: Path) -> None: + self.runtime_dir = runtime_dir + self.state_path = runtime_dir / "training_coordination.json" + self.upload_root = runtime_dir / ".training_uploads" + self._lock = Lock() + + def status(self) -> dict[str, Any]: + with self._lock: + state = self._load_state() + self._expire_stale_jobs(state) + self._save_state(state) + return self._public_status(state) + + def request_retrain(self, payload: dict[str, Any] | None = None) -> dict[str, Any]: + payload = payload or {} + with self._lock: + state = self._load_state() + self._expire_stale_jobs(state) + 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)} + + now = _now() + job = { + "id": str(uuid.uuid4()), + "status": "pending", + "requested_at": now, + "requested_by": str(payload.get("source") or "api"), + "parameters": _safe_parameters(payload.get("parameters")), + "message": "", + "artifacts": [], + } + state.setdefault("jobs", []).append(job) + self._trim_jobs(state) + self._save_state(state) + return {"queued": True, "job": job, "status": self._public_status(state)} + + def heartbeat(self, payload: dict[str, Any] | None = None) -> dict[str, Any]: + payload = payload or {} + with self._lock: + state = self._load_state() + worker = self._worker_from_payload(payload) + state["worker"] = worker + self._save_state(state) + return {"ok": True, "worker": worker, "status": self._public_status(state)} + + def claim(self, payload: dict[str, Any] | None = None) -> dict[str, Any]: + payload = payload or {} + with self._lock: + state = self._load_state() + self._expire_stale_jobs(state) + worker = self._worker_from_payload(payload) + state["worker"] = worker + job = self._oldest_pending_job(state) + if job is None: + self._save_state(state) + return {"claimed": False, "job": None, "status": self._public_status(state)} + + now = _now() + job["status"] = "running" + job["claimed_at"] = now + job["claimed_by"] = worker["id"] + job["worker"] = worker + self._save_state(state) + return {"claimed": True, "job": job, "status": self._public_status(state)} + + def save_artifact_chunk(self, job_id: str, payload: dict[str, Any]) -> dict[str, Any]: + name = Path(str(payload.get("name") or "")).name + if name not in ALLOWED_TRAINING_ARTIFACTS: + raise ValueError(f"artifact is not allowed: {name}") + index = int(payload.get("index", -1)) + total = int(payload.get("total", 0)) + sha256 = str(payload.get("sha256") or "").strip().lower() + if index < 0 or total <= 0 or index >= total: + raise ValueError("invalid artifact chunk index") + if not sha256: + raise ValueError("artifact sha256 is required") + try: + chunk = base64.b64decode(str(payload.get("data_base64") or ""), validate=True) + except (ValueError, TypeError) as exc: + raise ValueError("invalid artifact chunk payload") from exc + + chunk_dir = self.upload_root / job_id / name + chunk_dir.mkdir(parents=True, exist_ok=True) + (chunk_dir / f"{index:06d}.part").write_bytes(chunk) + + if not all((chunk_dir / f"{part:06d}.part").is_file() for part in range(total)): + return {"complete": False, "received": index + 1, "total": total} + + target_tmp = self.runtime_dir / f".{name}.{job_id}.tmp" + digest = hashlib.sha256() + with target_tmp.open("wb") as output: + for part in range(total): + data = (chunk_dir / f"{part:06d}.part").read_bytes() + digest.update(data) + output.write(data) + if digest.hexdigest().lower() != sha256: + target_tmp.unlink(missing_ok=True) + raise ValueError("artifact sha256 mismatch") + + self.runtime_dir.mkdir(parents=True, exist_ok=True) + os.replace(target_tmp, self.runtime_dir / name) + _remove_tree(chunk_dir) + + with self._lock: + state = self._load_state() + job = self._job_by_id(state, job_id) + if job is not None: + artifacts = job.setdefault("artifacts", []) + artifacts = [item for item in artifacts if item.get("name") != name] + artifacts.append({"name": name, "sha256": sha256, "uploaded_at": _now()}) + job["artifacts"] = artifacts + self._save_state(state) + return {"complete": True, "name": name, "sha256": sha256} + + def complete(self, job_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: + payload = payload or {} + with self._lock: + state = self._load_state() + job = self._job_by_id(state, job_id) + if job is None: + raise ValueError(f"training job not found: {job_id}") + success = bool(payload.get("success", payload.get("status") == "completed")) + job["status"] = "completed" if success else "failed" + job["completed_at"] = _now() + job["message"] = str(payload.get("message") or "") + if isinstance(payload.get("summary"), dict): + job["summary"] = payload["summary"] + self._save_state(state) + return {"ok": True, "job": job, "status": self._public_status(state)} + + def _load_state(self) -> dict[str, Any]: + try: + data = json.loads(self.state_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + data = {} + if not isinstance(data, dict): + data = {} + data.setdefault("jobs", []) + return data + + def _save_state(self, state: dict[str, Any]) -> None: + self.runtime_dir.mkdir(parents=True, exist_ok=True) + tmp = self.state_path.with_suffix(".tmp") + tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + os.replace(tmp, self.state_path) + + def _worker_from_payload(self, payload: dict[str, Any]) -> dict[str, Any]: + return { + "id": str(payload.get("worker_id") or payload.get("id") or "windows-training-host"), + "name": str(payload.get("name") or "DESKTOP-TMFDL0H"), + "path": str(payload.get("path") or "C:\\Repos\\TradeBot"), + "version": str(payload.get("version") or "1"), + "last_seen_at": _now(), + } + + def _public_status(self, state: dict[str, Any]) -> dict[str, Any]: + worker = state.get("worker") if isinstance(state.get("worker"), dict) else {} + last_seen = _parse_time(str(worker.get("last_seen_at") or "")) + active = self._active_job(state) + latest = _latest_job(state) + return { + "available": True, + "agent_online": bool(last_seen and datetime.now(UTC) - last_seen <= ONLINE_WINDOW), + "worker": worker, + "active_job": active, + "latest_job": latest, + "pending_jobs": sum(1 for job in state.get("jobs", []) if job.get("status") == "pending"), + } + + 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"}: + return job + return None + + def _oldest_pending_job(self, state: dict[str, Any]) -> dict[str, Any] | None: + for job in state.get("jobs", []): + if job.get("status") == "pending": + return job + return None + + def _job_by_id(self, state: dict[str, Any], job_id: str) -> dict[str, Any] | None: + for job in state.get("jobs", []): + if job.get("id") == job_id: + return job + return None + + def _expire_stale_jobs(self, state: dict[str, Any]) -> None: + now = datetime.now(UTC) + 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: + job["status"] = "failed" + job["completed_at"] = _now() + job["message"] = "training worker timeout" + + def _trim_jobs(self, state: dict[str, Any]) -> None: + jobs = state.get("jobs", []) + if isinstance(jobs, list) and len(jobs) > 30: + state["jobs"] = jobs[-30:] + + +def _safe_parameters(value: Any) -> dict[str, Any]: + if not isinstance(value, dict): + return {} + allowed = {"symbols", "limit", "lookbacks", "architectures", "hidden_sizes", "layers", "dropouts", "epochs"} + return {key: value[key] for key in allowed if key in value} + + +def _latest_job(state: dict[str, Any]) -> dict[str, Any] | None: + jobs = state.get("jobs", []) + if not jobs: + return None + latest = jobs[-1] + return latest if isinstance(latest, dict) else None + + +def _now() -> str: + return datetime.now(UTC).isoformat(timespec="seconds") + + +def _parse_time(value: str) -> datetime | None: + if not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + + +def _remove_tree(path: Path) -> None: + if not path.exists(): + return + for child in path.iterdir(): + if child.is_dir(): + _remove_tree(child) + else: + child.unlink(missing_ok=True) + path.rmdir() diff --git a/tests/test_training_coordination.py b/tests/test_training_coordination.py new file mode 100644 index 0000000..8738cf7 --- /dev/null +++ b/tests/test_training_coordination.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import base64 +import hashlib + +from crypto_spot_bot.training_coordination import TrainingCoordinator + + +def test_training_coordinator_claims_and_completes_job(tmp_path) -> None: + coordinator = TrainingCoordinator(tmp_path) + + requested = coordinator.request_retrain({"source": "android"}) + job_id = requested["job"]["id"] + heartbeat = coordinator.heartbeat({"worker_id": "win-1", "name": "DESKTOP-TMFDL0H"}) + claimed = coordinator.claim({"worker_id": "win-1", "name": "DESKTOP-TMFDL0H"}) + + assert requested["queued"] is True + assert heartbeat["status"]["agent_online"] is True + assert claimed["claimed"] is True + assert claimed["job"]["id"] == job_id + assert coordinator.status()["active_job"]["status"] == "running" + + completed = coordinator.complete(job_id, {"success": True, "message": "ok"}) + + assert completed["job"]["status"] == "completed" + assert coordinator.status()["active_job"] is None + + +def test_training_coordinator_accepts_chunked_artifact_upload(tmp_path) -> None: + coordinator = TrainingCoordinator(tmp_path) + job = coordinator.request_retrain({"source": "test"})["job"] + payload = b'{"type":"pytorch_recurrent_forecaster","symbols":{}}\n' + sha256 = hashlib.sha256(payload).hexdigest() + first = payload[:20] + second = payload[20:] + + part_1 = coordinator.save_artifact_chunk( + job["id"], + { + "name": "lstm_forecaster.json", + "index": 0, + "total": 2, + "sha256": sha256, + "data_base64": base64.b64encode(first).decode("ascii"), + }, + ) + part_2 = coordinator.save_artifact_chunk( + job["id"], + { + "name": "lstm_forecaster.json", + "index": 1, + "total": 2, + "sha256": sha256, + "data_base64": base64.b64encode(second).decode("ascii"), + }, + ) + + assert part_1["complete"] is False + assert part_2["complete"] is True + assert (tmp_path / "lstm_forecaster.json").read_bytes() == payload + assert coordinator.status()["latest_job"]["artifacts"][0]["sha256"] == sha256 diff --git a/tools/install_windows_training_agent.ps1 b/tools/install_windows_training_agent.ps1 new file mode 100644 index 0000000..fc82c8b --- /dev/null +++ b/tools/install_windows_training_agent.ps1 @@ -0,0 +1,98 @@ +[CmdletBinding()] +param( + [string]$TaskName = "TradeBot Windows Training Agent", + [string]$ApiBaseUrl = "https://tb.kusoft.xyz", + [string]$ApiAuth = "", + [int]$PollSeconds = 60, + [string]$RepoRoot = "", + [switch]$StartNow, + [switch]$KeepLegacyRetrainer +) + +$ErrorActionPreference = "Stop" + +if (-not $RepoRoot) { + $RepoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path +} +$Agent = Join-Path $RepoRoot "tools\windows_training_agent.py" +if (-not (Test-Path $Agent)) { + throw "Windows training agent not found: $Agent" +} + +function Resolve-Python { + $venvPython = Join-Path $RepoRoot ".venv\Scripts\python.exe" + if (Test-Path $venvPython) { + return $venvPython + } + + $userPython = Join-Path $env:LOCALAPPDATA "Programs\TradeBotPython312\python.exe" + if (Test-Path $userPython) { + return $userPython + } + + foreach ($candidate in @("python.exe", "python")) { + $command = Get-Command $candidate -ErrorAction SilentlyContinue + if ($command) { + return $command.Source + } + } + throw "Python was not found. Create .venv or install Python 3.12." +} + +if ($ApiAuth) { + [Environment]::SetEnvironmentVariable("TRADEBOT_API_AUTH", $ApiAuth, "User") + $env:TRADEBOT_API_AUTH = $ApiAuth +} +[Environment]::SetEnvironmentVariable("TRADEBOT_API_BASE_URL", $ApiBaseUrl, "User") +[Environment]::SetEnvironmentVariable("TRADEBOT_TRAINING_WORKER_NAME", $env:COMPUTERNAME, "User") +$env:TRADEBOT_API_BASE_URL = $ApiBaseUrl +$env:TRADEBOT_TRAINING_WORKER_NAME = $env:COMPUTERNAME + +if (-not $KeepLegacyRetrainer) { + foreach ($legacyName in @("TradeBot PyTorch Forecaster Retrainer", "TradeBot LSTM Retrainer")) { + $legacyTask = Get-ScheduledTask -TaskName $legacyName -ErrorAction SilentlyContinue + if ($legacyTask) { + Unregister-ScheduledTask -TaskName $legacyName -Confirm:$false + Write-Host "Removed legacy scheduled task '$legacyName'." + } + } +} + +$python = Resolve-Python +$arguments = @( + "-u", + "`"$Agent`"", + "--repo-root", "`"$RepoRoot`"", + "--api-base-url", "`"$ApiBaseUrl`"", + "--poll-seconds", $PollSeconds.ToString() +) -join " " + +$action = New-ScheduledTaskAction -Execute $python -Argument $arguments -WorkingDirectory $RepoRoot +$trigger = New-ScheduledTaskTrigger -AtLogOn -User ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) +$principal = New-ScheduledTaskPrincipal ` + -UserId ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) ` + -LogonType Interactive ` + -RunLevel Limited +$settings = New-ScheduledTaskSettingsSet ` + -StartWhenAvailable ` + -MultipleInstances IgnoreNew ` + -AllowStartIfOnBatteries ` + -DontStopIfGoingOnBatteries ` + -ExecutionTimeLimit (New-TimeSpan -Days 30) + +Register-ScheduledTask ` + -TaskName $TaskName ` + -Action $action ` + -Trigger $trigger ` + -Principal $principal ` + -Settings $settings ` + -Description "Keeps the TradeBot Windows training agent online and polls the public bot API for retrain jobs." ` + -Force | Out-Null + +if ($StartNow) { + Start-ScheduledTask -TaskName $TaskName +} + +Write-Host "Registered scheduled task '$TaskName' for Windows logon." +Write-Host "Agent API: $ApiBaseUrl" +Write-Host "Agent script: $Agent" diff --git a/tools/windows_training_agent.py b/tools/windows_training_agent.py new file mode 100644 index 0000000..2469298 --- /dev/null +++ b/tools/windows_training_agent.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import platform +import subprocess +import sys +import time +from datetime import datetime +from pathlib import Path +from typing import Any +from urllib.error import HTTPError +from urllib.error import URLError +from urllib.request import Request +from urllib.request import urlopen + + +ARTIFACT_NAMES = ( + "lstm_forecaster.json", + "torch_retrain_guard.json", + "torch_threshold_calibration.json", +) + + +def main() -> None: + args = parse_args() + repo_root = Path(args.repo_root).resolve() + runtime_dir = repo_root / "runtime" + runtime_dir.mkdir(parents=True, exist_ok=True) + log_path = Path(args.log_file).resolve() if args.log_file else runtime_dir / "windows_training_agent.log" + + log(log_path, f"TradeBot Windows training agent started for {args.api_base_url}") + while True: + try: + poll_once(args, repo_root, runtime_dir, log_path) + except Exception as exc: # noqa: BLE001 - agent must keep running. + log(log_path, f"ERROR: {exc}") + if args.once: + break + time.sleep(max(5, args.poll_seconds)) + + +def poll_once(args: argparse.Namespace, repo_root: Path, runtime_dir: Path, log_path: Path) -> None: + worker = worker_payload(args, repo_root) + api_json(args, "/api/training/heartbeat", worker) + claim = api_json(args, "/api/training/claim", worker) + if not claim.get("claimed"): + return + job = claim.get("job") if isinstance(claim.get("job"), dict) else {} + job_id = str(job.get("id") or "") + if not job_id: + return + log(log_path, f"Claimed retrain job {job_id}") + success = False + message = "" + summary: dict[str, Any] = {} + try: + run_retrain(job, repo_root, log_path) + summary = read_json(runtime_dir / "torch_retrain_guard.json") + for name in ARTIFACT_NAMES: + path = runtime_dir / name + if path.is_file(): + upload_artifact(args, job_id, path, log_path) + success = True + message = "training completed" + log(log_path, f"Completed retrain job {job_id}") + except Exception as exc: # noqa: BLE001 - report failure to the bot. + message = str(exc) + log(log_path, f"Job {job_id} failed: {message}") + finally: + payload = {"success": success, "message": message, "summary": summary} + api_json(args, f"/api/training/jobs/{job_id}/complete", payload) + + +def run_retrain(job: dict[str, Any], repo_root: Path, log_path: Path) -> None: + script = repo_root / "tools" / "run_torch_retrain.ps1" + if not script.is_file(): + raise RuntimeError(f"retrain script not found: {script}") + cmd = [ + "powershell.exe", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + str(script), + ] + parameters = job.get("parameters") if isinstance(job.get("parameters"), dict) else {} + arg_map = { + "symbols": "-Symbols", + "limit": "-Limit", + "lookbacks": "-Lookbacks", + "architectures": "-Architectures", + "hidden_sizes": "-HiddenSizes", + "layers": "-Layers", + "dropouts": "-Dropouts", + "epochs": "-Epochs", + } + for key, ps_arg in arg_map.items(): + value = parameters.get(key) + if value not in (None, ""): + cmd.extend([ps_arg, str(value)]) + log(log_path, "Running retrain: " + " ".join(quote_for_log(part) for part in cmd)) + with subprocess.Popen( + cmd, + cwd=str(repo_root), + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + ) as process: + assert process.stdout is not None + for line in process.stdout: + log(log_path, line.rstrip()) + code = process.wait() + if code != 0: + raise RuntimeError(f"retrain failed with exit code {code}") + + +def upload_artifact(args: argparse.Namespace, job_id: str, path: Path, log_path: Path) -> None: + digest = hashlib.sha256(path.read_bytes()).hexdigest() + size = path.stat().st_size + chunk_size = max(64 * 1024, args.chunk_size) + total = max(1, (size + chunk_size - 1) // chunk_size) + log(log_path, f"Uploading {path.name}: {size} bytes, {total} chunks") + with path.open("rb") as source: + for index in range(total): + data = source.read(chunk_size) + payload = { + "name": path.name, + "index": index, + "total": total, + "sha256": digest, + "data_base64": base64.b64encode(data).decode("ascii"), + } + api_json(args, f"/api/training/jobs/{job_id}/artifacts/chunk", payload, timeout=120) + + +def api_json(args: argparse.Namespace, path: str, payload: dict[str, Any], timeout: int = 30) -> dict[str, Any]: + url = args.api_base_url.rstrip("/") + path + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + token = args.api_auth or os.environ.get("TRADEBOT_API_AUTH", "") + headers.update(auth_headers(token)) + request = Request(url, data=body, headers=headers, method="POST") + try: + with urlopen(request, timeout=timeout) as response: + text = response.read().decode("utf-8") + except HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"HTTP {exc.code} {path}: {detail[:300]}") from exc + except URLError as exc: + raise RuntimeError(f"network error {path}: {exc.reason}") from exc + return json.loads(text) if text.strip() else {} + + +def auth_headers(token: str) -> dict[str, str]: + value = token.strip() + if not value: + return {} + headers = {"X-TradeBot-Token": value} + if value.lower().startswith(("basic ", "bearer ")): + headers["Authorization"] = value + elif ":" in value: + encoded = base64.b64encode(value.encode("utf-8")).decode("ascii") + headers["Authorization"] = f"Basic {encoded}" + else: + headers["Authorization"] = f"Bearer {value}" + return headers + + +def worker_payload(args: argparse.Namespace, repo_root: Path) -> dict[str, Any]: + name = args.worker_name or platform.node() or "Windows training host" + return { + "worker_id": args.worker_id or f"{name}:{repo_root}", + "name": name, + "path": str(repo_root), + "version": "1", + } + + +def log(path: Path, message: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + stamp = datetime.now().astimezone().isoformat(timespec="seconds") + line = f"[{stamp}] {message}" + print(line, flush=True) + with path.open("a", encoding="utf-8") as handle: + handle.write(line + "\n") + + +def read_json(path: Path) -> dict[str, Any]: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return data if isinstance(data, dict) else {} + + +def quote_for_log(value: str) -> str: + return f'"{value}"' if " " in value else value + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Poll TradeBot for retrain jobs and execute them on Windows.") + parser.add_argument("--api-base-url", default=os.environ.get("TRADEBOT_API_BASE_URL", "https://tb.kusoft.xyz")) + parser.add_argument("--api-auth", default=os.environ.get("TRADEBOT_API_AUTH", "")) + parser.add_argument("--repo-root", default=str(Path(__file__).resolve().parents[1])) + parser.add_argument("--worker-id", default=os.environ.get("TRADEBOT_TRAINING_WORKER_ID", "")) + parser.add_argument("--worker-name", default=os.environ.get("TRADEBOT_TRAINING_WORKER_NAME", "")) + parser.add_argument("--poll-seconds", type=int, default=int(os.environ.get("TRADEBOT_TRAINING_POLL_SECONDS", "60"))) + parser.add_argument("--chunk-size", type=int, default=int(os.environ.get("TRADEBOT_TRAINING_CHUNK_SIZE", str(512 * 1024)))) + parser.add_argument("--log-file", default=os.environ.get("TRADEBOT_TRAINING_LOG", "")) + parser.add_argument("--once", action="store_true") + return parser.parse_args() + + +if __name__ == "__main__": + main()