diff --git a/.env.example b/.env.example index 6a83b51..0866cc0 100644 --- a/.env.example +++ b/.env.example @@ -3,13 +3,18 @@ HOST=127.0.0.1 PORT=8787 BYBIT_TESTNET=false +# Official regional public endpoints for this deployment. Before future live +# trading they must match the Bybit site where the API key was created. +BYBIT_REST_BASE_URL=https://api.bybit.kz +BYBIT_WEBSOCKET_URL=wss://stream.bybit.kz/v5/public/spot BYBIT_API_KEY= BYBIT_API_SECRET= STARTING_BALANCE_USDT=100 -AUTO_SELECT_SYMBOLS=false +AUTO_SELECT_SYMBOLS=true TOP_SYMBOLS_COUNT=12 -SYMBOLS=BTCUSDT,ETHUSDT,HYPEUSDT,SOLUSDT,XRPUSDT,XPLUSDT,WLDUSDT,MNTUSDT,HUSDT,XAUTUSDT,IPUSDT,AAVEUSDT +# Leave empty to discover the most liquid eligible USDT Spot pairs from Bybit. +SYMBOLS= STRATEGY_MODE=torch_forecast BASE_INTERVAL=60 diff --git a/Dockerfile b/Dockerfile index 4fbea3d..1e62195 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,12 +5,16 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ WORKDIR /app COPY requirements.txt /app/requirements.txt -RUN pip install --no-cache-dir --upgrade pip \ - && pip install --no-cache-dir -r /app/requirements.txt +RUN pip install --no-cache-dir --disable-pip-version-check --upgrade pip \ + && pip install --no-cache-dir --disable-pip-version-check -r /app/requirements.txt \ + && groupadd --gid 1000 tradebot \ + && useradd --uid 1000 --gid tradebot --home-dir /app --shell /usr/sbin/nologin tradebot -COPY crypto_spot_bot /app/crypto_spot_bot -COPY README.md /app/README.md -RUN mkdir -p /app/runtime +COPY --chown=1000:1000 crypto_spot_bot /app/crypto_spot_bot +COPY --chown=1000:1000 README.md /app/README.md +RUN mkdir -p /app/runtime && chown -R 1000:1000 /app/runtime EXPOSE 8787 +USER 1000:1000 +STOPSIGNAL SIGTERM CMD ["python", "-m", "crypto_spot_bot.main"] diff --git a/README.md b/README.md index 1730154..2f05732 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Spot-бот для демо-торговли криптовалютой на р ## Что реализовано - Реальные market data Bybit Spot: REST bootstrap и WebSocket-обновления. -- Фиксированный набор 12 USDT spot-пар для основной стратегии: `BTCUSDT`, `ETHUSDT`, `HYPEUSDT`, `SOLUSDT`, `XRPUSDT`, `XPLUSDT`, `WLDUSDT`, `MNTUSDT`, `HUSDT`, `XAUTUSDT`, `IPUSDT`, `AAVEUSDT`. +- Торговый universe автоматически строится из актуальных Bybit Spot-инструментов: выбираются до 12 ликвидных USDT-пар по `turnover24h`, исключаются stablecoin-to-stablecoin и leveraged-token пары; фиксированный список можно задать только явным `SYMBOLS`. - Paper trading с учетом cash, комиссий, проскальзывания, stop-loss, take-profit и trailing stop. - Spot-only логика: покупка базовой монеты за USDT и продажа обратно, без short и без плеча. - Live spot-ордеры явно отправляются без плеча: `category=spot`, `isLeverage=0`. @@ -18,13 +18,14 @@ Spot-бот для демо-торговли криптовалютой на р - DCA/мартингейл отключены: в режиме `trend_macd` брокер не разрешает вторую позицию по той же паре. - Grid, rebound, adaptive learning, Kelly sizing и time-series forecast выключены по умолчанию и не участвуют в принятии решений `trend_macd`. - Быстрый режим торговли: отдельный короткий интервал цикла, короткий cooldown после выхода и лимит новых входов в минуту; выходы по риску этим лимитом не блокируются. -- Веб-dashboard на русском: equity, cash, PnL, позиции, сделки, сигналы, события, свечные графики, переключатель быстрой торговли и индикаторы работы обучения. -- Android-монитор в `android/TradeBotMonitor`: русский мобильный интерфейс для просмотра 12 пар, свечей, Torch/Kelly параметров, расписания удалённого retrain и live-чеклиста. +- Защищённый JSON API: equity, cash, PnL, позиции, сделки, сигналы, события, свечи, управление paper-циклом и состоянием обучения. +- Android-монитор в `android/TradeBotMonitor`: русский мобильный интерфейс для динамического списка Bybit-пар, свечей, Torch/Kelly параметров, WorkManager-расписания удалённого retrain и live-чеклиста. - SQLite runtime-хранилище в `runtime/tradebot.sqlite3`. - Liveness `/api/health`, readiness `/api/ready`, объединенный mobile snapshot `/api/mobile/snapshot` и Prometheus-compatible `/metrics`. - Все приватные API endpoints требуют токен или подтвержденный reverse-proxy user header; health и metrics остаются доступными для локального мониторинга. -- Docker Compose для установки на Raspberry Pi 5 или другой Linux-хост. +- Hardened Docker Compose для установки на Dell/Linux: non-root user, read-only root filesystem, dropped capabilities, healthcheck и ротация container logs. - Live trading guard: live не стартует без `ENABLE_LIVE_TRADING=true`, `LIVE_TRADING_CONFIRM=I_ACCEPT_REAL_RISK` и Bybit API-ключей. +- Внешний production endpoint сохраняется на `https://tb.kusoft.xyz`; Caddy завершает TLS и проксирует API к контейнеру на loopback. ## Источники и принятые параметры @@ -34,6 +35,8 @@ Spot-бот для демо-торговли криптовалютой на р Популярность пар определяется через `/v5/market/tickers`, потому что Bybit Spot ticker возвращает `turnover24h`, `volume24h`, `bid1Price`, `ask1Price` и `lastPrice`: . +Для текущего paper/training-развёртывания используется официальный региональный endpoint `api.bybit.kz`; Bybit перечисляет его в Integration Guidance. `BYBIT_REST_BASE_URL` и `BYBIT_WEBSOCKET_URL` остаются явными настройками, потому что перед будущим live-режимом домен обязан соответствовать площадке выпуска API-ключа: . + Лучшие bid/ask берутся из `/v5/market/orderbook`; документация Bybit описывает `GET /v5/market/orderbook` с `category=spot`: . WebSocket-стакан использует topic `orderbook.{depth}.{symbol}`; Bybit документирует snapshot/delta-поведение и частоты push для Spot depth 1/50/200/1000: . @@ -45,7 +48,7 @@ Live market orders используют `/v5/order/create`; Bybit докумен - Investopedia перечисляет важные свойства algo trading software: real-time market data, low latency, configurability, backtesting, broker/exchange integration, fees/costs и APIs: . - Investopedia отдельно указывает, что automated trading systems задают правила entry/exit/money management, но требуют мониторинга и несут риск mechanical failures и over-optimization: . - QuantInsti описывает типовой путь разработки: стратегия, backtesting, paper trading, затем live trading, плюс GUI, order management и risk management: . -- Hochreiter и Schmidhuber описали LSTM как recurrent neural network architecture для последовательностей; обучение LSTM/GRU в проекте выполняется локально через PyTorch, а Raspberry Pi исполняет только экспортированные JSON-веса без PyTorch runtime: . +- Hochreiter и Schmidhuber описали LSTM как recurrent neural network architecture для последовательностей; обучение LSTM/GRU в проекте выполняется локально через PyTorch, а Dell исполняет только прошедшие quality gate экспортированные JSON-веса без PyTorch runtime: . Я не могу подтвердить, что эта стратегия будет прибыльной. Источники выше описывают технические свойства и риски автоматической торговли, но не гарантируют прибыль. @@ -59,17 +62,17 @@ Copy-Item .env.example .env python -m crypto_spot_bot.main ``` -Dashboard: +Liveness: -## Локальное обучение PyTorch LSTM/GRU +## Локальное обучение PyTorch LSTM -Обучение запускается на основной Windows-машине, а Raspberry Pi остается только для исполнения торгового цикла. PyTorch нужен только на машине обучения; в JSON экспортируются веса, а runtime на Raspberry Pi считает inference обычным Python-кодом: +Обучение запускается на основной Windows-машине, а Dell остается для исполнения торгового цикла. PyTorch нужен только на машине обучения; в JSON экспортируются веса, а runtime на Dell считает inference обычным Python-кодом: ```powershell .\.venv\Scripts\python.exe -m pip install torch --index-url https://download.pytorch.org/whl/cpu .\.venv\Scripts\python.exe tools\train_torch_recurrent_forecaster.py ` --limit 3000 ` - --architectures lstm,gru ` + --architectures lstm ` --lookbacks 64 ` --hidden-sizes 64,96 ` --layers 2 ` @@ -86,11 +89,10 @@ Dashboard: Файл из `TIME_SERIES_LSTM_MODEL_PATH` читается ботом автоматически, если `TIME_SERIES_FORECAST_ENABLED=true`. В стратегии `torch_forecast` экспортированная PyTorch LSTM/GRU модель является единственным направляющим сигналом для входа и forecast-выхода. Экспортированные модели появляются в dashboard как `PyTorch LSTM` или `PyTorch GRU`; старый легкий reservoir LSTM-кандидат и все встроенные не-torch прогнозы удалены. -Автопереобучение на Windows запускает PyTorch trainer, пишет лог в `runtime/torch_retrain.log` и защищается от параллельных запусков: +Локальный retrain на Windows запускает PyTorch trainer, пишет лог в `runtime/torch_retrain.log` и защищается от параллельных запусков: ```powershell 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-машина сама подключается к интернету, забирает задания, обучает модель и загружает артефакты обратно: @@ -99,21 +101,15 @@ powershell -ExecutionPolicy Bypass -File tools\install_windows_torch_retrainer.p powershell -ExecutionPolicy Bypass -File tools\install_windows_training_agent.ps1 -ApiAuth "" -StartNow ``` -Установщик сохраняет worker-токен через Windows DPAPI, удаляет его старую plaintext-копию из пользовательского окружения и включает постоянный запуск агента. С правами администратора используется Scheduled Task с watchdog; без повышения прав — штатный ярлык в пользовательской папке Startup. Старые локальные retrain-задачи удаляются, чтобы обучение запускалось через очередь, а не двумя независимыми механизмами. +Установщик сохраняет worker-токен через Windows DPAPI, удаляет его старую plaintext-копию из пользовательского окружения и включает постоянный запуск агента. С правами администратора используется Scheduled Task с watchdog; без повышения прав — штатный ярлык в пользовательской папке Startup. Сервер выдаёт каждой попытке 10-минутную возобновляемую lease; зависшая попытка автоматически возвращается в очередь, а устаревший процесс не может загрузить артефакты по старой lease. -По умолчанию Windows-agent обучает отдельную PyTorch `LSTM/GRU` для каждой пары на `6000` часовых свечах. Это не заставляет разнородные активы делить одну архитектуру и один набор recurrent-весов. Прогноз усредняется по seed `7/19`, модели сравниваются на validation-folds, а пороги калибруются отдельно для каждой пары. Ensemble guard выполняется пакетно на GPU, а экспорт не дублирует первый набор весов. Search space использует lookback `32/64/128`, hidden `64/96`, dropout `0.20`, AdamW learning rate `0.0007` и weight decay `0.0005`; untouched holdout и quality gate не ослабляются. Для диагностического pooled-запуска используется ключ `-Pooled`. Параметры можно переопределить через 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_ENSEMBLE_SEEDS`, `TORCH_RETRAIN_SELECTION_FOLDS`, `TORCH_RETRAIN_LEARNING_RATE`, `TORCH_RETRAIN_WEIGHT_DECAY`, `TORCH_RETRAIN_EPOCHS`, `TORCH_RETRAIN_PATIENCE`, `TORCH_RETRAIN_INTERVAL`, `TORCH_RETRAIN_ENV`. +По умолчанию Windows-agent обучает одну pooled PyTorch LSTM на динамическом наборе пар и `4000` часовых свечах на пару. Базовый профиль использует lookback `64`, hidden size `64`, два recurrent-слоя, dropout `0.20`, до `50` эпох, seed-ensemble `7/19`, три validation-fold и AdamW с learning rate `0.0007`/weight decay `0.0005`. Untouched holdout и quality gate не ослабляются. Параметр задания `pooled=false` включает независимые модели по парам; `architectures=gru` оставлен только как явная экспериментальная опция. Параметры можно переопределить через 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_ENSEMBLE_SEEDS`, `TORCH_RETRAIN_SELECTION_FOLDS`, `TORCH_RETRAIN_LEARNING_RATE`, `TORCH_RETRAIN_WEIGHT_DECAY`, `TORCH_RETRAIN_EPOCHS`, `TORCH_RETRAIN_PATIENCE`, `TORCH_RETRAIN_INTERVAL`, `TORCH_RETRAIN_ENV`. Loss и выбор гиперпараметров учитывают after-cost trading utility, ошибку ожидаемого чистого PnL, quantile-loss и focal BCE для события `TP before SL`, а не только MAE направления цены. В каждом walk-forward fold вероятность успеха калибруется Platt-моделью исключительно на train-части; затем на этой же train-части выбираются глобальные и per-symbol пороги, которые применяются к test-части. Для выбора порога требуется минимум 24 непересекающиеся сделки, а финальный quality gate по-прежнему требует не менее 30 OOS-сделок. Калибратор не имеет fallback на единичные сделки: если минимальная статистика не набрана, кандидат получает `calibration_insufficient` и не может пройти gate. Основной decision horizon — `12h`, дополнительные горизонты — `3/6/12/24`. Размеры обучающих барьеров берутся из `STOP_LOSS_PERCENT` и `TAKE_PROFIT_PERCENT`, а round-trip cost — из fee/slippage настроек. Threshold search оценивается тем же execution replay со stop-loss, take-profit, ATR trailing и forecast-exit, который используется в walk-forward. `holdout_skill` остаётся только в финальном отчёте и никогда не участвует в фильтрации входов или подборе порогов. -Если retrain запускается с `-DeployToPi`, после успешного guard он синхронизирует `runtime/lstm_forecaster.json`, `runtime/torch_retrain_guard.json` и `runtime/torch_threshold_calibration.json` на Raspberry Pi через SSH-ключ и перезапускает сервис `tradebot`. Отдельный запуск sync: - -```powershell -powershell -ExecutionPolicy Bypass -File tools\sync_torch_artifacts_to_pi.ps1 -RemoteHost 192.168.0.185 -RemoteUser sevenhill -RemoteRoot /mnt/data/tradebot -``` - -Внутри recurrent модели используются exportable attention pooling и LayerNorm. После recurrent-контекста добавлена нелинейная GELU-проекция и две отдельные экспортируемые головы: одна для ожидаемого PnL/quantiles, вторая для `P(TP before SL)`. Raspberry Pi по-прежнему исполняет модель из JSON без PyTorch runtime. +Внутри recurrent модели используются exportable attention pooling и LayerNorm. После recurrent-контекста добавлена нелинейная GELU-проекция и две отдельные экспортируемые головы: одна для ожидаемого PnL/quantiles, вторая для `P(TP before SL)`. Принятый bundle загружается агентом через защищённый API `tb.kusoft.xyz`, проходит серверную проверку SHA-256/guard/calibration и атомарно становится активным на Dell. ## Docker @@ -123,18 +119,20 @@ docker compose up -d --build docker compose logs -f tradebot ``` -Dashboard: `http://:8787/` +Локальная проверка: `http://127.0.0.1:8787/api/health`; внешний адрес: `https://tb.kusoft.xyz`. -Для Raspberry Pi 5 проект использует `python:3.12-slim`, без Node.js build step. Runtime-данные лежат в volume `./runtime:/app/runtime`; на внешнем диске можно разместить папку проекта или заменить volume на абсолютный путь внешнего диска. +На Dell проект использует `python:3.12-slim`, без Node.js build step. Runtime-данные лежат в bind mount `./runtime:/app/runtime`; корневая файловая система контейнера read-only, процесс работает как UID/GID 1000, а container logs ротируются по `10 MiB × 3`. ## Основные env-параметры ```env TRADING_MODE=paper STARTING_BALANCE_USDT=100 -AUTO_SELECT_SYMBOLS=false +BYBIT_REST_BASE_URL=https://api.bybit.kz +BYBIT_WEBSOCKET_URL=wss://stream.bybit.kz/v5/public/spot +AUTO_SELECT_SYMBOLS=true TOP_SYMBOLS_COUNT=12 -SYMBOLS=BTCUSDT,ETHUSDT,HYPEUSDT,SOLUSDT,XRPUSDT,XPLUSDT,WLDUSDT,MNTUSDT,HUSDT,XAUTUSDT,IPUSDT,AAVEUSDT +SYMBOLS= STRATEGY_MODE=torch_forecast BASE_INTERVAL=60 TREND_INTERVAL=D diff --git a/android/TradeBotMonitor/README.md b/android/TradeBotMonitor/README.md index 67506b0..4b364d4 100644 --- a/android/TradeBotMonitor/README.md +++ b/android/TradeBotMonitor/README.md @@ -11,7 +11,7 @@ - Kelly/размер позиции: текущий размер, Kelly-цель, занятая экспозиция, остаток, множители edge/P(up)/skill. - Обзор equity/cash/exposure/PnL и последних решений. - Удалённый запуск retrain через очередь заданий на боте и закреплённый Windows-компьютер обучения. -- Расписание retrain на телефоне: Android отправляет команду по расписанию, но обучение идёт на Windows-машине. +- Расписание retrain через Android WorkManager: команда отправляется только при наличии сети, а обучение идёт на Windows-машине. - Настройки API, токена команд, тёмной/светлой темы. - Live-чеклист: приложение показывает, готов ли сервер к реальной торговле, и не включает live одной опасной кнопкой. @@ -42,7 +42,7 @@ https://tb.kusoft.xyz ## Переобучение -Телефон не обучает модель локально. Вкладка `Обучение` ставит задание в очередь на `tb.kusoft.xyz`, а Windows-agent на закреплённой машине `SEVENHILL` (`G:\Repos\TradeBot`) сам выходит в интернет, забирает задание, обучает модель и отправляет артефакты обратно боту. Так телефон становится пультом запуска/расписания, а тяжёлый PyTorch retrain остаётся на нормальном компьютере даже если он находится в другой сети. +Телефон не обучает модель локально. Вкладка `Обучение` ставит задание в очередь на `tb.kusoft.xyz`, а Windows-agent на этой машине сам выходит в интернет, забирает задание, обучает модель и отправляет проверенный bundle обратно боту. Имя и путь активного worker приложение получает от сервера, без прошитого имени компьютера. ## Live-торговля diff --git a/android/TradeBotMonitor/app/build.gradle.kts b/android/TradeBotMonitor/app/build.gradle.kts index 07dcf64..977e794 100644 --- a/android/TradeBotMonitor/app/build.gradle.kts +++ b/android/TradeBotMonitor/app/build.gradle.kts @@ -4,13 +4,17 @@ plugins { android { namespace = "xyz.kusoft.tradebotmonitor" - compileSdk = 36 + compileSdk = 37 defaultConfig { applicationId = "xyz.kusoft.tradebotmonitor" minSdk = 26 - targetSdk = 36 - versionCode = 21 - versionName = "0.4.2" + targetSdk = 37 + versionCode = 22 + versionName = "0.5.0" } } + +dependencies { + implementation("androidx.work:work-runtime:2.11.2") +} diff --git a/android/TradeBotMonitor/app/src/main/AndroidManifest.xml b/android/TradeBotMonitor/app/src/main/AndroidManifest.xml index 8ab007f..9be22c1 100644 --- a/android/TradeBotMonitor/app/src/main/AndroidManifest.xml +++ b/android/TradeBotMonitor/app/src/main/AndroidManifest.xml @@ -2,12 +2,13 @@ - - - - - - - - - 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 4b6a6b7..c810aaa 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 @@ -18,12 +18,9 @@ class AppPrefs(context: Context) { } val trainingComputerName = prefs.getString("training_computer_name", null)?.trim() val trainingComputerPath = prefs.getString("training_computer_path", null)?.trim() - if ( - trainingComputerName.isNullOrBlank() || - trainingComputerName == LEGACY_TRAINING_COMPUTER_NAME || - trainingComputerPath.isNullOrBlank() || - trainingComputerPath == LEGACY_TRAINING_COMPUTER_PATH - ) { + val staleFallback = trainingComputerName in setOf("SEVENHILL", "DESKTOP-TMFDL0H") || + trainingComputerPath in setOf("G:\\Repos\\TradeBot", "C:\\Repos\\TradeBot") + if (trainingComputerName.isNullOrBlank() || trainingComputerPath.isNullOrBlank() || staleFallback) { pinDefaultTrainingComputer() } } @@ -141,10 +138,8 @@ class AppPrefs(context: Context) { private companion object { const val DEFAULT_API_BASE_URL = "https://tb.kusoft.xyz" const val LEGACY_PI_API_BASE_URL = "http://192.168.0.185:8787" - const val DEFAULT_TRAINING_COMPUTER_NAME = "SEVENHILL" - const val DEFAULT_TRAINING_COMPUTER_PATH = "G:\\Repos\\TradeBot" - const val LEGACY_TRAINING_COMPUTER_NAME = "DESKTOP-TMFDL0H" - const val LEGACY_TRAINING_COMPUTER_PATH = "C:\\Repos\\TradeBot" + const val DEFAULT_TRAINING_COMPUTER_NAME = "Ожидание Windows-agent" + const val DEFAULT_TRAINING_COMPUTER_PATH = "Имя и путь поступят от сервера" const val TOKEN_KEY_ALIAS = "tradebot_api_auth_v1" } } 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 60d0c83..1d4f4a3 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 @@ -1512,12 +1512,15 @@ class MainActivity : Activity() { private fun rankedMarkets(data: BotSnapshot): List = orderedMarkets(data.markets).sortedWith( compareByDescending { marketRankScore(it, data.signalsBySymbol[it.symbol]) } - .thenBy { fixedSymbolIndex(it.symbol) } + .thenByDescending { it.ticker?.turnover24h ?: 0.0 } .thenBy { it.symbol }, ) private fun orderedMarkets(markets: List): List = - markets.sortedWith(compareBy({ fixedSymbolIndex(it.symbol) }, { it.symbol })) + markets.sortedWith( + compareByDescending { it.ticker?.turnover24h ?: 0.0 } + .thenBy { it.symbol }, + ) private fun marketRankScore(market: MarketItem, signal: SignalData?): Double { val actionScore = when (normalizedAction(signal?.action)) { @@ -1615,11 +1618,6 @@ class MainActivity : Activity() { } } - private fun fixedSymbolIndex(symbol: String): Int { - val index = FIXED_SYMBOLS.indexOf(symbol.uppercase(Locale.US)) - return if (index >= 0) index else FIXED_SYMBOLS.size + 1 - } - private fun trainingStatusSignature(retrain: JSONObject): String = (retrain.optJSONObject("coordination") ?: JSONObject()).toString() @@ -1895,20 +1893,4 @@ class MainActivity : Activity() { Toast.makeText(this, message, Toast.LENGTH_LONG).show() } - private companion object { - val FIXED_SYMBOLS = listOf( - "BTCUSDT", - "ETHUSDT", - "HYPEUSDT", - "SOLUSDT", - "XRPUSDT", - "XPLUSDT", - "WLDUSDT", - "MNTUSDT", - "HUSDT", - "XAUTUSDT", - "IPUSDT", - "AAVEUSDT", - ) - } } 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 1309409..a6e853b 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 @@ -1,65 +1,55 @@ package xyz.kusoft.tradebotmonitor -import android.app.AlarmManager -import android.app.PendingIntent -import android.content.BroadcastReceiver import android.content.Context -import android.content.Intent -import java.util.concurrent.Executors +import androidx.work.Constraints +import androidx.work.ExistingPeriodicWorkPolicy +import androidx.work.NetworkType +import androidx.work.PeriodicWorkRequest +import androidx.work.WorkManager +import androidx.work.Worker +import androidx.work.WorkerParameters +import java.util.concurrent.TimeUnit object RetrainScheduler { - private const val ACTION_RETRAIN = "xyz.kusoft.tradebotmonitor.RETRAIN" - private const val REQUEST_CODE = 6406 + private const val UNIQUE_WORK_NAME = "tradebot-periodic-retrain" fun schedule(context: Context, hours: Int) { - val interval = hours.coerceAtLeast(1) * 60L * 60L * 1000L - val manager = context.getSystemService(Context.ALARM_SERVICE) as AlarmManager - manager.setInexactRepeating( - AlarmManager.RTC_WAKEUP, - System.currentTimeMillis() + interval, - interval, - pendingIntent(context), + val constraints = Constraints.Builder() + .setRequiredNetworkType(NetworkType.CONNECTED) + .build() + val request = PeriodicWorkRequest.Builder( + RetrainWorker::class.java, + hours.coerceAtLeast(1).toLong(), + TimeUnit.HOURS, + ) + .setConstraints(constraints) + .build() + WorkManager.getInstance(context).enqueueUniquePeriodicWork( + UNIQUE_WORK_NAME, + ExistingPeriodicWorkPolicy.UPDATE, + request, ) } fun cancel(context: Context) { - val manager = context.getSystemService(Context.ALARM_SERVICE) as AlarmManager - manager.cancel(pendingIntent(context)) + WorkManager.getInstance(context).cancelUniqueWork(UNIQUE_WORK_NAME) } - - private fun pendingIntent(context: Context): PendingIntent = - PendingIntent.getBroadcast( - context, - REQUEST_CODE, - Intent(context, RetrainAlarmReceiver::class.java).setAction(ACTION_RETRAIN), - PendingIntent.FLAG_UPDATE_CURRENT or PendingIntent.FLAG_IMMUTABLE, - ) } -class RetrainAlarmReceiver : BroadcastReceiver() { - override fun onReceive(context: Context, intent: Intent) { - val pending = goAsync() - val executor = Executors.newSingleThreadExecutor() - executor.execute { - try { - val prefs = AppPrefs(context) - if (prefs.retrainScheduleEnabled) { - TradeBotApi(prefs.apiBaseUrl, prefs.commandToken).requestRetrain() - } - } finally { - pending.finish() - executor.shutdown() - } +class RetrainWorker( + context: Context, + parameters: WorkerParameters, +) : Worker(context, parameters) { + override fun doWork(): Result { + val prefs = AppPrefs(applicationContext) + if (!prefs.retrainScheduleEnabled || prefs.commandToken.isBlank()) { + return Result.success() } - } -} - -class BootReceiver : BroadcastReceiver() { - override fun onReceive(context: Context, intent: Intent) { - if (intent.action != Intent.ACTION_BOOT_COMPLETED) return - val prefs = AppPrefs(context) - if (prefs.retrainScheduleEnabled) { - RetrainScheduler.schedule(context, prefs.retrainIntervalHours) + return try { + TradeBotApi(prefs.apiBaseUrl, prefs.commandToken).requestRetrain() + Result.success() + } catch (_: Exception) { + Result.retry() } } } 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 0eea0e7..dd0e115 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 @@ -183,7 +183,7 @@ class TradeBotApi( qualityScore = quality.optDouble("score", 0.0), ) } - return output.sortedBy { it.symbol } + return output } private fun parseTicker(row: JSONObject): TickerData = diff --git a/android/TradeBotMonitor/app/src/main/res/xml/backup_rules.xml b/android/TradeBotMonitor/app/src/main/res/xml/backup_rules.xml new file mode 100644 index 0000000..ce0b324 --- /dev/null +++ b/android/TradeBotMonitor/app/src/main/res/xml/backup_rules.xml @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/android/TradeBotMonitor/app/src/main/res/xml/data_extraction_rules.xml b/android/TradeBotMonitor/app/src/main/res/xml/data_extraction_rules.xml new file mode 100644 index 0000000..491c90f --- /dev/null +++ b/android/TradeBotMonitor/app/src/main/res/xml/data_extraction_rules.xml @@ -0,0 +1,17 @@ + + + + + + + + + + + + + + + + + diff --git a/android/TradeBotMonitor/app/src/main/res/xml/network_security_config.xml b/android/TradeBotMonitor/app/src/main/res/xml/network_security_config.xml new file mode 100644 index 0000000..683208f --- /dev/null +++ b/android/TradeBotMonitor/app/src/main/res/xml/network_security_config.xml @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/android/TradeBotMonitor/design/option2-ai-console-mockups.svg b/android/TradeBotMonitor/design/option2-ai-console-mockups.svg index c9999ca..8f64ae5 100644 --- a/android/TradeBotMonitor/design/option2-ai-console-mockups.svg +++ b/android/TradeBotMonitor/design/option2-ai-console-mockups.svg @@ -142,7 +142,7 @@ 91% Рынки - 12 фиксированных spot-пар + Динамические Bybit spot-пары Поиск пары или сигнала diff --git a/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.properties b/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.properties index c61a118..193c941 100644 --- a/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.properties +++ b/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.properties @@ -1,7 +1,7 @@ distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists distributionUrl=https\://services.gradle.org/distributions/gradle-9.4.1-bin.zip -networkTimeout=10000 +networkTimeout=60000 validateDistributionUrl=true zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists diff --git a/crypto_spot_bot/__init__.py b/crypto_spot_bot/__init__.py index 45b9c54..400bd7d 100644 --- a/crypto_spot_bot/__init__.py +++ b/crypto_spot_bot/__init__.py @@ -1,3 +1,3 @@ """Crypto spot trading bot package.""" -__version__ = "0.1.0" +__version__ = "1.0.0" diff --git a/crypto_spot_bot/config.py b/crypto_spot_bot/config.py index 5737e92..d43b3bb 100644 --- a/crypto_spot_bot/config.py +++ b/crypto_spot_bot/config.py @@ -169,17 +169,23 @@ class Settings: hold_signal_sample_seconds: int = 60 storage_retention_days: int = 30 storage_prune_interval_seconds: int = 3600 + bybit_rest_base_url_override: str = "" + bybit_websocket_url_override: str = "" @property def rest_base_url(self) -> str: - return "https://api-testnet.bybit.com" if self.bybit_testnet else "https://api.bybit.com" + if self.bybit_rest_base_url_override: + return self.bybit_rest_base_url_override.rstrip("/") + return "https://api-testnet.bybit.com" if self.bybit_testnet else "https://api.bybit.kz" @property def websocket_url(self) -> str: + if self.bybit_websocket_url_override: + return self.bybit_websocket_url_override return ( "wss://stream-testnet.bybit.com/v5/public/spot" if self.bybit_testnet - else "wss://stream.bybit.com/v5/public/spot" + else "wss://stream.bybit.kz/v5/public/spot" ) @property @@ -220,7 +226,7 @@ def load_settings(env_file: str | Path | None = None) -> Settings: strategy_mode = os.getenv("STRATEGY_MODE", "torch_forecast").strip().lower() if strategy_mode not in STRATEGY_MODES: raise ValueError("STRATEGY_MODE must be legacy, trend_macd or torch_forecast") - auto_select_symbols = _bool_env("AUTO_SELECT_SYMBOLS", False) + auto_select_symbols = _bool_env("AUTO_SELECT_SYMBOLS", True) top_symbols_count = _int_env("TOP_SYMBOLS_COUNT", len(FIXED_SPOT_SYMBOLS)) requested_symbols = _symbols_env("SYMBOLS") symbols = requested_symbols if requested_symbols else (() if auto_select_symbols else FIXED_SPOT_SYMBOLS) @@ -343,6 +349,11 @@ def load_settings(env_file: str | Path | None = None) -> Settings: hold_signal_sample_seconds=_int_env("HOLD_SIGNAL_SAMPLE_SECONDS", 60), storage_retention_days=_int_env("STORAGE_RETENTION_DAYS", 30), storage_prune_interval_seconds=_int_env("STORAGE_PRUNE_INTERVAL_SECONDS", 3600), + bybit_rest_base_url_override=os.getenv( + "BYBIT_REST_BASE_URL", + "" if _bool_env("BYBIT_TESTNET", False) else "https://api.bybit.kz", + ).strip(), + bybit_websocket_url_override=os.getenv("BYBIT_WEBSOCKET_URL", "").strip(), ) _validate_settings(settings) if settings.trading_mode == "live" and not settings.live_ready: diff --git a/crypto_spot_bot/dashboard.py b/crypto_spot_bot/dashboard.py index 7d9e4b7..a34e844 100644 --- a/crypto_spot_bot/dashboard.py +++ b/crypto_spot_bot/dashboard.py @@ -12,6 +12,7 @@ from fastapi.responses import JSONResponse, PlainTextResponse from crypto_spot_bot.analytics import analytics_snapshot from crypto_spot_bot.auth import ApiAuthorizer from crypto_spot_bot.bot import CryptoSpotBot +from crypto_spot_bot import __version__ from crypto_spot_bot.bybit import BybitClient from crypto_spot_bot.config import Settings, load_settings, update_env_value from crypto_spot_bot.execution import LiveBroker, PaperBroker @@ -58,7 +59,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: finally: await bot.stop() - app = FastAPI(title="Крипто спот-бот", lifespan=lifespan) + app = FastAPI(title="Крипто спот-бот", version=__version__, lifespan=lifespan) app.state.settings = settings app.state.storage = storage app.state.bot = bot @@ -76,6 +77,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: "running": bot.running, "mode": settings.trading_mode, "auth_configured": authorizer.configured(), + "version": __version__, } @app.get("/api/ready") diff --git a/crypto_spot_bot/training_coordination.py b/crypto_spot_bot/training_coordination.py index 9c487c4..fa2e810 100644 --- a/crypto_spot_bot/training_coordination.py +++ b/crypto_spot_bot/training_coordination.py @@ -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", []) diff --git a/docker-compose.yml b/docker-compose.yml index 261d872..6e51ea0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,9 +7,9 @@ services: environment: HOST: 0.0.0.0 PYTHONDONTWRITEBYTECODE: "1" - user: "1000:1000" init: true read_only: true + pids_limit: 128 cap_drop: - ALL security_opt: @@ -27,4 +27,10 @@ services: timeout: 10s retries: 3 start_period: 30s + stop_grace_period: 30s + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" restart: unless-stopped diff --git a/requirements-dev.txt b/requirements-dev.txt index 44dca33..1441c92 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,2 +1,2 @@ -r requirements.txt -pytest==8.4.2 +pytest==9.1.1 diff --git a/requirements.txt b/requirements.txt index fb2257d..eba9f2c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -fastapi==0.115.6 -uvicorn[standard]==0.34.0 -requests==2.32.3 -websockets==14.1 +fastapi==0.139.0 +uvicorn[standard]==0.51.0 +requests==2.34.2 +websockets==16.1 diff --git a/tests/test_config.py b/tests/test_config.py index b2edff0..810f050 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -78,7 +78,7 @@ def test_llm_advisor_is_disabled_by_default(tmp_path, monkeypatch) -> None: assert settings.llm_advisor_enabled is False -def test_default_symbols_are_fixed_trend_pairs(tmp_path, monkeypatch) -> None: +def test_default_symbols_are_discovered_from_bybit(tmp_path, monkeypatch) -> None: monkeypatch.delenv("AUTO_SELECT_SYMBOLS", raising=False) monkeypatch.delenv("TOP_SYMBOLS_COUNT", raising=False) monkeypatch.delenv("SYMBOLS", raising=False) @@ -89,9 +89,9 @@ def test_default_symbols_are_fixed_trend_pairs(tmp_path, monkeypatch) -> None: settings = load_settings(env_file) - assert settings.auto_select_symbols is False + assert settings.auto_select_symbols is True assert settings.top_symbols_count == len(FIXED_SPOT_SYMBOLS) - assert settings.symbols == FIXED_SPOT_SYMBOLS + assert settings.symbols == () assert settings.strategy_mode == "torch_forecast" assert settings.base_interval == "60" assert settings.trend_interval == "D" diff --git a/tests/test_training_coordination.py b/tests/test_training_coordination.py index 486b312..279ffe7 100644 --- a/tests/test_training_coordination.py +++ b/tests/test_training_coordination.py @@ -3,6 +3,7 @@ from __future__ import annotations import base64 import hashlib import json +from datetime import UTC, datetime, timedelta import pytest @@ -16,6 +17,7 @@ def test_training_coordinator_claims_and_completes_job(tmp_path) -> None: 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"}) + lease_token = claimed["lease_token"] assert requested["queued"] is True assert heartbeat["status"]["agent_online"] is True @@ -25,14 +27,23 @@ def test_training_coordinator_claims_and_completes_job(tmp_path) -> None: progress = coordinator.progress( job_id, - {"status": "running", "phase": "training", "progress_percent": 42, "message": "epoch 1"}, + { + "status": "running", + "phase": "training", + "progress_percent": 42, + "message": "epoch 1", + "lease_token": lease_token, + }, ) assert progress["job"]["phase"] == "training" assert progress["job"]["progress_percent"] == 42 assert coordinator.status()["active_job"]["message"] == "epoch 1" - completed = coordinator.complete(job_id, {"success": True, "message": "ok"}) + completed = coordinator.complete( + job_id, + {"success": True, "message": "ok", "lease_token": lease_token}, + ) assert completed["job"]["status"] == "completed" assert coordinator.status()["active_job"] is None @@ -104,7 +115,7 @@ def test_training_coordinator_reports_worker_identity_from_heartbeat(tmp_path) - def test_training_coordinator_records_rejected_candidate_as_completed_training(tmp_path) -> None: coordinator = TrainingCoordinator(tmp_path) job = coordinator.request_retrain({"source": "android"})["job"] - coordinator.claim({"worker_id": "worker-1"}) + lease_token = coordinator.claim({"worker_id": "worker-1"})["lease_token"] completed = coordinator.complete( job["id"], @@ -112,6 +123,7 @@ def test_training_coordinator_records_rejected_candidate_as_completed_training(t "success": True, "message": "training completed; candidate rejected by quality gate", "summary": {"accepted": False, "reason": "candidate_failed_honest_validation"}, + "lease_token": lease_token, }, ) @@ -124,7 +136,7 @@ def test_training_coordinator_records_rejected_candidate_as_completed_training(t def test_training_coordinator_accepts_chunked_artifact_upload(tmp_path) -> None: coordinator = TrainingCoordinator(tmp_path) job = coordinator.request_retrain({"source": "test"})["job"] - coordinator.claim({"worker_id": "test-worker"}) + lease_token = coordinator.claim({"worker_id": "test-worker"})["lease_token"] payload = b'{"type":"pytorch_recurrent_forecaster","symbols":{}}\n' sha256 = hashlib.sha256(payload).hexdigest() first = payload[:20] @@ -138,6 +150,7 @@ def test_training_coordinator_accepts_chunked_artifact_upload(tmp_path) -> None: "total": 2, "sha256": sha256, "data_base64": base64.b64encode(first).decode("ascii"), + "lease_token": lease_token, }, ) part_2 = coordinator.save_artifact_chunk( @@ -148,6 +161,7 @@ def test_training_coordinator_accepts_chunked_artifact_upload(tmp_path) -> None: "total": 2, "sha256": sha256, "data_base64": base64.b64encode(second).decode("ascii"), + "lease_token": lease_token, }, ) @@ -199,6 +213,35 @@ def test_running_claimed_job_keeps_agent_online_when_heartbeat_is_stale(tmp_path assert status["agent_online"] is True +def test_stale_training_lease_is_requeued_and_old_lease_is_rejected(tmp_path) -> None: + coordinator = TrainingCoordinator(tmp_path) + job = coordinator.request_retrain({"source": "android"})["job"] + first_claim = coordinator.claim({"worker_id": "worker-1"}) + + state_path = tmp_path / "training_coordination.json" + state = json.loads(state_path.read_text(encoding="utf-8")) + state["jobs"][0]["updated_at"] = ( + datetime.now(UTC) - timedelta(minutes=11) + ).isoformat() + state_path.write_text(json.dumps(state), encoding="utf-8") + + second_claim = coordinator.claim({"worker_id": "worker-2"}) + + assert second_claim["claimed"] is True + assert second_claim["job"]["id"] == job["id"] + assert second_claim["job"]["attempts"] == 2 + assert second_claim["lease_token"] != first_claim["lease_token"] + with pytest.raises(ValueError, match="lease"): + coordinator.progress( + job["id"], + { + "phase": "training", + "progress_percent": 10, + "lease_token": first_claim["lease_token"], + }, + ) + + def test_training_upload_rejects_unknown_job(tmp_path) -> None: coordinator = TrainingCoordinator(tmp_path) payload = b"{}" @@ -219,7 +262,7 @@ def test_training_upload_rejects_unknown_job(tmp_path) -> None: def test_training_bundle_promotes_only_after_successful_guard(tmp_path) -> None: coordinator = TrainingCoordinator(tmp_path) job = coordinator.request_retrain({"source": "test"})["job"] - coordinator.claim({"worker_id": "worker-1"}) + lease_token = coordinator.claim({"worker_id": "worker-1"})["lease_token"] model = { "type": "pytorch_recurrent_forecaster", "symbols": { @@ -260,10 +303,14 @@ def test_training_bundle_promotes_only_after_successful_guard(tmp_path) -> None: "total": 1, "sha256": hashlib.sha256(payload).hexdigest(), "data_base64": base64.b64encode(payload).decode("ascii"), + "lease_token": lease_token, }, ) - completed = coordinator.complete(job["id"], {"success": True}) + completed = coordinator.complete( + job["id"], + {"success": True, "lease_token": lease_token}, + ) assert completed["job"]["status"] == "completed" assert json.loads((tmp_path / "lstm_forecaster.json").read_text())["symbols"]["BTCUSDT"] diff --git a/tools/install_windows_torch_retrainer.ps1 b/tools/install_windows_torch_retrainer.ps1 deleted file mode 100644 index e712e61..0000000 --- a/tools/install_windows_torch_retrainer.ps1 +++ /dev/null @@ -1,94 +0,0 @@ -[CmdletBinding()] -param( - [string]$TaskName = "TradeBot PyTorch Forecaster Retrainer", - [int]$EveryHours = 6, - [string]$Symbols = "", - [int]$Limit = 3000, - [int]$Horizon = 0, - [string]$Horizons = "", - [string]$Features = "", - [string]$ContextSymbols = "", - [int]$FirstRunMinutes = 0, - [switch]$DeployToPi, - [string]$PiHost = "192.168.0.185", - [string]$PiUser = "sevenhill", - [string]$PiRoot = "/mnt/data/tradebot", - [string]$PiSshKeyPath = "" -) - -$ErrorActionPreference = "Stop" - -$RepoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path -$Runner = Join-Path $RepoRoot "tools\run_torch_retrain.ps1" -if (-not (Test-Path $Runner)) { - throw "Runner not found: $Runner" -} - -$LegacyTaskName = "TradeBot LSTM Retrainer" -if ($TaskName -ne $LegacyTaskName) { - $legacyTask = Get-ScheduledTask -TaskName $LegacyTaskName -ErrorAction SilentlyContinue - if ($legacyTask) { - Unregister-ScheduledTask -TaskName $LegacyTaskName -Confirm:$false - } -} - -$actionArgs = "-NoProfile -ExecutionPolicy Bypass -File `"$Runner`"" -if ($Symbols) { - $actionArgs += " -Symbols `"$Symbols`"" -} -if ($Limit -gt 0) { - $actionArgs += " -Limit $Limit" -} -if ($Horizon -gt 0) { - $actionArgs += " -Horizon $Horizon" -} -if ($Horizons) { - $actionArgs += " -Horizons `"$Horizons`"" -} -if ($Features) { - $actionArgs += " -Features `"$Features`"" -} -if ($ContextSymbols) { - $actionArgs += " -ContextSymbols `"$ContextSymbols`"" -} -if ($DeployToPi) { - $actionArgs += " -DeployToPi" -} -if ($PiHost) { - $actionArgs += " -PiHost `"$PiHost`"" -} -if ($PiUser) { - $actionArgs += " -PiUser `"$PiUser`"" -} -if ($PiRoot) { - $actionArgs += " -PiRoot `"$PiRoot`"" -} -if ($PiSshKeyPath) { - $actionArgs += " -PiSshKeyPath `"$PiSshKeyPath`"" -} -$action = New-ScheduledTaskAction -Execute "powershell.exe" -Argument $actionArgs -WorkingDirectory $RepoRoot -$trigger = New-ScheduledTaskTrigger ` - -Once ` - -At (Get-Date).AddMinutes($(if ($FirstRunMinutes -gt 0) { $FirstRunMinutes } else { $EveryHours * 60 })) ` - -RepetitionInterval (New-TimeSpan -Hours $EveryHours) ` - -RepetitionDuration (New-TimeSpan -Days 3650) -$principal = New-ScheduledTaskPrincipal ` - -UserId ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) ` - -LogonType Interactive ` - -RunLevel Limited -$settings = New-ScheduledTaskSettingsSet ` - -StartWhenAvailable ` - -MultipleInstances IgnoreNew ` - -AllowStartIfOnBatteries ` - -DontStopIfGoingOnBatteries - -Register-ScheduledTask ` - -TaskName $TaskName ` - -Action $action ` - -Trigger $trigger ` - -Principal $principal ` - -Settings $settings ` - -Description "Retrains TradeBot PyTorch recurrent forecast parameters every $EveryHours hours." ` - -Force | Out-Null - -Write-Host "Registered scheduled task '$TaskName' every $EveryHours hours." diff --git a/tools/run_retrain_until_replay8.ps1 b/tools/run_retrain_until_replay8.ps1 deleted file mode 100644 index ac7ecbd..0000000 --- a/tools/run_retrain_until_replay8.ps1 +++ /dev/null @@ -1,152 +0,0 @@ -[CmdletBinding()] -param( - [int]$MinReplayTrades = 8, - [int]$MaxAttempts = 0, - [string]$Symbols = "", - [int]$Limit = 3000, - [switch]$DeployToPi, - [string]$PiHost = "192.168.0.185", - [string]$PiUser = "sevenhill", - [string]$PiRoot = "/mnt/data/tradebot", - [string]$PiSshKeyPath = "", - [int]$SeedStart = 0 -) - -$ErrorActionPreference = "Stop" - -$RepoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path -$RuntimeDir = Join-Path $RepoRoot "runtime" -$LoopLog = Join-Path $RuntimeDir "torch_retrain_until_replay8.log" -$GuardReport = Join-Path $RuntimeDir "torch_retrain_guard.json" -$ActiveCalibration = Join-Path $RuntimeDir "torch_threshold_calibration.json" -$Runner = Join-Path $RepoRoot "tools\run_torch_retrain.ps1" -New-Item -ItemType Directory -Force -Path $RuntimeDir | Out-Null - -function Write-LoopLog { - param([string]$Message) - $timestamp = Get-Date -Format "yyyy-MM-dd HH:mm:ssK" - "[$timestamp] $Message" | Tee-Object -FilePath $LoopLog -Append -} - -function ConvertTo-IntOrZero { - param($Value) - try { - if ($null -eq $Value) { - return 0 - } - return [int]$Value - } - catch { - return 0 - } -} - -function Read-GuardSummary { - if (-not (Test-Path $GuardReport)) { - return [pscustomobject]@{ - Accepted = $false - Reason = "guard_report_missing" - CandidateReplayTrades = 0 - CurrentReplayTrades = 0 - WalkForwardTrades = 0 - } - } - try { - $payload = Get-Content -Raw -LiteralPath $GuardReport | ConvertFrom-Json - return [pscustomobject]@{ - Accepted = [bool]$payload.accepted - Reason = [string]$payload.reason - CandidateReplayTrades = ConvertTo-IntOrZero $payload.candidate.full_replay.trades - CurrentReplayTrades = ConvertTo-IntOrZero $payload.current.full_replay.trades - WalkForwardTrades = ConvertTo-IntOrZero $payload.candidate.walk_forward_summary.trades - } - } - catch { - return [pscustomobject]@{ - Accepted = $false - Reason = "guard_report_unreadable" - CandidateReplayTrades = 0 - CurrentReplayTrades = 0 - WalkForwardTrades = 0 - } - } -} - -function Read-ActiveReplayTrades { - if (-not (Test-Path $ActiveCalibration)) { - return 0 - } - try { - $payload = Get-Content -Raw -LiteralPath $ActiveCalibration | ConvertFrom-Json - return ConvertTo-IntOrZero $payload.full_replay.trades - } - catch { - return 0 - } -} - -function Read-ActiveValidationPassed { - if (-not (Test-Path $ActiveCalibration)) { - return $false - } - try { - $payload = Get-Content -Raw -LiteralPath $ActiveCalibration | ConvertFrom-Json - return [bool]$payload.validation.passed - } - catch { - return $false - } -} - -$attempt = 0 -while ($true) { - $activeReplayTrades = Read-ActiveReplayTrades - if (Read-ActiveValidationPassed) { - Write-LoopLog "Stop condition reached: active calibration passed honest validation with full_replay.trades=$activeReplayTrades." - exit 0 - } - - $attempt += 1 - if ($SeedStart -gt 0) { - $attemptSeed = $SeedStart + $attempt - 1 - } - else { - $attemptSeed = Get-Random -Minimum 1 -Maximum 2147483647 - } - Write-LoopLog "Attempt $attempt started; seed=$attemptSeed; target full_replay.trades >= $MinReplayTrades." - - $runnerArgs = @( - "-NoProfile", - "-ExecutionPolicy", "Bypass", - "-File", $Runner, - "-Limit", $Limit.ToString(), - "-Seed", $attemptSeed.ToString() - ) - if ($Symbols) { - $runnerArgs += @("-Symbols", $Symbols) - } - if ($DeployToPi) { - $runnerArgs += "-DeployToPi" - if ($PiHost) { $runnerArgs += @("-PiHost", $PiHost) } - if ($PiUser) { $runnerArgs += @("-PiUser", $PiUser) } - if ($PiRoot) { $runnerArgs += @("-PiRoot", $PiRoot) } - if ($PiSshKeyPath) { $runnerArgs += @("-PiSshKeyPath", $PiSshKeyPath) } - } - - & powershell.exe @runnerArgs 2>&1 | Tee-Object -FilePath $LoopLog -Append - $runnerExit = $LASTEXITCODE - $summary = Read-GuardSummary - Write-LoopLog "Attempt $attempt finished; runner_exit=$runnerExit accepted=$($summary.Accepted) reason=$($summary.Reason) candidate_full_replay.trades=$($summary.CandidateReplayTrades) current_full_replay.trades=$($summary.CurrentReplayTrades) walk_forward.trades=$($summary.WalkForwardTrades)." - - if ($summary.Accepted -and (Read-ActiveValidationPassed)) { - Write-LoopLog "Stop condition reached: accepted candidate passed honest validation with full_replay.trades=$($summary.CandidateReplayTrades)." - exit 0 - } - - if ($MaxAttempts -gt 0 -and $attempt -ge $MaxAttempts) { - Write-LoopLog "MaxAttempts=$MaxAttempts reached before replay target." - exit 2 - } - - Start-Sleep -Seconds 10 -} diff --git a/tools/run_torch_retrain.ps1 b/tools/run_torch_retrain.ps1 index 73d6c24..0facd2e 100644 --- a/tools/run_torch_retrain.ps1 +++ b/tools/run_torch_retrain.ps1 @@ -22,12 +22,6 @@ param( [int]$HoldoutWindow = 0, [string]$Interval = "", [string]$EnvFile = "", - [switch]$DeployToPi, - [string]$PiHost = "", - [string]$PiUser = "", - [string]$PiRoot = "", - [string]$PiSshKeyPath = "", - [switch]$NoPiRestart, [switch]$Pooled, [switch]$SkipGuard, [switch]$ResumeCandidate @@ -105,44 +99,13 @@ function Test-TorchArtifactFile { } } -function Sync-AcceptedArtifactsToPi { - if (-not ($DeployToPi -or $env:TORCH_RETRAIN_DEPLOY_TO_PI)) { - Write-RetrainLog "Pi artifact sync disabled." - return - } - - $syncScript = Join-Path $RepoRoot "tools\sync_torch_artifacts_to_pi.ps1" - if (-not (Test-Path $syncScript)) { - throw "Pi sync script not found: $syncScript" - } - - $syncArgs = @( - "-NoProfile", - "-ExecutionPolicy", "Bypass", - "-File", $syncScript, - "-RepoRoot", $RepoRoot - ) - if ($PiHost) { $syncArgs += @("-RemoteHost", $PiHost) } - if ($PiUser) { $syncArgs += @("-RemoteUser", $PiUser) } - if ($PiRoot) { $syncArgs += @("-RemoteRoot", $PiRoot) } - if ($PiSshKeyPath) { $syncArgs += @("-SshKeyPath", $PiSshKeyPath) } - if ($NoPiRestart) { $syncArgs += "-NoRestart" } - - Write-RetrainLog "Syncing accepted Torch artifacts to Raspberry Pi." - & powershell.exe @syncArgs 2>&1 | Tee-Object -FilePath $LogFile -Append - if ($LASTEXITCODE -ne 0) { - throw "Pi artifact sync failed with exit code $LASTEXITCODE." - } - Write-RetrainLog "Pi artifact sync completed." -} - if (-not $Symbols -and $env:TORCH_RETRAIN_SYMBOLS) { $Symbols = $env:TORCH_RETRAIN_SYMBOLS } if ($Limit -le 0) { - $Limit = if ($env:TORCH_RETRAIN_LIMIT) { [int]$env:TORCH_RETRAIN_LIMIT } else { 6000 } + $Limit = if ($env:TORCH_RETRAIN_LIMIT) { [int]$env:TORCH_RETRAIN_LIMIT } else { 4000 } } -if (-not $Lookbacks) { $Lookbacks = if ($env:TORCH_RETRAIN_LOOKBACKS) { $env:TORCH_RETRAIN_LOOKBACKS } else { "32,64,128" } } -if (-not $Architectures) { $Architectures = if ($env:TORCH_RETRAIN_ARCHITECTURES) { $env:TORCH_RETRAIN_ARCHITECTURES } else { "lstm,gru" } } -if (-not $HiddenSizes) { $HiddenSizes = if ($env:TORCH_RETRAIN_HIDDEN_SIZES) { $env:TORCH_RETRAIN_HIDDEN_SIZES } else { "64,96" } } +if (-not $Lookbacks) { $Lookbacks = if ($env:TORCH_RETRAIN_LOOKBACKS) { $env:TORCH_RETRAIN_LOOKBACKS } else { "64" } } +if (-not $Architectures) { $Architectures = if ($env:TORCH_RETRAIN_ARCHITECTURES) { $env:TORCH_RETRAIN_ARCHITECTURES } else { "lstm" } } +if (-not $HiddenSizes) { $HiddenSizes = if ($env:TORCH_RETRAIN_HIDDEN_SIZES) { $env:TORCH_RETRAIN_HIDDEN_SIZES } else { "64" } } if (-not $Layers) { $Layers = if ($env:TORCH_RETRAIN_LAYERS) { $env:TORCH_RETRAIN_LAYERS } else { "2" } } if (-not $Dropouts) { $Dropouts = if ($env:TORCH_RETRAIN_DROPOUTS) { $env:TORCH_RETRAIN_DROPOUTS } else { "0.20" } } if ($Horizon -le 0) { $Horizon = if ($env:TORCH_RETRAIN_HORIZON) { [int]$env:TORCH_RETRAIN_HORIZON } else { 12 } } @@ -154,7 +117,7 @@ if (-not $EnsembleSeeds) { $EnsembleSeeds = if ($env:TORCH_RETRAIN_ENSEMBLE_SEED if ($SelectionFolds -le 0) { $SelectionFolds = if ($env:TORCH_RETRAIN_SELECTION_FOLDS) { [int]$env:TORCH_RETRAIN_SELECTION_FOLDS } else { 3 } } if ($LearningRate -le 0) { $LearningRate = if ($env:TORCH_RETRAIN_LEARNING_RATE) { [double]$env:TORCH_RETRAIN_LEARNING_RATE } else { 0.0007 } } if ($WeightDecay -le 0) { $WeightDecay = if ($env:TORCH_RETRAIN_WEIGHT_DECAY) { [double]$env:TORCH_RETRAIN_WEIGHT_DECAY } else { 0.0005 } } -if ($Epochs -le 0) { $Epochs = if ($env:TORCH_RETRAIN_EPOCHS) { [int]$env:TORCH_RETRAIN_EPOCHS } else { 70 } } +if ($Epochs -le 0) { $Epochs = if ($env:TORCH_RETRAIN_EPOCHS) { [int]$env:TORCH_RETRAIN_EPOCHS } else { 50 } } if ($Patience -le 0) { $Patience = if ($env:TORCH_RETRAIN_PATIENCE) { [int]$env:TORCH_RETRAIN_PATIENCE } else { 8 } } if ($ValidationWindow -le 0) { $ValidationWindow = if ($env:TORCH_RETRAIN_VALIDATION_WINDOW) { [int]$env:TORCH_RETRAIN_VALIDATION_WINDOW } else { 720 } } if ($HoldoutWindow -le 0) { $HoldoutWindow = if ($env:TORCH_RETRAIN_HOLDOUT_WINDOW) { [int]$env:TORCH_RETRAIN_HOLDOUT_WINDOW } else { 1000 } } @@ -302,7 +265,6 @@ try { Write-RetrainLog "Updated active threshold calibration: $(Join-Path $RuntimeDir "torch_threshold_calibration.json")" } Write-RetrainLog "Candidate accepted by guard. Active artifact: $ModelFile" - Sync-AcceptedArtifactsToPi } catch { Write-RetrainLog "ERROR: $($_.Exception.Message)" diff --git a/tools/sync_torch_artifacts_to_pi.ps1 b/tools/sync_torch_artifacts_to_pi.ps1 deleted file mode 100644 index 79878e6..0000000 --- a/tools/sync_torch_artifacts_to_pi.ps1 +++ /dev/null @@ -1,95 +0,0 @@ -[CmdletBinding()] -param( - [string]$RepoRoot = "", - [string]$RemoteHost = "", - [string]$RemoteUser = "", - [string]$RemoteRoot = "", - [string]$SshKeyPath = "", - [string]$ServiceName = "tradebot", - [switch]$NoRestart, - [switch]$DryRun -) - -$ErrorActionPreference = "Stop" - -if (-not $RepoRoot) { $RepoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path } -if (-not $RemoteHost -and $env:TORCH_DEPLOY_PI_HOST) { $RemoteHost = $env:TORCH_DEPLOY_PI_HOST } -if (-not $RemoteUser -and $env:TORCH_DEPLOY_PI_USER) { $RemoteUser = $env:TORCH_DEPLOY_PI_USER } -if (-not $RemoteRoot -and $env:TORCH_DEPLOY_PI_ROOT) { $RemoteRoot = $env:TORCH_DEPLOY_PI_ROOT } -if (-not $SshKeyPath -and $env:TORCH_DEPLOY_PI_SSH_KEY) { $SshKeyPath = $env:TORCH_DEPLOY_PI_SSH_KEY } -if (-not $RemoteHost) { $RemoteHost = "192.168.0.185" } -if (-not $RemoteUser) { $RemoteUser = "sevenhill" } -if (-not $RemoteRoot) { $RemoteRoot = "/mnt/data/tradebot" } - -$RuntimeDir = Join-Path $RepoRoot "runtime" -$artifactNames = @( - "lstm_forecaster.json", - "torch_retrain_guard.json", - "torch_threshold_calibration.json" -) -$localFiles = @() -foreach ($name in $artifactNames) { - $path = Join-Path $RuntimeDir $name - if (Test-Path $path) { - $localFiles += (Resolve-Path $path).Path - } -} -if ($localFiles.Count -eq 0) { - throw "No Torch artifacts found in $RuntimeDir." -} - -function ConvertTo-RemoteSingleQuoted { - param([string]$Value) - return "'" + ($Value -replace "'", "'\''") + "'" -} - -function Invoke-LoggedCommand { - param( - [string]$Exe, - [string[]]$Arguments - ) - $rendered = @($Exe) + $Arguments - Write-Host ($rendered -join " ") - if ($DryRun) { - return - } - & $Exe @Arguments - if ($LASTEXITCODE -ne 0) { - throw "$Exe failed with exit code $LASTEXITCODE." - } -} - -$ssh = (Get-Command "ssh.exe" -ErrorAction SilentlyContinue) -if (-not $ssh) { $ssh = Get-Command "ssh" -ErrorAction Stop } -$scp = (Get-Command "scp.exe" -ErrorAction SilentlyContinue) -if (-not $scp) { $scp = Get-Command "scp" -ErrorAction Stop } - -$commonSshArgs = @("-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new", "-o", "ConnectTimeout=15") -if ($SshKeyPath) { - $expandedKey = $ExecutionContext.SessionState.Path.GetUnresolvedProviderPathFromPSPath($SshKeyPath) - $commonSshArgs += @("-i", $expandedKey) -} - -$remote = "${RemoteUser}@${RemoteHost}" -$remoteRuntime = "$RemoteRoot/runtime" -$remoteIncoming = "$remoteRuntime/.incoming-torch" -$mkdirCommand = "mkdir -p $(ConvertTo-RemoteSingleQuoted $remoteIncoming) $(ConvertTo-RemoteSingleQuoted $remoteRuntime)" -Invoke-LoggedCommand $ssh.Source (@($commonSshArgs + @($remote, $mkdirCommand))) - -$destination = "${remote}:$remoteIncoming/" -Invoke-LoggedCommand $scp.Source (@($commonSshArgs + $localFiles + @($destination))) - -$moveParts = @() -foreach ($path in $localFiles) { - $name = Split-Path $path -Leaf - $moveParts += "mv -f $(ConvertTo-RemoteSingleQuoted "$remoteIncoming/$name") $(ConvertTo-RemoteSingleQuoted "$remoteRuntime/$name")" -} -$moveCommand = $moveParts -join " && " -Invoke-LoggedCommand $ssh.Source (@($commonSshArgs + @($remote, $moveCommand))) - -if (-not $NoRestart) { - $restartCommand = "cd $(ConvertTo-RemoteSingleQuoted $RemoteRoot) && docker compose restart $(ConvertTo-RemoteSingleQuoted $ServiceName)" - Invoke-LoggedCommand $ssh.Source (@($commonSshArgs + @($remote, $restartCommand))) -} - -Write-Host "Synced Torch artifacts to ${remote}:$remoteRuntime" diff --git a/tools/windows_training_agent.py b/tools/windows_training_agent.py index 6ef8f2c..4faf351 100644 --- a/tools/windows_training_agent.py +++ b/tools/windows_training_agent.py @@ -54,23 +54,34 @@ def poll_once(args: argparse.Namespace, repo_root: Path, runtime_dir: Path, log_ return job = claim.get("job") if isinstance(claim.get("job"), dict) else {} job_id = str(job.get("id") or "") + lease_token = str(claim.get("lease_token") or "") if not job_id: return + if not lease_token: + raise RuntimeError("training server did not issue a job lease") log(log_path, f"Claimed retrain job {job_id}") - report_progress(args, job_id, "running", "claimed", 2, "Задание получено Windows-agent") + report_progress(args, job_id, lease_token, "running", "claimed", 2, "Задание получено Windows-agent") success = False message = "" summary: dict[str, Any] = {} try: - run_retrain(args, job_id, job, repo_root, log_path) + run_retrain(args, job_id, lease_token, job, repo_root, log_path) summary = read_json(runtime_dir / "torch_retrain_guard.json") accepted = summary.get("accepted") is True if accepted: - report_progress(args, job_id, "running", "uploading", 72, "Обучение завершено, загружаю артефакты") + report_progress( + args, + job_id, + lease_token, + "running", + "uploading", + 72, + "Обучение завершено, загружаю артефакты", + ) for name in ARTIFACT_NAMES: path = runtime_dir / name if path.is_file(): - upload_artifact(args, job_id, path, log_path) + upload_artifact(args, job_id, lease_token, path, log_path) message = "training completed; candidate accepted" log(log_path, f"Completed retrain job {job_id}; candidate accepted") else: @@ -82,11 +93,23 @@ def poll_once(args: argparse.Namespace, repo_root: Path, runtime_dir: Path, log_ message = str(exc) log(log_path, f"Job {job_id} failed: {message}") finally: - payload = {"success": success, "message": message, "summary": summary} + payload = { + "success": success, + "message": message, + "summary": summary, + "lease_token": lease_token, + } api_json(args, f"/api/training/jobs/{job_id}/complete", payload) -def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo_root: Path, log_path: Path) -> None: +def run_retrain( + args: argparse.Namespace, + job_id: str, + lease_token: str, + 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}") @@ -126,12 +149,20 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo value = parameters.get(key) if value not in (None, ""): cmd.extend([ps_arg, str(value)]) - if parameters.get("pooled") is True: + if parameters.get("pooled", True) is True: cmd.append("-Pooled") if parameters.get("resume_candidate") is True: cmd.append("-ResumeCandidate") log(log_path, "Running retrain: " + " ".join(quote_for_log(part) for part in cmd)) - report_progress(args, job_id, "running", "training", 8, "PyTorch retrain запущен") + report_progress( + args, + job_id, + lease_token, + "running", + "training", + 8, + "PyTorch retrain запущен", + ) line_count = 0 output_queue: queue.Queue[str] = queue.Queue() @@ -175,7 +206,16 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo report_message = last_message if not got_line: report_message = training_heartbeat_message(now, started_at, last_output_at, last_message) - safe_report_progress(args, job_id, "running", "training", progress, report_message, log_path) + safe_report_progress( + args, + job_id, + lease_token, + "running", + "training", + progress, + report_message, + log_path, + ) last_report_at = now if process.poll() is not None and output_queue.empty(): @@ -185,7 +225,15 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo code = process.wait() if code != 0: raise RuntimeError(f"retrain failed with exit code {code}") - report_progress(args, job_id, "running", "guard", 70, "Guard завершён, подготавливаю артефакты") + report_progress( + args, + job_id, + lease_token, + "running", + "guard", + 70, + "Guard завершён, подготавливаю артефакты", + ) def friendly_training_message(message: str) -> str: @@ -282,7 +330,13 @@ def format_duration(seconds: float) -> str: return f"{seconds_part}с" -def upload_artifact(args: argparse.Namespace, job_id: str, path: Path, log_path: Path) -> None: +def upload_artifact( + args: argparse.Namespace, + job_id: str, + lease_token: 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) @@ -297,16 +351,26 @@ def upload_artifact(args: argparse.Namespace, job_id: str, path: Path, log_path: "total": total, "sha256": digest, "data_base64": base64.b64encode(data).decode("ascii"), + "lease_token": lease_token, } api_json(args, f"/api/training/jobs/{job_id}/artifacts/chunk", payload, timeout=120) if index == 0 or index == total - 1 or index % 10 == 0: progress = 72 + int(((index + 1) / total) * 23) - report_progress(args, job_id, "running", "uploading", progress, f"Загружаю {path.name}: {index + 1}/{total}") + report_progress( + args, + job_id, + lease_token, + "running", + "uploading", + progress, + f"Загружаю {path.name}: {index + 1}/{total}", + ) def report_progress( args: argparse.Namespace, job_id: str, + lease_token: str, status: str, phase: str, progress_percent: int, @@ -321,6 +385,7 @@ def report_progress( "progress_percent": progress_percent, "message": message, "worker": worker_payload(args, Path(args.repo_root).resolve()), + "lease_token": lease_token, }, ) @@ -328,6 +393,7 @@ def report_progress( def safe_report_progress( args: argparse.Namespace, job_id: str, + lease_token: str, status: str, phase: str, progress_percent: int, @@ -337,7 +403,15 @@ def safe_report_progress( last_error: Exception | None = None for attempt in range(1, 4): try: - report_progress(args, job_id, status, phase, progress_percent, message) + report_progress( + args, + job_id, + lease_token, + status, + phase, + progress_percent, + message, + ) return except Exception as exc: # noqa: BLE001 - keep the local training process alive. last_error = exc @@ -385,7 +459,7 @@ def worker_payload(args: argparse.Namespace, repo_root: Path) -> dict[str, Any]: "worker_id": args.worker_id or f"{name}:{repo_root}", "name": name, "path": str(repo_root), - "version": "1", + "version": "2", }