diff --git a/README.md b/README.md index 58fd99c..f9d12e3 100644 --- a/README.md +++ b/README.md @@ -95,12 +95,16 @@ powershell -ExecutionPolicy Bypass -File tools\install_windows_torch_retrainer.p Для удалённого запуска с телефона или с бота используется Windows training agent. Бот на `tb.kusoft.xyz` хранит очередь заданий, а Windows-машина сама подключается к интернету, забирает задания, обучает модель и загружает артефакты обратно: ```powershell -powershell -ExecutionPolicy Bypass -File tools\install_windows_training_agent.ps1 -ApiAuth "login:password" -StartNow +powershell -ExecutionPolicy Bypass -File tools\install_windows_training_agent.ps1 -ApiAuth "" -StartNow ``` -Установщик регистрирует Scheduled Task `TradeBot Windows Training Agent` при входе в Windows и удаляет старые локальные retrain-задачи, чтобы обучение запускалось через очередь, а не двумя независимыми механизмами. +Установщик сохраняет worker-токен через Windows DPAPI, удаляет его старую plaintext-копию из пользовательского окружения и включает постоянный запуск агента. С правами администратора используется Scheduled Task с watchdog; без повышения прав — штатный ярлык в пользовательской папке Startup. Старые локальные 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`. +По умолчанию Windows-agent обучает pooled multi-asset PyTorch `LSTM/GRU` на `6000` часовых свечах: общие recurrent-веса получают one-hot embedding символа, прогноз усредняется по seed `7/19/43`, модели сравниваются на validation-folds, а пороги калибруются отдельно для каждой пары. 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 не ослабляются. Параметры можно переопределить через 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 и ранговую связь прогноза с будущей доходностью, а не только MAE. В каждом walk-forward fold вероятность `P(up)` калибруется Platt-моделью исключительно на train-части; затем на этой же train-части выбираются глобальные и per-symbol пороги, которые применяются к test-части. Калибратор не имеет fallback на единичные сделки: если минимальная статистика не набрана, кандидат получает `calibration_insufficient` и не может пройти gate. + +Основной decision horizon — `12h`, дополнительные горизонты — `3/6/12/24`. Это согласует прогноз с round-trip cost: при текущих fee/slippage полный вход-выход стоит около `0.26%`, поэтому прежний `3h` target чаще описывал шум, который не покрывал издержки. 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: diff --git a/android/TradeBotMonitor/README.md b/android/TradeBotMonitor/README.md index b2c5358..1d82a97 100644 --- a/android/TradeBotMonitor/README.md +++ b/android/TradeBotMonitor/README.md @@ -38,15 +38,11 @@ https://tb.kusoft.xyz Этот адрес установлен в приложении по умолчанию. Если в настройках ввести просто `tb.kusoft.xyz`, приложение само добавит `https://`. -Если домен защищён авторизацией, в поле `API auth` можно указать: - -- `login:password` — приложение отправит HTTP Basic; -- `Basic ...` — готовый Basic header; -- `Bearer ...` или просто токен — приложение отправит Bearer. +В поле `API-токен` указывается отдельный токен Android-клиента (`TRADEBOT_API_TOKEN` на сервере). Логин и пароль reverse proxy приложению не нужны. Токен отправляется как Bearer/X-TradeBot-Token и хранится зашифрованным ключом Android Keystore. ## Переобучение -Телефон не обучает модель локально. Вкладка `Обучение` ставит задание в очередь на `tb.kusoft.xyz`, а Windows-agent на закреплённой машине `DESKTOP-TMFDL0H` сам выходит в интернет, забирает задание, обучает модель и отправляет артефакты обратно боту. Так телефон становится пультом запуска/расписания, а тяжёлый PyTorch retrain остаётся на нормальном компьютере даже если он находится в другой сети. +Телефон не обучает модель локально. Вкладка `Обучение` ставит задание в очередь на `tb.kusoft.xyz`, а Windows-agent на закреплённой машине `SEVENHILL` (`G:\Repos\TradeBot`) сам выходит в интернет, забирает задание, обучает модель и отправляет артефакты обратно боту. Так телефон становится пультом запуска/расписания, а тяжёлый PyTorch retrain остаётся на нормальном компьютере даже если он находится в другой сети. ## Live-торговля diff --git a/android/TradeBotMonitor/app/build.gradle.kts b/android/TradeBotMonitor/app/build.gradle.kts index f681b5a..0c9f416 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 = 18 - versionName = "0.3.0" + versionCode = 20 + versionName = "0.4.1" } } 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 1851deb..4b6a6b7 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 @@ -16,7 +16,14 @@ class AppPrefs(context: Context) { if (saved.isNullOrBlank() || saved == LEGACY_PI_API_BASE_URL) { prefs.edit().putString("api_base_url", DEFAULT_API_BASE_URL).apply() } - if (prefs.getString("training_computer_name", null).isNullOrBlank()) { + 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 + ) { pinDefaultTrainingComputer() } } @@ -134,8 +141,10 @@ 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 = "DESKTOP-TMFDL0H" - const val DEFAULT_TRAINING_COMPUTER_PATH = "C:\\Repos\\TradeBot" + 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 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 4c4383b..c517a17 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 @@ -489,21 +489,19 @@ class MainActivity : Activity() { liveSignature = liveBlockSignature(snapshot), ) - val (savedLogin, savedPassword) = authParts(prefs.commandToken) val apiInput = input("Адрес API бота", prefs.apiBaseUrl) - val loginInput = input("Логин API", savedLogin) - val passwordInput = input("Пароль API", savedPassword).apply { + val tokenInput = input("API-токен", prefs.commandToken).apply { inputType = InputType.TYPE_CLASS_TEXT or InputType.TYPE_TEXT_VARIATION_PASSWORD } box.addView(section("Подключение", LinearLayout(this).apply { orientation = LinearLayout.VERTICAL addView(apiInput) - addView(loginInput.top(dp(8))) - addView(passwordInput.top(dp(8))) + addView(tokenInput.top(dp(8))) + addView(text("Токен выдаётся сервером для Android-клиента и хранится в Android Keystore.", 12f, Typeface.NORMAL, palette.muted).top(dp(8))) addView(actionRow( "Сохранить" to { prefs.apiBaseUrl = apiInput.text.toString() - prefs.commandToken = authToken(loginInput.text.toString(), passwordInput.text.toString()) + prefs.commandToken = tokenInput.text.toString() toast("Подключение сохранено") refreshData(silent = false) }, @@ -1020,6 +1018,13 @@ class MainActivity : Activity() { LinearLayout(this).apply { val coordination = retrain.optJSONObject("coordination") ?: JSONObject() val activeJob = coordination.optJSONObject("active_job") + val worker = coordination.optJSONObject("worker") + val trainingComputerName = worker?.optStringClean("name") + ?.takeIf { it.isNotBlank() } + ?: prefs.trainingComputerName + val trainingComputerPath = worker?.optStringClean("path") + ?.takeIf { it.isNotBlank() } + ?: prefs.trainingComputerPath val agentRecentlySeen = coordination.optBoolean( "agent_recently_seen", coordination.optBoolean("agent_online", false), @@ -1036,8 +1041,8 @@ class MainActivity : Activity() { val connectionColor = if (agentRecentlySeen || agentBusy) palette.green else palette.amber 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(text(trainingComputerName, 18f, Typeface.BOLD, palette.green).top(dp(5))) + addView(text(trainingComputerPath, 12f, Typeface.NORMAL, palette.muted).top(dp(4))) addView(keyValueLine("Связь агента", connectionText, connectionColor).top(dp(4))) addView(text("Бот ставит задания через tb.kusoft.xyz, а этот Windows-agent сам забирает их через интернет и возвращает результат.", 12f, Typeface.NORMAL, palette.muted).top(dp(8))) } @@ -1052,10 +1057,9 @@ class MainActivity : Activity() { val agentOnline = coordination.optBoolean("agent_online", false) addView(keyValueLine("Состояние", if (agentOnline) "готов к запуску" else "ждет Windows-agent", if (agentOnline) palette.green else palette.amber).top(dp(8))) if (latestJob != null) { - val latestStatus = latestJob.optStringClean("status") - val latestPhase = latestJob.optStringClean("phase") val latestMessage = latestJob.optStringClean("message") - addView(keyValueLine("Последнее обучение", trainingJobLabel(latestStatus, latestPhase), trainingJobColor(latestStatus, latestPhase)).top(dp(8))) + addView(keyValueLine("Последнее обучение", trainingJobLabel(latestJob), trainingJobColor(latestJob)).top(dp(8))) + addView(trainingModelDecisionLine(latestJob).top(dp(6))) if (latestMessage.isNotBlank()) { addView(text(latestMessage.take(160), 12f, Typeface.NORMAL, palette.muted).top(dp(6))) } @@ -1064,12 +1068,11 @@ class MainActivity : Activity() { } val job = activeJob val status = job.optStringClean("status") - val phase = job.optStringClean("phase") val progress = job.optInt("progress_percent", if (status == "completed") 100 else 0).coerceIn(0, 100) val message = job.optStringClean("message") - addView(keyValueLine("Состояние", trainingJobLabel(status, phase), trainingJobColor(status, phase)).top(dp(8))) + addView(keyValueLine("Состояние", trainingJobLabel(job), trainingJobColor(job)).top(dp(8))) addView(allocationBar(progress / 100.0).top(dp(10))) - addView(keyValueLine("Прогресс", "$progress%", trainingJobColor(status, phase)).top(dp(8))) + addView(keyValueLine("Прогресс", "$progress%", trainingJobColor(job)).top(dp(8))) if (message.isNotBlank()) { addView(text(message.take(160), 12f, Typeface.NORMAL, palette.text).top(dp(8))) } @@ -1241,22 +1244,19 @@ class MainActivity : Activity() { if (!isAuthError()) { return emptyState("Нет данных от API. ${lastError.ifBlank { "Проверь подключение." }}") } - val (savedLogin, savedPassword) = authParts(prefs.commandToken) val apiInput = input("Адрес API бота", prefs.apiBaseUrl) - val loginInput = input("Логин API", savedLogin) - val passwordInput = input("Пароль API", savedPassword).apply { + val tokenInput = input("API-токен", prefs.commandToken).apply { inputType = InputType.TYPE_CLASS_TEXT or InputType.TYPE_TEXT_VARIATION_PASSWORD } - return section("Нужен вход в API", LinearLayout(this).apply { + return section("Нужен API-токен", LinearLayout(this).apply { orientation = LinearLayout.VERTICAL - addView(text("Сервер tb.kusoft.xyz отвечает 401, значит он доступен, но требует авторизацию. Введите логин и пароль, приложение само отправит Basic Auth.", 13f, Typeface.NORMAL, palette.muted)) + addView(text("Сервер доступен, но Android API-токен отсутствует или неверен. Введите отдельный токен клиента; логин и пароль сайта здесь не используются.", 13f, Typeface.NORMAL, palette.muted)) addView(apiInput.top(dp(12))) - addView(loginInput.top(dp(8))) - addView(passwordInput.top(dp(8))) + addView(tokenInput.top(dp(8))) addView(actionRow( "Подключиться" to { prefs.apiBaseUrl = apiInput.text.toString() - prefs.commandToken = authToken(loginInput.text.toString(), passwordInput.text.toString()) + prefs.commandToken = tokenInput.text.toString() toast("Доступ сохранен, проверяю API") refreshData(silent = false) }, @@ -1642,11 +1642,40 @@ class MainActivity : Activity() { }, ).joinToString("|") - private fun trainingJobLabel(status: String, phase: String): String = - when (status) { + private fun trainingModelDecision(job: JSONObject): String { + val explicit = job.optStringClean("model_decision") + if (explicit in setOf("accepted", "rejected")) return explicit + val summary = job.optJSONObject("summary") ?: return "" + return if (summary.has("accepted")) { + if (summary.optBoolean("accepted", false)) "accepted" else "rejected" + } else { + "" + } + } + + private fun trainingModelDecisionLine(job: JSONObject): View { + val decision = trainingModelDecision(job) + val value = when (decision) { + "accepted" -> "кандидат принят" + "rejected" -> "кандидат отклонён" + else -> "нет решения gate" + } + val color = when (decision) { + "accepted" -> palette.green + "rejected" -> palette.amber + else -> palette.muted + } + return keyValueLine("Результат модели", value, color) + } + + private fun trainingJobLabel(job: JSONObject): String { + val status = job.optStringClean("status") + val phase = job.optStringClean("phase") + val completedWithDecision = trainingModelDecision(job).isNotBlank() + return when (status) { "pending" -> "ждет Windows-agent" "completed" -> "завершено успешно" - "failed" -> "ошибка обучения" + "failed" -> if (completedWithDecision) "завершено успешно" else "ошибка обучения" "running" -> when (phase) { "claimed" -> "задание получено" "training" -> "идет обучение" @@ -1656,14 +1685,19 @@ class MainActivity : Activity() { } else -> "готов к запуску" } + } - private fun trainingJobColor(status: String, phase: String): Int = - when { - status == "completed" -> palette.green + private fun trainingJobColor(job: JSONObject): Int { + val status = job.optStringClean("status") + val phase = job.optStringClean("phase") + val completedWithDecision = trainingModelDecision(job).isNotBlank() + return when { + status == "completed" || completedWithDecision -> palette.green status == "failed" -> palette.red status == "pending" || status == "running" || phase.isNotBlank() -> palette.amber else -> palette.text } + } private fun normalizedAction(raw: String?): String { val value = raw.orEmpty().uppercase(Locale.US) @@ -1700,27 +1734,6 @@ class MainActivity : Activity() { else -> value.ifBlank { "нет данных" } } - private fun authParts(value: String): Pair { - val trimmed = value.trim() - if (trimmed.isBlank() || trimmed.startsWith("Bearer ", ignoreCase = true)) { - return "" to "" - } - val basicPrefix = "Basic " - val raw = if (trimmed.startsWith(basicPrefix, ignoreCase = true)) "" else trimmed - val separator = raw.indexOf(':') - return if (separator >= 0) { - raw.substring(0, separator) to raw.substring(separator + 1) - } else { - raw to "" - } - } - - private fun authToken(login: String, password: String): String { - val cleanLogin = login.trim() - val cleanPassword = password.trim() - return if (cleanLogin.isBlank() && cleanPassword.isBlank()) "" else "$cleanLogin:$cleanPassword" - } - private fun modelLabel(value: String): String = when { value.contains("gru", ignoreCase = true) -> "PyTorch GRU" 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 1731298..86b3e48 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 @@ -1,6 +1,5 @@ package xyz.kusoft.tradebotmonitor -import android.util.Base64 import org.json.JSONArray import org.json.JSONObject import java.io.BufferedReader @@ -98,7 +97,7 @@ class TradeBotApi( connection.disconnect() if (code !in 200..299) { if (code == HttpURLConnection.HTTP_UNAUTHORIZED) { - throw IllegalStateException("HTTP 401: сервер требует логин и пароль") + throw IllegalStateException("HTTP 401: API-токен отсутствует или неверен") } throw IllegalStateException("HTTP $code: ${text.take(240)}") } @@ -282,15 +281,8 @@ class TradeBotApi( private fun applyAuthHeaders(connection: HttpURLConnection, token: String) { val value = token.trim() if (value.isBlank()) return - connection.setRequestProperty("X-TradeBot-Token", value) - val authorization = when { - value.startsWith("Basic ", ignoreCase = true) -> value - value.startsWith("Bearer ", ignoreCase = true) -> value - ":" in value -> { - val encoded = Base64.encodeToString(value.toByteArray(StandardCharsets.UTF_8), Base64.NO_WRAP) - "Basic $encoded" - } - else -> "Bearer $value" - } + val rawToken = value.removePrefix("Bearer ").removePrefix("bearer ").trim() + connection.setRequestProperty("X-TradeBot-Token", rawToken) + val authorization = "Bearer $rawToken" connection.setRequestProperty("Authorization", authorization) } diff --git a/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.jar b/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.jar new file mode 100644 index 0000000..d997cfc Binary files /dev/null and b/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.jar differ diff --git a/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.properties b/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.properties new file mode 100644 index 0000000..c61a118 --- /dev/null +++ b/android/TradeBotMonitor/gradle/wrapper/gradle-wrapper.properties @@ -0,0 +1,7 @@ +distributionBase=GRADLE_USER_HOME +distributionPath=wrapper/dists +distributionUrl=https\://services.gradle.org/distributions/gradle-9.4.1-bin.zip +networkTimeout=10000 +validateDistributionUrl=true +zipStoreBase=GRADLE_USER_HOME +zipStorePath=wrapper/dists diff --git a/android/TradeBotMonitor/gradlew b/android/TradeBotMonitor/gradlew new file mode 100644 index 0000000..739907d --- /dev/null +++ b/android/TradeBotMonitor/gradlew @@ -0,0 +1,248 @@ +#!/bin/sh + +# +# Copyright © 2015 the original authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 +# + +############################################################################## +# +# Gradle start up script for POSIX generated by Gradle. +# +# Important for running: +# +# (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is +# noncompliant, but you have some other compliant shell such as ksh or +# bash, then to run this script, type that shell name before the whole +# command line, like: +# +# ksh Gradle +# +# Busybox and similar reduced shells will NOT work, because this script +# requires all of these POSIX shell features: +# * functions; +# * expansions «$var», «${var}», «${var:-default}», «${var+SET}», +# «${var#prefix}», «${var%suffix}», and «$( cmd )»; +# * compound commands having a testable exit status, especially «case»; +# * various built-in commands including «command», «set», and «ulimit». +# +# Important for patching: +# +# (2) This script targets any POSIX shell, so it avoids extensions provided +# by Bash, Ksh, etc; in particular arrays are avoided. +# +# The "traditional" practice of packing multiple parameters into a +# space-separated string is a well documented source of bugs and security +# problems, so this is (mostly) avoided, by progressively accumulating +# options in "$@", and eventually passing that to Java. +# +# Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS, +# and GRADLE_OPTS) rely on word-splitting, this is performed explicitly; +# see the in-line comments for details. +# +# There are tweaks for specific operating systems such as AIX, CygWin, +# Darwin, MinGW, and NonStop. +# +# (3) This script is generated from the Groovy template +# https://github.com/gradle/gradle/blob/2d6327017519d23b96af35865dc997fcb544fb40/platforms/jvm/plugins-application/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt +# within the Gradle project. +# +# You can find Gradle at https://github.com/gradle/gradle/. +# +############################################################################## + +# Attempt to set APP_HOME + +# Resolve links: $0 may be a link +app_path=$0 + +# Need this for daisy-chained symlinks. +while + APP_HOME=${app_path%"${app_path##*/}"} # leaves a trailing /; empty if no leading path + [ -h "$app_path" ] +do + ls=$( ls -ld "$app_path" ) + link=${ls#*' -> '} + case $link in #( + /*) app_path=$link ;; #( + *) app_path=$APP_HOME$link ;; + esac +done + +# This is normally unused +# shellcheck disable=SC2034 +APP_BASE_NAME=${0##*/} +# Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036) +APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s\n' "$PWD" ) || exit + +# Use the maximum available, or set MAX_FD != -1 to use that value. +MAX_FD=maximum + +warn () { + echo "$*" +} >&2 + +die () { + echo + echo "$*" + echo + exit 1 +} >&2 + +# OS specific support (must be 'true' or 'false'). +cygwin=false +msys=false +darwin=false +nonstop=false +case "$( uname )" in #( + CYGWIN* ) cygwin=true ;; #( + Darwin* ) darwin=true ;; #( + MSYS* | MINGW* ) msys=true ;; #( + NONSTOP* ) nonstop=true ;; +esac + + + +# Determine the Java command to use to start the JVM. +if [ -n "$JAVA_HOME" ] ; then + if [ -x "$JAVA_HOME/jre/sh/java" ] ; then + # IBM's JDK on AIX uses strange locations for the executables + JAVACMD=$JAVA_HOME/jre/sh/java + else + JAVACMD=$JAVA_HOME/bin/java + fi + if [ ! -x "$JAVACMD" ] ; then + die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +else + JAVACMD=java + if ! command -v java >/dev/null 2>&1 + then + die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +fi + +# Increase the maximum file descriptors if we can. +if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then + case $MAX_FD in #( + max*) + # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked. + # shellcheck disable=SC2039,SC3045 + MAX_FD=$( ulimit -H -n ) || + warn "Could not query maximum file descriptor limit" + esac + case $MAX_FD in #( + '' | soft) :;; #( + *) + # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked. + # shellcheck disable=SC2039,SC3045 + ulimit -n "$MAX_FD" || + warn "Could not set maximum file descriptor limit to $MAX_FD" + esac +fi + +# Collect all arguments for the java command, stacking in reverse order: +# * args from the command line +# * the main class name +# * -classpath +# * -D...appname settings +# * --module-path (only if needed) +# * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables. + +# For Cygwin or MSYS, switch paths to Windows format before running java +if "$cygwin" || "$msys" ; then + APP_HOME=$( cygpath --path --mixed "$APP_HOME" ) + + JAVACMD=$( cygpath --unix "$JAVACMD" ) + + # Now convert the arguments - kludge to limit ourselves to /bin/sh + for arg do + if + case $arg in #( + -*) false ;; # don't mess with options #( + /?*) t=${arg#/} t=/${t%%/*} # looks like a POSIX filepath + [ -e "$t" ] ;; #( + *) false ;; + esac + then + arg=$( cygpath --path --ignore --mixed "$arg" ) + fi + # Roll the args list around exactly as many times as the number of + # args, so each arg winds up back in the position where it started, but + # possibly modified. + # + # NB: a `for` loop captures its iteration list before it begins, so + # changing the positional parameters here affects neither the number of + # iterations, nor the values presented in `arg`. + shift # remove old arg + set -- "$@" "$arg" # push replacement arg + done +fi + + +# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"' + +# Collect all arguments for the java command: +# * DEFAULT_JVM_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments, +# and any embedded shellness will be escaped. +# * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be +# treated as '${Hostname}' itself on the command line. + +set -- \ + "-Dorg.gradle.appname=$APP_BASE_NAME" \ + -jar "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" \ + "$@" + +# Stop when "xargs" is not available. +if ! command -v xargs >/dev/null 2>&1 +then + die "xargs is not available" +fi + +# Use "xargs" to parse quoted args. +# +# With -n1 it outputs one arg per line, with the quotes and backslashes removed. +# +# In Bash we could simply go: +# +# readarray ARGS < <( xargs -n1 <<<"$var" ) && +# set -- "${ARGS[@]}" "$@" +# +# but POSIX shell has neither arrays nor command substitution, so instead we +# post-process each arg (as a line of input to sed) to backslash-escape any +# character that might be a shell metacharacter, then use eval to reverse +# that process (while maintaining the separation between arguments), and wrap +# the whole thing up as a single "set" statement. +# +# This will of course break if any of these variables contains a newline or +# an unmatched quote. +# + +eval "set -- $( + printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" | + xargs -n1 | + sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' | + tr '\n' ' ' + )" '"$@"' + +exec "$JAVACMD" "$@" diff --git a/android/TradeBotMonitor/gradlew.bat b/android/TradeBotMonitor/gradlew.bat new file mode 100644 index 0000000..c4bdd3a --- /dev/null +++ b/android/TradeBotMonitor/gradlew.bat @@ -0,0 +1,93 @@ +@rem +@rem Copyright 2015 the original author or authors. +@rem +@rem Licensed under the Apache License, Version 2.0 (the "License"); +@rem you may not use this file except in compliance with the License. +@rem You may obtain a copy of the License at +@rem +@rem https://www.apache.org/licenses/LICENSE-2.0 +@rem +@rem Unless required by applicable law or agreed to in writing, software +@rem distributed under the License is distributed on an "AS IS" BASIS, +@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +@rem See the License for the specific language governing permissions and +@rem limitations under the License. +@rem +@rem SPDX-License-Identifier: Apache-2.0 +@rem + +@if "%DEBUG%"=="" @echo off +@rem ########################################################################## +@rem +@rem Gradle startup script for Windows +@rem +@rem ########################################################################## + +@rem Set local scope for the variables with windows NT shell +if "%OS%"=="Windows_NT" setlocal + +set DIRNAME=%~dp0 +if "%DIRNAME%"=="" set DIRNAME=. +@rem This is normally unused +set APP_BASE_NAME=%~n0 +set APP_HOME=%DIRNAME% + +@rem Resolve any "." and ".." in APP_HOME to make it shorter. +for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi + +@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m" + +@rem Find java.exe +if defined JAVA_HOME goto findJavaFromJavaHome + +set JAVA_EXE=java.exe +%JAVA_EXE% -version >NUL 2>&1 +if %ERRORLEVEL% equ 0 goto execute + +echo. 1>&2 +echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. 1>&2 +echo. 1>&2 +echo Please set the JAVA_HOME variable in your environment to match the 1>&2 +echo location of your Java installation. 1>&2 + +goto fail + +:findJavaFromJavaHome +set JAVA_HOME=%JAVA_HOME:"=% +set JAVA_EXE=%JAVA_HOME%/bin/java.exe + +if exist "%JAVA_EXE%" goto execute + +echo. 1>&2 +echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% 1>&2 +echo. 1>&2 +echo Please set the JAVA_HOME variable in your environment to match the 1>&2 +echo location of your Java installation. 1>&2 + +goto fail + +:execute +@rem Setup the command line + + + +@rem Execute Gradle +"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -jar "%APP_HOME%\gradle\wrapper\gradle-wrapper.jar" %* + +:end +@rem End local scope for the variables with windows NT shell +if %ERRORLEVEL% equ 0 goto mainEnd + +:fail +rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of +rem the _cmd.exe /c_ return code! +set EXIT_CODE=%ERRORLEVEL% +if %EXIT_CODE% equ 0 set EXIT_CODE=1 +if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE% +exit /b %EXIT_CODE% + +:mainEnd +if "%OS%"=="Windows_NT" endlocal + +:omega diff --git a/crypto_spot_bot/strategy.py b/crypto_spot_bot/strategy.py index 4706af1..6a87bb3 100644 --- a/crypto_spot_bot/strategy.py +++ b/crypto_spot_bot/strategy.py @@ -644,8 +644,12 @@ def _torch_forecast_entry_signal( expected_return = _safe_float(forecast.get("expected_return_percent"), 0.0) probability_up = _safe_float(forecast.get("probability_up"), 0.5) skill = _safe_float(forecast.get("skill"), 0.0) - min_edge = max(0.0, settings.time_series_min_edge_percent) - min_probability = _torch_min_probability(settings) + min_edge = max(0.0, _safe_float(forecast.get("calibrated_min_edge_percent"), settings.time_series_min_edge_percent)) + min_probability = _clamp( + _safe_float(forecast.get("calibrated_min_probability_up"), _torch_min_probability(settings)), + 0.5, + 0.95, + ) probe_min_edge = max(0.0, min(settings.time_series_probe_min_edge_percent, min_edge)) probe_min_probability = round( _clamp(settings.time_series_probe_min_probability_up, min_probability, 0.85), @@ -716,7 +720,7 @@ def _torch_forecast_entry_signal( and expected_return >= 0.0 and probability_up >= rebound_model_probability_min and skill > 0.0 - and confidence >= settings.time_series_min_confidence + and confidence >= _safe_float(forecast.get("calibrated_min_confidence"), settings.time_series_min_confidence) ) fallback_rebound_entry_ok = bool( settings.time_series_rebound_fallback_enabled @@ -725,7 +729,7 @@ def _torch_forecast_entry_signal( and quality_gate_ok and model_fresh_ok and not bool(forecast.get("block_entry", False)) - and confidence >= settings.time_series_min_confidence + and confidence >= _safe_float(forecast.get("calibrated_min_confidence"), settings.time_series_min_confidence) ) rebound_entry_ok = model_rebound_entry_ok or fallback_rebound_entry_ok if rebound_entry_ok and position_notional > 0: @@ -896,8 +900,12 @@ def _torch_forecast_exit_signal( expected_return = _safe_float(forecast.get("expected_return_percent"), 0.0) probability_up = _safe_float(forecast.get("probability_up"), 0.5) skill = _safe_float(forecast.get("skill"), 0.0) - min_edge = max(0.0, settings.time_series_min_edge_percent) - min_probability = _torch_min_probability(settings) + min_edge = max(0.0, _safe_float(forecast.get("calibrated_min_edge_percent"), settings.time_series_min_edge_percent)) + min_probability = _clamp( + _safe_float(forecast.get("calibrated_min_probability_up"), _torch_min_probability(settings)), + 0.5, + 0.95, + ) estimated_exit_net_percent = _estimated_exit_net_percent(position, price, settings) min_exit_net_percent = _min_exit_net_percent(settings) entry_path = str(position.entry_diagnostics.get("entry_path", "")) diff --git a/crypto_spot_bot/time_series.py b/crypto_spot_bot/time_series.py index aa2065f..43e9e9b 100644 --- a/crypto_spot_bot/time_series.py +++ b/crypto_spot_bot/time_series.py @@ -160,6 +160,9 @@ class TimeSeriesForecast: model_created_at: str = "" model_age_hours: float | None = None model_fresh: bool = False + calibrated_min_edge_percent: float = 0.0 + calibrated_min_probability_up: float = 0.0 + calibrated_min_confidence: float = 0.0 def as_dict(self) -> dict[str, Any]: return asdict(self) @@ -196,8 +199,20 @@ class TimeSeriesForecaster: artifact, self.settings.time_series_model_max_age_hours, ) - quality_gate = self._load_quality_gate() + calibration = self._load_quality_gate() + quality_gate = ( + calibration.get("validation") + if isinstance(calibration.get("validation"), dict) + else calibration + ) quality_gate_passed = _quality_gate_passed(quality_gate) + calibrated = _calibrated_thresholds( + calibration, + symbol, + edge=self.settings.time_series_min_edge_percent, + probability=self.settings.time_series_min_probability_up, + confidence=self.settings.time_series_min_confidence, + ) entry = _torch_recurrent_entry(symbol, artifact) model = _torch_recurrent_model_name(symbol, artifact) clip = _clamp(_float_entry(entry or {}, "clip", 8.0), 1.0, 50.0) @@ -249,7 +264,7 @@ class TimeSeriesForecaster: q90_percent = (math.exp(float(selected.get("q90", expected_return))) - 1) * 100 skill = _clamp(_float_entry(entry, "skill", 0.0), -1.0, 1.0) horizon = int(selected.get("horizon", _entry_horizon(entry, self.settings.time_series_forecast_horizon))) - min_edge = max(0.0, self.settings.time_series_min_edge_percent) + min_edge = calibrated["edge"] confidence_adjustment = _confidence_adjustment( expected_return_percent=expected_return_percent, probability_up=probability_up, @@ -299,6 +314,9 @@ class TimeSeriesForecaster: model_created_at=model_created_at, model_age_hours=model_age_hours, model_fresh=model_fresh, + calibrated_min_edge_percent=calibrated["edge"], + calibrated_min_probability_up=calibrated["probability"], + calibrated_min_confidence=calibrated["confidence"], ) direct_horizon = _is_direct_horizon(entry) @@ -318,7 +336,7 @@ class TimeSeriesForecaster: expected_return_percent = (math.exp(expected_return) - 1) * 100 probability_up = _normal_cdf(expected_return / max(uncertainty, 1e-9)) skill = _clamp(_float_entry(entry, "skill", 0.0), -1.0, 1.0) - min_edge = max(0.0, self.settings.time_series_min_edge_percent) + min_edge = calibrated["edge"] confidence_adjustment = _confidence_adjustment( expected_return_percent=expected_return_percent, probability_up=probability_up, @@ -364,6 +382,9 @@ class TimeSeriesForecaster: model_created_at=model_created_at, model_age_hours=model_age_hours, model_fresh=model_fresh, + calibrated_min_edge_percent=calibrated["edge"], + calibrated_min_probability_up=calibrated["probability"], + calibrated_min_confidence=calibrated["confidence"], ) def _load_lstm_artifact(self) -> dict[str, Any]: @@ -400,8 +421,7 @@ class TimeSeriesForecaster: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): data = {} - validation = data.get("validation") if isinstance(data, dict) else {} - self._quality_gate = validation if isinstance(validation, dict) else {} + self._quality_gate = data if isinstance(data, dict) else {} self._calibration_mtime = stat.st_mtime return self._quality_gate @@ -443,6 +463,9 @@ def _empty_forecast(enabled: bool, reason: str) -> TimeSeriesForecast: def _quality_gate_passed(quality_gate: dict[str, Any]) -> bool | None: if not quality_gate: return None + validation = quality_gate.get("validation") + if isinstance(validation, dict): + return _quality_gate_passed(validation) if "passed" in quality_gate: return bool(quality_gate.get("passed")) status = str(quality_gate.get("status", "")).strip().lower() @@ -453,6 +476,26 @@ def _quality_gate_passed(quality_gate: dict[str, Any]) -> bool | None: return None +def _calibrated_thresholds( + calibration: dict[str, Any], + symbol: str | None, + *, + edge: float, + probability: float, + confidence: float, +) -> dict[str, float]: + recommended = calibration.get("recommended") if isinstance(calibration, dict) else None + per_symbol = calibration.get("symbol_recommendations") if isinstance(calibration, dict) else None + if symbol and isinstance(per_symbol, dict) and isinstance(per_symbol.get(symbol.upper()), dict): + recommended = per_symbol[symbol.upper()] + row = recommended if isinstance(recommended, dict) else {} + return { + "edge": max(0.0, float(row.get("edge", edge) or edge)), + "probability": _clamp(float(row.get("probability", probability) or probability), 0.5, 0.95), + "confidence": _clamp(float(row.get("confidence", confidence) or confidence), 0.0, 1.0), + } + + def _model_freshness(artifact: dict[str, Any], max_age_hours: float) -> tuple[str, float | None, bool]: raw = str(artifact.get("created_at", "")).strip() if isinstance(artifact, dict) else "" if not raw: @@ -525,6 +568,8 @@ def _feature_context( def _feature_value(name: str, candles: list[Candle], index: int, candle: Candle, context: dict[str, Any]) -> float: close = max(float(candle.close), 1e-12) previous = candles[index - 1] if index >= 1 else candle + if name.startswith("symbol_is_"): + return 1.0 if context.get("symbol") == name.removeprefix("symbol_is_").upper() else 0.0 if name == "return_1": return _log_change(candle.close, previous.close) if name == "return_3": @@ -990,6 +1035,31 @@ def _torch_recurrent_predict( model_name = _torch_recurrent_model_name(symbol, artifact) if not entry or not model_name: return None + ensemble_members = entry.get("ensemble_members") + if isinstance(ensemble_members, list) and ensemble_members: + predictions: list[float | dict[str, Any]] = [] + for member in ensemble_members: + if not isinstance(member, dict): + continue + member_entry = {**entry, **member} + member_entry.pop("ensemble_members", None) + member_entry.pop("ensemble_size", None) + member_artifact: dict[str, Any] = {"type": "pytorch_recurrent_forecaster"} + if symbol: + member_artifact["symbols"] = {symbol.upper(): member_entry} + else: + member_artifact["default"] = member_entry + prediction = _torch_recurrent_predict( + returns, + symbol, + member_artifact, + feature_rows=feature_rows, + closes=closes, + candles=candles, + ) + if isinstance(prediction, (int, float, dict)): + predictions.append(prediction) + return _average_ensemble_predictions(predictions) lookback = int(_clamp(_float_entry(entry, "lookback", 0.0), 4.0, 512.0)) hidden_size = int(_clamp(_float_entry(entry, "hidden_size", 0.0), 1.0, 512.0)) num_layers = int(_clamp(_float_entry(entry, "num_layers", 1.0), 1.0, 8.0)) @@ -1050,6 +1120,33 @@ def _torch_recurrent_predict( return _clamp(prediction, -cap, cap) +def _average_ensemble_predictions(predictions: list[float | dict[str, Any]]) -> float | dict[str, Any] | None: + if not predictions: + return None + numeric = [float(value) for value in predictions if isinstance(value, (int, float))] + if numeric: + return sum(numeric) / len(numeric) + mappings = [value for value in predictions if isinstance(value, dict)] + if not mappings: + return None + first = mappings[0] + output: dict[str, Any] = {} + for key, value in first.items(): + if key == "horizons" and isinstance(value, dict): + horizons: dict[str, Any] = {} + for horizon, row in value.items(): + rows = [item.get("horizons", {}).get(horizon) for item in mappings] + rows = [item for item in rows if isinstance(item, dict)] + if rows: + horizons[horizon] = _average_ensemble_predictions(rows) + output[key] = horizons + continue + values = [item.get(key) for item in mappings] + finite = [float(item) for item in values if isinstance(item, (int, float)) and math.isfinite(float(item))] + output[key] = sum(finite) / len(finite) if finite else value + return output + + def _torch_head_outputs(context: list[float], entry: dict[str, Any], hidden_size: int) -> list[float]: context = _apply_context_norm(context, entry) raw_weight = entry.get("head_weight") diff --git a/crypto_spot_bot/training_coordination.py b/crypto_spot_bot/training_coordination.py index 82e84dc..4969020 100644 --- a/crypto_spot_bot/training_coordination.py +++ b/crypto_spot_bot/training_coordination.py @@ -215,6 +215,10 @@ class TrainingCoordinator: job["message"] = str(payload.get("message") or "") if isinstance(payload.get("summary"), dict): job["summary"] = payload["summary"] + if isinstance(payload["summary"].get("accepted"), bool): + job["model_decision"] = ( + "accepted" if payload["summary"]["accepted"] else "rejected" + ) self._save_state(state) return {"ok": True, "job": job, "status": self._public_status(state)} @@ -294,10 +298,11 @@ class TrainingCoordinator: os.replace(tmp, self.state_path) def _worker_from_payload(self, payload: dict[str, Any]) -> dict[str, Any]: + worker_id = str(payload.get("worker_id") or payload.get("id") or "windows-training-host").strip() 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"), + "id": worker_id, + "name": str(payload.get("name") or worker_id).strip(), + "path": str(payload.get("path") or "").strip(), "version": str(payload.get("version") or "1"), "last_seen_at": _now(), } diff --git a/tests/test_calibrate_thresholds.py b/tests/test_calibrate_thresholds.py new file mode 100644 index 0000000..0583ff8 --- /dev/null +++ b/tests/test_calibrate_thresholds.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +from tools.calibrate_torch_thresholds import ( + CalibrationResult, + ForecastRecord, + _apply_platt_calibration, + _choose_recommendation, + _fit_platt_calibration, + _entry_validation_skill, +) + + +def _result(*, trades: int, average: float, total: float, profit_factor: float) -> CalibrationResult: + return CalibrationResult( + edge=0.05, + probability=0.52, + confidence=0.4, + trades=trades, + wins=max(0, trades // 2), + win_rate=0.5, + total_net_percent=total, + average_net_percent=average, + max_drawdown_percent=1.0, + profit_factor=profit_factor, + score=1.0, + ) + + +def _record(index: int, probability: float, future: float) -> ForecastRecord: + return ForecastRecord( + symbol="BTCUSDT", + index=index, + timestamp=index, + close=100.0, + high=101.0, + low=99.0, + next_open=100.0, + next_timestamp=index + 1, + atr=1.0, + expected_percent=0.1, + probability_up=probability, + confidence=0.5, + skill=0.1, + q50_percent=0.1, + block_entry=False, + future_net_percent=future, + benchmark_entry=False, + benchmark_exit=False, + ) + + +def test_calibration_does_not_fallback_to_too_few_trades() -> None: + selected = _choose_recommendation( + [_result(trades=1, average=2.0, total=2.0, profit_factor=999.0)], + min_trades=30, + ) + + assert selected is None + + +def test_calibration_selects_only_viable_result() -> None: + viable = _result(trades=30, average=0.2, total=6.0, profit_factor=1.4) + + assert _choose_recommendation([viable], min_trades=30) is viable + + +def test_platt_calibration_learns_probability_direction_from_train_records() -> None: + records = [ + _record(index, 0.8 if index % 2 else 0.2, -1.0 if index % 2 else 1.0) + for index in range(100) + ] + + calibration = _fit_platt_calibration(records) + calibrated = _apply_platt_calibration( + [_record(101, 0.8, -1.0), _record(102, 0.2, 1.0)], + calibration, + ) + + assert calibration["slope"] < 0 + assert calibrated[0].probability_up < calibrated[1].probability_up + + +def test_entry_quality_never_falls_back_to_holdout_skill() -> None: + entry = {"validation_skill": 0.12, "skill": 0.99, "holdout_skill": 0.99} + + assert _entry_validation_skill(entry) == 0.12 + assert _entry_validation_skill({"skill": 0.99, "holdout_skill": 0.99}) == 0.0 diff --git a/tests/test_time_series.py b/tests/test_time_series.py index 9c0a516..10a1657 100644 --- a/tests/test_time_series.py +++ b/tests/test_time_series.py @@ -304,6 +304,60 @@ def test_time_series_forecaster_attaches_quality_gate(make_settings, tmp_path) - assert forecast.quality_gate["status"] == "fail" +def test_time_series_forecaster_uses_symbol_calibration(make_settings, tmp_path) -> None: + artifact_path = tmp_path / "lstm_forecaster.json" + _write_torch_gru_artifact(artifact_path, head_bias=0.2) + (tmp_path / "torch_threshold_calibration.json").write_text( + json.dumps( + { + "validation": {"status": "pass", "passed": True}, + "recommended": {"edge": 0.08, "probability": 0.52, "confidence": 0.4}, + "symbol_recommendations": { + "BTCUSDT": {"edge": 0.03, "probability": 0.55, "confidence": 0.45} + }, + } + ), + encoding="utf-8", + ) + settings = make_settings( + tmp_path, + time_series_lstm_model_path=artifact_path, + time_series_forecast_horizon=1, + ) + + forecast = TimeSeriesForecaster(settings).forecast( + _candles_from_returns([0.0001] * 140), symbol="BTCUSDT" + ) + + assert forecast.calibrated_min_edge_percent == 0.03 + assert forecast.calibrated_min_probability_up == 0.55 + assert forecast.calibrated_min_confidence == 0.45 + + +def test_time_series_forecaster_averages_ensemble_members(make_settings, tmp_path) -> None: + artifact_path = tmp_path / "lstm_forecaster.json" + _write_torch_gru_artifact(artifact_path, head_bias=0.9) + artifact = json.loads(artifact_path.read_text(encoding="utf-8")) + entry = artifact["symbols"]["BTCUSDT"] + entry["ensemble_members"] = [ + {"state_dict": entry["state_dict"], "head_weight": [0.0, 0.0], "head_bias": bias} + for bias in (0.1, 0.3) + ] + artifact_path.write_text(json.dumps(artifact), encoding="utf-8") + settings = make_settings( + tmp_path, + time_series_lstm_model_path=artifact_path, + time_series_forecast_horizon=1, + ) + + forecast = TimeSeriesForecaster(settings).forecast( + _candles_from_returns([0.0001] * 140), symbol="BTCUSDT" + ) + + assert forecast.usable is True + assert 0.015 <= forecast.expected_return_percent <= 0.025 + + def test_time_series_forecaster_reads_multifeature_direct_horizon_artifact(make_settings, tmp_path) -> None: artifact_path = tmp_path / "lstm_forecaster.json" _write_multifeature_torch_gru_artifact(artifact_path, head_bias=0.2) diff --git a/tests/test_training_coordination.py b/tests/test_training_coordination.py index 1686a0b..264c4f7 100644 --- a/tests/test_training_coordination.py +++ b/tests/test_training_coordination.py @@ -48,6 +48,42 @@ def test_training_coordinator_preserves_boolean_resume_candidate_parameter(tmp_p assert requested["job"]["parameters"] == {"resume_candidate": True} +def test_training_coordinator_reports_worker_identity_from_heartbeat(tmp_path) -> None: + coordinator = TrainingCoordinator(tmp_path) + + heartbeat = coordinator.heartbeat( + { + "worker_id": "SEVENHILL:G:\\Repos\\TradeBot", + "name": "SEVENHILL", + "path": "G:\\Repos\\TradeBot", + } + ) + + assert heartbeat["worker"]["name"] == "SEVENHILL" + assert heartbeat["worker"]["path"] == "G:\\Repos\\TradeBot" + assert heartbeat["status"]["worker"] == heartbeat["worker"] + + +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"}) + + completed = coordinator.complete( + job["id"], + { + "success": True, + "message": "training completed; candidate rejected by quality gate", + "summary": {"accepted": False, "reason": "candidate_failed_honest_validation"}, + }, + ) + + assert completed["job"]["status"] == "completed" + assert completed["job"]["phase"] == "completed" + assert completed["job"]["progress_percent"] == 100 + assert completed["job"]["model_decision"] == "rejected" + + def test_training_coordinator_accepts_chunked_artifact_upload(tmp_path) -> None: coordinator = TrainingCoordinator(tmp_path) job = coordinator.request_retrain({"source": "test"})["job"] diff --git a/tools/calibrate_torch_thresholds.py b/tools/calibrate_torch_thresholds.py index 5dda0e1..62c45bf 100644 --- a/tools/calibrate_torch_thresholds.py +++ b/tools/calibrate_torch_thresholds.py @@ -6,7 +6,7 @@ import json import math import sys import time -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from typing import Any @@ -130,13 +130,15 @@ def main() -> None: if not records: raise SystemExit("No forecast records could be built for calibration.") - results = _calibrate( + results = _calibrate_strategy( records, edges=_float_grid(args.edge_grid), probabilities=_float_grid(args.probability_grid), confidences=_float_grid(args.confidence_grid), min_trades=args.min_trades, horizon=horizon, + round_trip_cost=round_trip_cost, + settings=settings, ) if not results: raise SystemExit("No calibration result produced trades. Use wider grids or more history.") @@ -156,6 +158,33 @@ def main() -> None: round_trip_cost=round_trip_cost, settings=settings, ) + symbol_recommendations: dict[str, dict[str, Any]] = {} + for symbol in symbols: + symbol_records = [record for record in records if record.symbol == symbol] + symbol_results = _calibrate_strategy( + symbol_records, + edges=_float_grid(args.edge_grid), + probabilities=_float_grid(args.probability_grid), + confidences=_float_grid(args.confidence_grid), + min_trades=max(3, min(args.min_trades, len(symbol_records) // 8)), + horizon=horizon, + round_trip_cost=round_trip_cost, + settings=settings, + ) + symbol_selected = _choose_recommendation( + symbol_results, + min_trades=max(3, min(args.min_trades, len(symbol_records) // 8)), + ) if symbol_results else None + if symbol_selected is not None: + symbol_recommendations[symbol] = _result_dict(symbol_selected) + calibration_insufficient = recommended is None + if recommended is None: + recommended = _empty_recommendation( + _float_grid(args.edge_grid), + _float_grid(args.probability_grid), + _float_grid(args.confidence_grid), + ) + full_backtest = {**_stats([]), "trades_detail": [], "symbol_breakdown": []} print("\nRECOMMENDED") print(_result_line(recommended)) print("\nFULL_REPLAY") @@ -188,6 +217,16 @@ def main() -> None: min_profit_factor=args.min_oos_profit_factor, min_benchmark_edge=args.min_benchmark_edge_percent, ) + deployment_recommended = recommended + deployment_symbol_recommendations = symbol_recommendations + if walk_forward.get("folds"): + last_fold = walk_forward["folds"][-1] + fold_thresholds = last_fold.get("thresholds") + if isinstance(fold_thresholds, dict): + deployment_recommended = _result_from_dict(fold_thresholds) + fold_symbols = last_fold.get("symbol_thresholds") + if isinstance(fold_symbols, dict): + deployment_symbol_recommendations = fold_symbols print("\nWALK_FORWARD") print(json.dumps(walk_forward["summary"], ensure_ascii=False, sort_keys=True)) print("\nBENCHMARK") @@ -206,7 +245,9 @@ def main() -> None: "artifact_sha256": artifact_sha256, "artifact": _artifact_summary(artifact), "records_by_symbol": per_symbol_counts, - "recommended": _result_dict(recommended), + "recommended": _result_dict(deployment_recommended), + "calibration_insufficient": calibration_insufficient, + "symbol_recommendations": deployment_symbol_recommendations, "full_replay": full_backtest, "walk_forward": walk_forward, "benchmark": benchmark, @@ -306,7 +347,9 @@ def _forecast_records( return batched_records records: list[ForecastRecord] = [] - skill = float(entry.get("skill", 0.0) or 0.0) + # Entry eligibility may use validation-derived quality only. Holdout metrics + # belong exclusively to the final quality gate and cannot influence replay. + skill = _entry_validation_skill(entry) for index in range(start, max(start, end)): prediction = _torch_recurrent_predict( _log_returns(closes[: index + 1]), @@ -394,7 +437,7 @@ def _batch_forecast_records( return [] records: list[ForecastRecord] = [] - skill = float(entry.get("skill", 0.0) or 0.0) + skill = _entry_validation_skill(entry) model.eval() with torch.no_grad(): for offset in range(0, len(indices), max(1, batch_size)): @@ -460,6 +503,10 @@ def _batch_forecast_records( def _build_torch_model(entry: dict[str, Any], model_name: str) -> Any | None: + if isinstance(entry.get("ensemble_members"), list) and entry["ensemble_members"]: + # Ensemble inference is handled by the shared pure-Python runtime so + # calibration and production use the exact same averaging path. + return None if torch is None or RecurrentReturnModel is None: return None architecture = "lstm" if model_name == "torch_lstm" else "gru" if model_name == "torch_gru" else "" @@ -590,6 +637,7 @@ def _full_backtest( round_trip_cost: float, settings: Any, detail_limit: int = 50, + symbol_thresholds: dict[str, CalibrationResult] | None = None, ) -> dict[str, Any]: positions: dict[str, dict[str, Any]] = {} trades: list[float] = [] @@ -600,6 +648,7 @@ def _full_backtest( stop_loss_exit_enabled = bool(getattr(settings, "stop_loss_exit_enabled", True)) atr_multiplier = max(0.5, min(10.0, float(settings.atr_trailing_multiplier))) for record in sorted(records, key=lambda item: (item.timestamp, item.symbol)): + active_thresholds = (symbol_thresholds or {}).get(record.symbol, thresholds) position = positions.get(record.symbol) if position is not None: position["highest"] = max(position["highest"], record.high) @@ -616,8 +665,8 @@ def _full_backtest( and (stop_loss_exit_enabled or atr_stop_level > position["entry_price"]) ) weak_forecast = ( - record.expected_percent < thresholds.edge - or record.probability_up < thresholds.probability + record.expected_percent < active_thresholds.edge + or record.probability_up < active_thresholds.probability or record.skill <= 0.0 ) exit_reason = "" @@ -633,7 +682,7 @@ def _full_backtest( elif atr_stop: exit_reason = "atr_trailing_stop" exit_price = float(atr_stop_level) - elif (record.expected_percent <= 0.0 or record.probability_up <= 0.50 or _candidate_blocks(record, thresholds.edge)): + elif (record.expected_percent <= 0.0 or record.probability_up <= 0.50 or _candidate_blocks(record, active_thresholds.edge)): exit_reason = "forecast_negative" elif weak_forecast and net_percent >= 0: exit_reason = "forecast_weak_profit_lock" @@ -659,7 +708,7 @@ def _full_backtest( if record.symbol in positions: continue - if _candidate_allows(record, thresholds.edge, thresholds.probability, thresholds.confidence): + if _candidate_allows(record, active_thresholds.edge, active_thresholds.probability, active_thresholds.confidence): positions[record.symbol] = { "entry_price": record.next_open, "entry_index": record.index + 1, @@ -815,24 +864,49 @@ def _walk_forward( test_end = timestamps[(fold + 1) * fold_size - 1] if fold < folds - 1 else timestamps[-1] train = [record for record in ordered if record.timestamp < test_start] test = [record for record in ordered if test_start <= record.timestamp <= test_end] - train_results = _calibrate( - train, + probability_calibration = _fit_platt_calibration(train) + calibrated_train = _apply_platt_calibration(train, probability_calibration) + calibrated_test = _apply_platt_calibration(test, probability_calibration) + train_results = _calibrate_strategy( + calibrated_train, edges=edges, probabilities=probabilities, confidences=confidences, min_trades=max(4, min_trades // 2), horizon=horizon, + round_trip_cost=round_trip_cost, + settings=settings, ) if not train_results: continue selected = _choose_recommendation(train_results, min_trades=max(4, min_trades // 2)) + if selected is None: + continue + symbol_thresholds: dict[str, CalibrationResult] = {} + train_symbols = sorted({record.symbol for record in calibrated_train}) + symbol_min_trades = max(3, min_trades // max(2, len(train_symbols) * 2)) + for symbol in train_symbols: + symbol_results = _calibrate_strategy( + [record for record in calibrated_train if record.symbol == symbol], + edges=edges, + probabilities=probabilities, + confidences=confidences, + min_trades=symbol_min_trades, + horizon=horizon, + round_trip_cost=round_trip_cost, + settings=settings, + ) + symbol_selected = _choose_recommendation(symbol_results, min_trades=symbol_min_trades) if symbol_results else None + if symbol_selected is not None: + symbol_thresholds[symbol] = symbol_selected test_backtest = _full_backtest( - test, + calibrated_test, selected, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings, detail_limit=0, + symbol_thresholds=symbol_thresholds, ) test_rows = test_backtest.get("trades_detail", []) test_trades = [float(row.get("net_percent", 0.0) or 0.0) for row in test_rows if isinstance(row, dict)] @@ -844,6 +918,10 @@ def _walk_forward( "train_records": len(train), "test_records": len(test), "thresholds": _result_dict(selected), + "symbol_thresholds": { + symbol: _result_dict(value) for symbol, value in symbol_thresholds.items() + }, + "probability_calibration": probability_calibration, "test": {key: value for key, value in test_backtest.items() if key != "trades_detail"}, } ) @@ -980,6 +1058,11 @@ def _candidate_blocks(record: ForecastRecord, edge: float) -> bool: ) +def _entry_validation_skill(entry: dict[str, Any]) -> float: + value = entry.get("validation_skill") + return float(value) if isinstance(value, (int, float)) and math.isfinite(float(value)) else 0.0 + + def _candidate_allows(record: ForecastRecord, edge: float, probability: float, confidence: float) -> bool: dynamic_confidence = _forecast_confidence(record.expected_percent, record.probability_up, record.skill, edge) return ( @@ -1146,6 +1229,101 @@ def _calibrate( return results +def _calibrate_strategy( + records: list[ForecastRecord], + *, + edges: list[float], + probabilities: list[float], + confidences: list[float], + min_trades: int, + horizon: int, + round_trip_cost: float, + settings: Any, +) -> list[CalibrationResult]: + results: list[CalibrationResult] = [] + for edge in edges: + for probability in probabilities: + for confidence in confidences: + thresholds = CalibrationResult( + edge=edge, + probability=probability, + confidence=confidence, + trades=0, + wins=0, + win_rate=0.0, + total_net_percent=0.0, + average_net_percent=0.0, + max_drawdown_percent=0.0, + profit_factor=0.0, + score=0.0, + ) + replay = _full_backtest( + records, + thresholds, + horizon=horizon, + round_trip_cost=round_trip_cost, + settings=settings, + detail_limit=0, + ) + trades = int(replay.get("trades", 0) or 0) + if trades <= 0: + continue + wins = int(replay.get("wins", 0) or 0) + total = float(replay.get("total_net_percent", 0.0) or 0.0) + average = float(replay.get("avg_net_percent", 0.0) or 0.0) + drawdown = float(replay.get("max_drawdown_percent", 0.0) or 0.0) + profit_factor = float(replay.get("profit_factor", 0.0) or 0.0) + trade_factor = min(1.0, trades / max(1, min_trades)) + score = ( + average * trade_factor + + total * 0.015 + - drawdown * 0.03 + + (wins / trades) * 0.04 + ) + results.append( + CalibrationResult( + edge=edge, + probability=probability, + confidence=confidence, + trades=trades, + wins=wins, + win_rate=wins / trades, + total_net_percent=total, + average_net_percent=average, + max_drawdown_percent=drawdown, + profit_factor=profit_factor, + score=score, + ) + ) + results.sort( + key=lambda item: ( + item.score, + item.average_net_percent, + item.total_net_percent, + item.profit_factor, + item.trades, + ), + reverse=True, + ) + return results + + +def _result_from_dict(value: dict[str, Any]) -> CalibrationResult: + return CalibrationResult( + edge=float(value.get("edge", 0.1) or 0.1), + probability=float(value.get("probability", 0.7) or 0.7), + confidence=float(value.get("confidence", 0.4) or 0.4), + trades=int(value.get("trades", 0) or 0), + wins=int(value.get("wins", 0) or 0), + win_rate=float(value.get("win_rate", 0.0) or 0.0), + total_net_percent=float(value.get("total_net_percent", 0.0) or 0.0), + average_net_percent=float(value.get("average_net_percent", 0.0) or 0.0), + max_drawdown_percent=float(value.get("max_drawdown_percent", 0.0) or 0.0), + profit_factor=float(value.get("profit_factor", 0.0) or 0.0), + score=float(value.get("score", 0.0) or 0.0), + ) + + def _selected_trades( records: list[ForecastRecord], edge: float, @@ -1164,7 +1342,7 @@ def _selected_trades( return trades -def _choose_recommendation(results: list[CalibrationResult], *, min_trades: int) -> CalibrationResult: +def _choose_recommendation(results: list[CalibrationResult], *, min_trades: int) -> CalibrationResult | None: viable = [ result for result in results @@ -1173,7 +1351,67 @@ def _choose_recommendation(results: list[CalibrationResult], *, min_trades: int) and result.total_net_percent > 0 and result.profit_factor >= 1.05 ] - return viable[0] if viable else results[0] + return viable[0] if viable else None + + +def _empty_recommendation( + edges: list[float], probabilities: list[float], confidences: list[float] +) -> CalibrationResult: + return CalibrationResult( + edge=max(edges or [1.0]), + probability=max(probabilities or [0.95]), + confidence=max(confidences or [1.0]), + trades=0, + wins=0, + win_rate=0.0, + total_net_percent=0.0, + average_net_percent=0.0, + max_drawdown_percent=0.0, + profit_factor=0.0, + score=-1.0, + ) + + +def _fit_platt_calibration(records: list[ForecastRecord]) -> dict[str, float]: + samples = [ + ( + math.log(_clamp(record.probability_up, 1e-5, 1.0 - 1e-5) / (1.0 - _clamp(record.probability_up, 1e-5, 1.0 - 1e-5))), + 1.0 if record.future_net_percent > 0 else 0.0, + ) + for record in records + ] + if len(samples) < 30: + return {"slope": 1.0, "intercept": 0.0, "samples": float(len(samples))} + slope = 1.0 + intercept = 0.0 + learning_rate = 0.05 + for _ in range(300): + grad_slope = 0.0 + grad_intercept = 0.0 + for logit, target in samples: + probability = 1.0 / (1.0 + math.exp(-_clamp(slope * logit + intercept, -30.0, 30.0))) + error = probability - target + grad_slope += error * logit + grad_intercept += error + grad_slope = grad_slope / len(samples) + 0.001 * (slope - 1.0) + grad_intercept /= len(samples) + slope -= learning_rate * grad_slope + intercept -= learning_rate * grad_intercept + return {"slope": round(slope, 8), "intercept": round(intercept, 8), "samples": float(len(samples))} + + +def _apply_platt_calibration( + records: list[ForecastRecord], calibration: dict[str, float] +) -> list[ForecastRecord]: + slope = float(calibration.get("slope", 1.0)) + intercept = float(calibration.get("intercept", 0.0)) + output: list[ForecastRecord] = [] + for record in records: + probability = _clamp(record.probability_up, 1e-5, 1.0 - 1e-5) + logit = math.log(probability / (1.0 - probability)) + calibrated = 1.0 / (1.0 + math.exp(-_clamp(slope * logit + intercept, -30.0, 30.0))) + output.append(replace(record, probability_up=calibrated)) + return output def _choose_replay_recommendation( @@ -1185,8 +1423,10 @@ def _choose_replay_recommendation( horizon: int, round_trip_cost: float, settings: Any, -) -> tuple[CalibrationResult, dict[str, Any]]: +) -> tuple[CalibrationResult | None, dict[str, Any]]: fallback = _choose_recommendation(results, min_trades=min_trades) + if fallback is None: + return None, {**_stats([]), "trades_detail": [], "symbol_breakdown": []} fallback_replay = _full_backtest(records, fallback, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings) if min_full_replay_trades <= 0: return fallback, fallback_replay @@ -1207,7 +1447,7 @@ def _choose_replay_recommendation( viable.append((result, replay)) if not viable: - return fallback, fallback_replay + return None, fallback_replay viable.sort( key=lambda item: ( item[0].score, diff --git a/tools/install_windows_training_agent.ps1 b/tools/install_windows_training_agent.ps1 index 5e83f0c..162b616 100644 --- a/tools/install_windows_training_agent.ps1 +++ b/tools/install_windows_training_agent.ps1 @@ -6,6 +6,7 @@ param( [int]$PollSeconds = 10, [int]$WatchdogMinutes = 5, [string]$RepoRoot = "", + [string]$CredentialPath = "", [switch]$StartNow, [switch]$KeepLegacyRetrainer ) @@ -15,105 +16,153 @@ $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" +if (-not $CredentialPath) { + $CredentialPath = Join-Path $env:LOCALAPPDATA "TradeBot\training-agent.token" } -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." -} - -function Resolve-WindowlessPython { - $python = Resolve-Python - $pythonw = Join-Path (Split-Path -Parent $python) "pythonw.exe" - if (Test-Path $pythonw) { - return $pythonw - } - return $python +$runner = Join-Path $RepoRoot "tools\run_windows_training_agent.ps1" +if (-not (Test-Path -LiteralPath $runner)) { + throw "Windows training agent runner not found: $runner" } +$credentialDirectory = Split-Path -Parent $CredentialPath +New-Item -ItemType Directory -Path $credentialDirectory -Force | Out-Null if ($ApiAuth) { - [Environment]::SetEnvironmentVariable("TRADEBOT_API_AUTH", $ApiAuth, "User") - $env:TRADEBOT_API_AUTH = $ApiAuth + $ApiAuth.Trim() | + ConvertTo-SecureString -AsPlainText -Force | + ConvertFrom-SecureString | + Set-Content -LiteralPath $CredentialPath -Encoding UTF8 } +if (-not (Test-Path -LiteralPath $CredentialPath)) { + throw "ApiAuth is required for the first installation." +} + +# Remove the legacy plaintext secret from the user environment. The new runner +# decrypts the DPAPI-protected credential only inside the agent process tree. +[Environment]::SetEnvironmentVariable("TRADEBOT_API_AUTH", $null, "User") +Remove-Item Env:TRADEBOT_API_AUTH -ErrorAction SilentlyContinue [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'." + try { + $legacyTask = Get-ScheduledTask -TaskName $legacyName -ErrorAction SilentlyContinue + if ($legacyTask) { + Unregister-ScheduledTask -TaskName $legacyName -Confirm:$false + Write-Host "Removed legacy scheduled task '$legacyName'." + } + } + catch { + Write-Warning "Could not remove legacy scheduled task '$legacyName': $($_.Exception.Message)" } } } -$python = Resolve-WindowlessPython $currentUser = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name -$arguments = @( - "-u", - "`"$Agent`"", - "--repo-root", "`"$RepoRoot`"", - "--api-base-url", "`"$ApiBaseUrl`"", - "--poll-seconds", $PollSeconds.ToString() +$principal = New-Object System.Security.Principal.WindowsPrincipal( + [System.Security.Principal.WindowsIdentity]::GetCurrent() +) +$isAdministrator = $principal.IsInRole([System.Security.Principal.WindowsBuiltInRole]::Administrator) +$powershell = (Get-Command powershell.exe -ErrorAction Stop).Source +$runnerArguments = @( + "-NoProfile", + "-WindowStyle", "Hidden", + "-ExecutionPolicy", "Bypass", + "-File", "`"$runner`"", + "-RepoRoot", "`"$RepoRoot`"", + "-ApiBaseUrl", "`"$ApiBaseUrl`"", + "-CredentialPath", "`"$CredentialPath`"", + "-WorkerName", "`"$env:COMPUTERNAME`"", + "-PollSeconds", $PollSeconds.ToString() ) -join " " -$action = New-ScheduledTaskAction -Execute $python -Argument $arguments -WorkingDirectory $RepoRoot -$trigger = @( - New-ScheduledTaskTrigger -AtLogOn -User $currentUser - New-ScheduledTaskTrigger -AtStartup - New-ScheduledTaskTrigger ` - -Once ` - -At (Get-Date).AddMinutes(1) ` - -RepetitionInterval (New-TimeSpan -Minutes $WatchdogMinutes) ` - -RepetitionDuration (New-TimeSpan -Days 3650) -) -$principal = New-ScheduledTaskPrincipal ` - -UserId $currentUser ` - -LogonType Interactive ` - -RunLevel Limited -$settings = New-ScheduledTaskSettingsSet ` - -StartWhenAvailable ` - -MultipleInstances IgnoreNew ` - -AllowStartIfOnBatteries ` - -DontStopIfGoingOnBatteries ` - -RestartCount 999 ` - -RestartInterval (New-TimeSpan -Minutes 1) ` - -ExecutionTimeLimit (New-TimeSpan -Days 30) +$startupShortcut = Join-Path ([Environment]::GetFolderPath("Startup")) "$TaskName.lnk" +$runKey = "HKCU:\Software\Microsoft\Windows\CurrentVersion\Run" +Remove-ItemProperty -Path $runKey -Name "TradeBotWindowsTrainingAgent" -ErrorAction SilentlyContinue -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 +$installMode = "startup shortcut" +if ($isAdministrator) { + if (Test-Path -LiteralPath $startupShortcut) { + Remove-Item -LiteralPath $startupShortcut -Force + } + $action = New-ScheduledTaskAction -Execute $powershell -Argument $runnerArguments -WorkingDirectory $RepoRoot + $trigger = @( + New-ScheduledTaskTrigger -AtLogOn -User $currentUser + New-ScheduledTaskTrigger -AtStartup + New-ScheduledTaskTrigger ` + -Once ` + -At (Get-Date).AddMinutes(1) ` + -RepetitionInterval (New-TimeSpan -Minutes $WatchdogMinutes) ` + -RepetitionDuration (New-TimeSpan -Days 3650) + ) + $taskPrincipal = New-ScheduledTaskPrincipal ` + -UserId $currentUser ` + -LogonType Interactive ` + -RunLevel Limited + $settings = New-ScheduledTaskSettingsSet ` + -StartWhenAvailable ` + -MultipleInstances IgnoreNew ` + -AllowStartIfOnBatteries ` + -DontStopIfGoingOnBatteries ` + -RestartCount 999 ` + -RestartInterval (New-TimeSpan -Minutes 1) ` + -ExecutionTimeLimit (New-TimeSpan -Days 30) -if ($StartNow) { - Start-ScheduledTask -TaskName $TaskName + Register-ScheduledTask ` + -TaskName $TaskName ` + -Action $action ` + -Trigger $trigger ` + -Principal $taskPrincipal ` + -Settings $settings ` + -Description "Keeps the TradeBot Windows training agent online and polls the bot API for retrain jobs." ` + -Force | Out-Null + $installMode = "scheduled task" +} +else { + try { + $existingTask = Get-ScheduledTask -TaskName $TaskName -ErrorAction SilentlyContinue + if ($existingTask) { + Unregister-ScheduledTask -TaskName $TaskName -Confirm:$false + } + } + catch { + Write-Warning "Could not remove an existing elevated task: $($_.Exception.Message)" + } + + $shell = New-Object -ComObject WScript.Shell + $shortcut = $shell.CreateShortcut($startupShortcut) + $shortcut.TargetPath = $powershell + $shortcut.Arguments = $runnerArguments + $shortcut.WorkingDirectory = $RepoRoot + $shortcut.WindowStyle = 7 + $shortcut.Description = "TradeBot Windows Training Agent" + $shortcut.Save() } -Write-Host "Registered scheduled task '$TaskName' for Windows startup, logon, and watchdog restarts." +Get-CimInstance Win32_Process | + Where-Object { + $_.ProcessId -ne $PID -and + $_.CommandLine -and + ($_.CommandLine -match [regex]::Escape("windows_training_agent.py") -or + $_.CommandLine -match [regex]::Escape("run_windows_training_agent.ps1")) + } | + ForEach-Object { Stop-Process -Id $_.ProcessId -Force -ErrorAction SilentlyContinue } + +if ($StartNow) { + if ($installMode -eq "scheduled task") { + Start-ScheduledTask -TaskName $TaskName + } + else { + Start-Process ` + -FilePath $powershell ` + -ArgumentList $runnerArguments ` + -WorkingDirectory $RepoRoot ` + -WindowStyle Hidden | Out-Null + } +} + +Write-Host "Installed '$TaskName' using $installMode." Write-Host "Agent API: $ApiBaseUrl" -Write-Host "Agent script: $Agent" +Write-Host "Encrypted credential: $CredentialPath" +Write-Host "Agent runner: $runner" diff --git a/tools/run_torch_retrain.ps1 b/tools/run_torch_retrain.ps1 index ac843a3..6f4fd53 100644 --- a/tools/run_torch_retrain.ps1 +++ b/tools/run_torch_retrain.ps1 @@ -12,6 +12,10 @@ param( [string]$Features = "", [string]$ContextSymbols = "", [int]$Seed = 0, + [string]$EnsembleSeeds = "", + [int]$SelectionFolds = 0, + [double]$LearningRate = 0, + [double]$WeightDecay = 0, [int]$Epochs = 0, [int]$Patience = 0, [int]$HoldoutWindow = 0, @@ -132,18 +136,22 @@ function Sync-AcceptedArtifactsToPi { 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 { 3000 } + $Limit = if ($env:TORCH_RETRAIN_LIMIT) { [int]$env:TORCH_RETRAIN_LIMIT } else { 6000 } } -if (-not $Lookbacks) { $Lookbacks = if ($env:TORCH_RETRAIN_LOOKBACKS) { $env:TORCH_RETRAIN_LOOKBACKS } else { "64" } } +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 $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.15" } } -if ($Horizon -le 0 -and $env:TORCH_RETRAIN_HORIZON) { $Horizon = [int]$env:TORCH_RETRAIN_HORIZON } -if (-not $Horizons -and $env:TORCH_RETRAIN_HORIZONS) { $Horizons = $env:TORCH_RETRAIN_HORIZONS } +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 } } +if (-not $Horizons) { $Horizons = if ($env:TORCH_RETRAIN_HORIZONS) { $env:TORCH_RETRAIN_HORIZONS } else { "3,6,12,24" } } if (-not $Features -and $env:TORCH_RETRAIN_FEATURES) { $Features = $env:TORCH_RETRAIN_FEATURES } if (-not $ContextSymbols -and $env:TORCH_RETRAIN_CONTEXT_SYMBOLS) { $ContextSymbols = $env:TORCH_RETRAIN_CONTEXT_SYMBOLS } if ($Seed -le 0 -and $env:TORCH_RETRAIN_SEED) { $Seed = [int]$env:TORCH_RETRAIN_SEED } +if (-not $EnsembleSeeds) { $EnsembleSeeds = if ($env:TORCH_RETRAIN_ENSEMBLE_SEEDS) { $env:TORCH_RETRAIN_ENSEMBLE_SEEDS } else { "7,19,43" } } +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 ($Patience -le 0) { $Patience = if ($env:TORCH_RETRAIN_PATIENCE) { [int]$env:TORCH_RETRAIN_PATIENCE } else { 8 } } if ($HoldoutWindow -le 0) { $HoldoutWindow = if ($env:TORCH_RETRAIN_HOLDOUT_WINDOW) { [int]$env:TORCH_RETRAIN_HOLDOUT_WINDOW } else { 240 } } @@ -182,6 +190,10 @@ try { "--epochs", $Epochs.ToString(), "--patience", $Patience.ToString(), "--holdout-window", $HoldoutWindow.ToString(), + "--ensemble-seeds", $EnsembleSeeds, + "--selection-folds", $SelectionFolds.ToString(), + "--learning-rate", $LearningRate.ToString([Globalization.CultureInfo]::InvariantCulture), + "--weight-decay", $WeightDecay.ToString([Globalization.CultureInfo]::InvariantCulture), "--output", $CandidateFile ) if ($Symbols) { $trainerArgs += @("--symbols", $Symbols) } @@ -222,8 +234,8 @@ try { $calibrationBaseArgs = @( "-u", "tools\calibrate_torch_thresholds.py", - "--limit", "3000", - "--calibration-window", "1200", + "--limit", $Limit.ToString(), + "--calibration-window", ([Math]::Min(2400, [Math]::Max(1200, [int]($Limit / 2)))).ToString(), "--min-trades", "60", "--walk-forward-folds", "8", "--confidence-grid", "0.40" diff --git a/tools/run_windows_training_agent.ps1 b/tools/run_windows_training_agent.ps1 new file mode 100644 index 0000000..5ec375a --- /dev/null +++ b/tools/run_windows_training_agent.ps1 @@ -0,0 +1,82 @@ +[CmdletBinding()] +param( + [string]$ApiBaseUrl = "https://tb.kusoft.xyz", + [string]$RepoRoot = "", + [string]$CredentialPath = "", + [string]$WorkerName = $env:COMPUTERNAME, + [int]$PollSeconds = 10, + [int]$RestartDelaySeconds = 10 +) + +$ErrorActionPreference = "Stop" + +if (-not $RepoRoot) { + $RepoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path +} +if (-not $CredentialPath) { + $CredentialPath = Join-Path $env:LOCALAPPDATA "TradeBot\training-agent.token" +} + +$agent = Join-Path $RepoRoot "tools\windows_training_agent.py" +if (-not (Test-Path -LiteralPath $agent)) { + throw "Windows training agent not found: $agent" +} +if (-not (Test-Path -LiteralPath $CredentialPath)) { + throw "Encrypted training credential not found: $CredentialPath" +} + +function Resolve-Python { + $venvPython = Join-Path $RepoRoot ".venv\Scripts\python.exe" + if (Test-Path -LiteralPath $venvPython) { + return $venvPython + } + + $userPython = Join-Path $env:LOCALAPPDATA "Programs\TradeBotPython312\python.exe" + if (Test-Path -LiteralPath $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." +} + +$createdNew = $false +$mutex = [System.Threading.Mutex]::new($false, "Local\TradeBotWindowsTrainingAgent", [ref]$createdNew) +if (-not $createdNew) { + $mutex.Dispose() + exit 0 +} + +$encryptedToken = (Get-Content -LiteralPath $CredentialPath -Raw -Encoding UTF8).Trim() +$secureToken = $encryptedToken | ConvertTo-SecureString +$tokenPointer = [Runtime.InteropServices.Marshal]::SecureStringToBSTR($secureToken) +try { + $env:TRADEBOT_API_AUTH = [Runtime.InteropServices.Marshal]::PtrToStringBSTR($tokenPointer) + $python = Resolve-Python + $workerId = "${WorkerName}:$RepoRoot" + $arguments = @( + "-u", + $agent, + "--repo-root", $RepoRoot, + "--api-base-url", $ApiBaseUrl, + "--worker-id", $workerId, + "--worker-name", $WorkerName, + "--poll-seconds", [Math]::Max(5, $PollSeconds).ToString() + ) + + while ($true) { + & $python @arguments + Start-Sleep -Seconds ([Math]::Max(5, $RestartDelaySeconds)) + } +} +finally { + $env:TRADEBOT_API_AUTH = $null + [Runtime.InteropServices.Marshal]::ZeroFreeBSTR($tokenPointer) + $mutex.ReleaseMutex() + $mutex.Dispose() +} diff --git a/tools/train_torch_recurrent_forecaster.py b/tools/train_torch_recurrent_forecaster.py index b6805d9..08f14d1 100644 --- a/tools/train_torch_recurrent_forecaster.py +++ b/tools/train_torch_recurrent_forecaster.py @@ -125,6 +125,9 @@ def main() -> None: decision_horizon = args.horizon if args.horizon > 0 else max(1, settings.time_series_forecast_horizon) target_horizons = _horizons(args.horizons, decision_horizon) feature_names = _feature_names_arg(args.features) + if args.pooled: + feature_names.extend(f"symbol_is_{symbol}" for symbol in symbols) + ensemble_seeds = _ints(args.ensemble_seeds) or [args.seed] round_trip_cost = max(0.0, 2.0 * (float(settings.taker_fee_rate) + float(settings.slippage_rate))) _progress( f"training started: symbols={len(symbols)} interval={interval} " @@ -151,9 +154,82 @@ def main() -> None: "feature_names": feature_names, "feature_count": len(feature_names), "device": str(device), + "ensemble_seeds": ensemble_seeds, + "selection_folds": args.selection_folds, "symbols": {}, } + if args.pooled: + artifact["version"] = 5 + artifact["pooled_multi_asset"] = True + artifact["symbol_embedding"] = "learned_one_hot_projection" + artifact["symbols"] = _train_pooled_symbols( + client=client, + symbols=symbols, + interval=interval, + limit=args.limit, + validation_window=args.validation_window, + holdout_window=args.holdout_window, + target_horizons=target_horizons, + decision_horizon=decision_horizon, + feature_names=feature_names, + round_trip_cost=round_trip_cost, + context_symbols=_strings(args.context_symbols), + architectures=_strings(args.architectures), + lookbacks=_ints(args.lookbacks), + hidden_sizes=_ints(args.hidden_sizes), + layers_values=_ints(args.layers), + dropouts=_floats(args.dropouts), + epochs=args.epochs, + patience=args.patience, + batch_size=args.batch_size, + learning_rate=args.learning_rate, + weight_decay=args.weight_decay, + clip=args.clip, + attention_pooling=args.attention_pooling, + context_norm=args.context_norm, + device=device, + seeds=ensemble_seeds, + selection_folds=args.selection_folds, + ) + else: + artifact["symbols"] = _train_independent_symbols( + client=client, + symbols=symbols, + interval=interval, + args=args, + target_horizons=target_horizons, + decision_horizon=decision_horizon, + feature_names=feature_names, + round_trip_cost=round_trip_cost, + device=device, + ensemble_seeds=ensemble_seeds, + ) + + for symbol, result in artifact["symbols"].items(): + _progress( + f"{symbol}: model={result['model']} lookback={result['lookback']} " + f"features={result['input_size']} hidden={result['hidden_size']} " + f"layers={result['num_layers']} horizons={','.join(map(str, result['target_horizons']))} " + f"mae={result['validation_mae_percent']:.5f}% " + f"baseline={result['baseline_mae_percent']:.5f}% " + f"skill={result['skill']:.4f} dir={result['directional_accuracy']:.3f} " + f"p_brier={result['probability_brier']:.4f}" + ) + + output.parent.mkdir(parents=True, exist_ok=True) + tmp_output = output.with_name(f"{output.name}.tmp") + tmp_output.write_text(json.dumps(artifact, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + tmp_output.replace(output) + _progress(f"saved {output}") + + +def _train_independent_symbols( + *, client: BybitClient, symbols: list[str], interval: str, args: argparse.Namespace, + target_horizons: list[int], decision_horizon: int, feature_names: list[str], + round_trip_cost: float, device: torch.device, ensemble_seeds: list[int], +) -> dict[str, Any]: + results: dict[str, Any] = {} total_symbols = len(symbols) for index, symbol in enumerate(symbols, start=1): _progress(f"{symbol}: training started ({index}/{total_symbols})") @@ -183,27 +259,221 @@ def main() -> None: attention_pooling=args.attention_pooling, context_norm=args.context_norm, device=device, - seed=args.seed, + seeds=ensemble_seeds, + selection_folds=args.selection_folds, ) if result is None: _progress(f"{symbol}: skipped, not enough candles or train/validation samples") continue - artifact["symbols"][symbol] = result - _progress( - f"{symbol}: model={result['model']} lookback={result['lookback']} " - f"features={result['input_size']} hidden={result['hidden_size']} " - f"layers={result['num_layers']} horizons={','.join(map(str, result['target_horizons']))} " - f"mae={result['validation_mae_percent']:.5f}% " - f"baseline={result['baseline_mae_percent']:.5f}% " - f"skill={result['skill']:.4f} dir={result['directional_accuracy']:.3f} " - f"p_brier={result['probability_brier']:.4f}" - ) + results[symbol] = result + return results - output.parent.mkdir(parents=True, exist_ok=True) - tmp_output = output.with_name(f"{output.name}.tmp") - tmp_output.write_text(json.dumps(artifact, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") - tmp_output.replace(output) - _progress(f"saved {output}") + +def _train_pooled_symbols( + *, client: BybitClient, symbols: list[str], interval: str, limit: int, + validation_window: int, holdout_window: int, target_horizons: list[int], + decision_horizon: int, feature_names: list[str], round_trip_cost: float, + context_symbols: list[str], architectures: list[str], lookbacks: list[int], + hidden_sizes: list[int], layers_values: list[int], dropouts: list[float], + epochs: int, patience: int, batch_size: int, learning_rate: float, + weight_decay: float, clip: float, attention_pooling: bool, context_norm: bool, + device: torch.device, seeds: list[int], selection_folds: int, +) -> dict[str, Any]: + market_candles: dict[str, list[Candle]] = {} + for symbol in sorted({item.upper() for item in symbols + context_symbols}): + rows = _historical_klines(client, symbol, interval, limit) + add_indicators(rows) + market_candles[symbol] = rows + _progress(f"{symbol}: pooled data loaded ({len(rows)} candles)") + trend_by_symbol: dict[str, list[Candle]] = {} + for symbol in symbols: + rows = _historical_klines(client, symbol, "D", min(max(260, limit // 24 + 260), 1000)) + add_indicators(rows) + trend_by_symbol[symbol] = rows + + best: dict[str, Any] | None = None + best_prepared: dict[str, PreparedData] = {} + for lookback in lookbacks: + prepared_by_symbol: dict[str, PreparedData] = {} + for symbol in symbols: + prepared = _prepare_data( + candles=market_candles[symbol], + feature_names=feature_names, + lookback=lookback, + target_horizons=target_horizons, + decision_horizon=decision_horizon, + round_trip_cost=round_trip_cost, + market_candles=market_candles, + trend_candles=trend_by_symbol[symbol], + validation_window=validation_window, + holdout_window=holdout_window, + clip=clip, + device=device, + ) + if prepared is not None: + prepared_by_symbol[symbol] = prepared + if len(prepared_by_symbol) < 2: + continue + for architecture in architectures: + if architecture not in {"lstm", "gru"}: + continue + for hidden_size in hidden_sizes: + for num_layers in layers_values: + for dropout in dropouts: + if num_layers <= 1 and dropout != 0.0: + continue + _progress( + f"pooled: fitting {architecture} lookback={lookback} hidden={hidden_size} " + f"layers={num_layers} dropout={dropout} symbols={len(prepared_by_symbol)}" + ) + members = [ + _fit_pooled_candidate( + prepared_by_symbol=prepared_by_symbol, + architecture=architecture, + input_size=len(feature_names), + output_size=len(target_horizons) * len(OUTPUT_LAYOUT), + hidden_size=hidden_size, + num_layers=num_layers, + dropout=dropout, + epochs=epochs, + patience=patience, + batch_size=batch_size, + learning_rate=learning_rate, + weight_decay=weight_decay, + clip=clip, + attention_pooling=attention_pooling, + context_norm=context_norm, + device=device, + seed=member_seed, + selection_folds=selection_folds, + ) + for member_seed in seeds + ] + candidate = _ensemble_candidate(members, seeds) + candidate.update( + model=f"torch_{architecture}", architecture=architecture, + lookback=lookback, hidden_size=hidden_size, num_layers=num_layers, + dropout=dropout if num_layers > 1 else 0.0, + attention_pooling=attention_pooling, context_norm=context_norm, + input_size=len(feature_names), output_size=len(target_horizons) * len(OUTPUT_LAYOUT), + ) + if best is None or _candidate_score(candidate) < _candidate_score(best): + best = candidate + best_prepared = prepared_by_symbol + if best is None: + return {} + + results: dict[str, Any] = {} + symbol_metrics = best.get("symbol_metrics", {}) + common = {key: value for key, value in best.items() if key != "symbol_metrics"} + for symbol, prepared in best_prepared.items(): + metrics = symbol_metrics.get(symbol, {}) if isinstance(symbol_metrics, dict) else {} + baseline = sum(abs(row[prepared.decision_horizon_index]) for row in prepared.validation_targets) / len(prepared.validation_targets) + validation_mae = float(metrics.get("validation_mae", baseline)) + results[symbol] = { + **common, **metrics, + "pooled_multi_asset": True, + "target_horizon": prepared.decision_horizon, + "target_horizons": prepared.target_horizons, + "direct_horizon": True, + "target_transform": "net_return_over_volatility", + "round_trip_cost": round(round_trip_cost, 10), + "output_layout": list(OUTPUT_LAYOUT), + "feature_names": feature_names, + "feature_means": prepared.feature_means, + "feature_scales": prepared.feature_scales, + "target_means": prepared.target_means, + "target_scales": prepared.target_scales, + "target_mean": prepared.target_means[prepared.decision_horizon_index], + "target_scale": prepared.target_scales[prepared.decision_horizon_index], + "clip": clip, + "validation_mae_percent": validation_mae * 100, + "baseline_mae_percent": baseline * 100, + "validation_skill": (baseline - validation_mae) / baseline if baseline > 0 else 0.0, + # Runtime and calibration may use validation skill. Untouched + # holdout skill is report-only and must never gate individual entries. + "skill": (baseline - validation_mae) / baseline if baseline > 0 else 0.0, + "train_samples": prepared.train_samples, + "validation_samples": prepared.validation_samples, + "holdout_samples": prepared.holdout_samples, + "holdout_start_timestamp": prepared.holdout_start_timestamp, + } + return results + + +def _fit_pooled_candidate( + *, prepared_by_symbol: dict[str, PreparedData], architecture: str, input_size: int, + output_size: int, hidden_size: int, num_layers: int, dropout: float, epochs: int, + patience: int, batch_size: int, learning_rate: float, weight_decay: float, + clip: float, attention_pooling: bool, context_norm: bool, device: torch.device, + seed: int, selection_folds: int, +) -> dict[str, Any]: + _seed(seed) + model = RecurrentReturnModel( + architecture=architecture, input_size=input_size, hidden_size=hidden_size, + num_layers=num_layers, dropout=dropout, output_size=output_size, + attention_pooling=attention_pooling, context_norm=context_norm, + ).to(device) + optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate, weight_decay=weight_decay) + loader = DataLoader( + TensorDataset( + torch.cat([row.train_x for row in prepared_by_symbol.values()]), + torch.cat([row.train_y for row in prepared_by_symbol.values()]), + torch.cat([row.train_up for row in prepared_by_symbol.values()]), + ), + batch_size=max(1, batch_size), shuffle=True, + generator=torch.Generator(device="cpu").manual_seed(seed), + ) + best_state: dict[str, torch.Tensor] | None = None + best_score = math.inf + stale = 0 + best_epoch = 0 + for epoch in range(1, max(1, epochs) + 1): + model.train() + for batch_x, batch_y, batch_up in loader: + optimizer.zero_grad(set_to_none=True) + loss = _forecast_loss(model(batch_x), batch_y, batch_up, len(next(iter(prepared_by_symbol.values())).target_horizons)) + loss.backward() + nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) + optimizer.step() + symbol_rows = { + symbol: _validation_metrics(model, prepared, clip) + for symbol, prepared in prepared_by_symbol.items() + } + score = sum(float(row["validation_mae"]) for row in symbol_rows.values()) / len(symbol_rows) + if score + 1e-12 < best_score: + best_score = score + best_epoch = epoch + best_state = {key: value.detach().cpu().clone() for key, value in model.state_dict().items()} + stale = 0 + else: + stale += 1 + if stale >= max(1, patience): + break + if best_state: + model.load_state_dict(best_state) + per_symbol: dict[str, dict[str, Any]] = {} + for symbol, prepared in prepared_by_symbol.items(): + metrics = _validation_metrics(model, prepared, clip) + metrics.update(_validation_stability_metrics(model, prepared, clip, selection_folds)) + metrics.update(_holdout_metrics(model, prepared, clip)) + per_symbol[symbol] = metrics + aggregate: dict[str, Any] = {"symbol_metrics": per_symbol} + for name in ( + "validation_mae", "directional_accuracy", "buy_precision", "probability_brier", + "holdout_skill", "validation_fold_mae_std", "validation_trade_mean", + "validation_trade_win_rate", + ): + values = [float(row[name]) for row in per_symbol.values() if isinstance(row.get(name), (int, float))] + aggregate[name] = sum(values) / len(values) if values else 0.0 + aggregate.update( + best_epoch=best_epoch, epochs_trained=best_epoch + stale, + state_dict=_export_recurrent_state(model), + head_weight=_round_nested(model.head.weight.detach().cpu().tolist()), + head_bias=_round_list(model.head.bias.detach().cpu().tolist()), + **_export_context_state(model), + ) + return aggregate def _progress(message: str) -> None: @@ -236,6 +506,9 @@ def _parse_args() -> argparse.Namespace: parser.add_argument("--attention-pooling", action=argparse.BooleanOptionalAction, default=True, help="Use exportable attention pooling over recurrent states.") parser.add_argument("--context-norm", action=argparse.BooleanOptionalAction, default=True, help="Use exportable LayerNorm before the forecast head.") parser.add_argument("--seed", type=int, default=7, help="Random seed.") + parser.add_argument("--ensemble-seeds", default="7,19,43", help="Comma-separated seeds averaged at inference time.") + parser.add_argument("--selection-folds", type=int, default=3, help="Validation slices used to penalize unstable candidates.") + parser.add_argument("--pooled", action=argparse.BooleanOptionalAction, default=True, help="Train shared multi-asset recurrent weights with learned symbol one-hot projection.") parser.add_argument("--threads", type=int, default=0, help="Torch CPU threads; 0 keeps torch default.") parser.add_argument("--device", default="auto", help="auto, cpu, cuda, or mps.") parser.add_argument("--output", default="", help="Output JSON path. Defaults to TIME_SERIES_LSTM_MODEL_PATH.") @@ -277,7 +550,8 @@ def _train_symbol( attention_pooling: bool, context_norm: bool, device: torch.device, - seed: int, + seeds: list[int], + selection_folds: int, ) -> dict[str, Any] | None: candles = _historical_klines(client, symbol, interval, limit) add_indicators(candles) @@ -339,25 +613,30 @@ def _train_symbol( f"lookback={lookback} hidden={hidden_size} " f"layers={num_layers} dropout={dropout}" ) - candidate = _fit_candidate( - prepared=prepared, - architecture=architecture, - input_size=len(feature_names), - output_size=len(target_horizons) * len(OUTPUT_LAYOUT), - hidden_size=hidden_size, - num_layers=num_layers, - dropout=dropout, - epochs=epochs, - patience=patience, - batch_size=batch_size, - learning_rate=learning_rate, - weight_decay=weight_decay, - clip=clip, - attention_pooling=attention_pooling, - context_norm=context_norm, - device=device, - seed=seed, - ) + members = [ + _fit_candidate( + prepared=prepared, + architecture=architecture, + input_size=len(feature_names), + output_size=len(target_horizons) * len(OUTPUT_LAYOUT), + hidden_size=hidden_size, + num_layers=num_layers, + dropout=dropout, + epochs=epochs, + patience=patience, + batch_size=batch_size, + learning_rate=learning_rate, + weight_decay=weight_decay, + clip=clip, + attention_pooling=attention_pooling, + context_norm=context_norm, + device=device, + seed=member_seed, + selection_folds=selection_folds, + ) + for member_seed in seeds + ] + candidate = _ensemble_candidate(members, seeds) validation_mae = float(candidate["validation_mae"]) skill = (baseline_mae - validation_mae) / baseline_mae if baseline_mae > 0 else 0.0 row = { @@ -408,7 +687,7 @@ def _train_symbol( if best is None: return None best["validation_skill"] = best.get("skill", 0.0) - best["skill"] = best.get("holdout_skill", 0.0) + best["skill"] = best["validation_skill"] best.pop("validation_mae", None) return best @@ -484,7 +763,7 @@ def _prepare_data( if len(train_samples) < 24 or len(validation_samples) < 8 or len(holdout_samples) < 16: return None - feature_means, feature_scales = _feature_stats(train_samples, len(feature_names)) + feature_means, feature_scales = _feature_stats(train_samples, feature_names) target_means, target_scales = _target_stats(train_samples, len(target_horizons)) decision_horizon = decision_horizon if decision_horizon in target_horizons else min( target_horizons, @@ -545,7 +824,8 @@ def _prepare_data( ) -def _feature_stats(samples: list[TrainingSample], input_size: int) -> tuple[list[float], list[float]]: +def _feature_stats(samples: list[TrainingSample], feature_names: list[str]) -> tuple[list[float], list[float]]: + input_size = len(feature_names) columns = [[] for _ in range(input_size)] for sample in samples: window = sample.window @@ -554,7 +834,11 @@ def _feature_stats(samples: list[TrainingSample], input_size: int) -> tuple[list columns[index].append(float(row[index] if index < len(row) else 0.0)) means: list[float] = [] scales: list[float] = [] - for values in columns: + for index, values in enumerate(columns): + if feature_names[index].startswith("symbol_is_"): + means.append(0.0) + scales.append(1.0) + continue if not values: means.append(0.0) scales.append(1.0) @@ -641,6 +925,7 @@ def _fit_candidate( context_norm: bool, device: torch.device, seed: int, + selection_folds: int, ) -> dict[str, Any]: _seed(seed) model = RecurrentReturnModel( @@ -676,6 +961,7 @@ def _fit_candidate( optimizer.step() metrics = _validation_metrics(model, prepared, clip) + metrics.update(_validation_stability_metrics(model, prepared, clip, selection_folds)) if metrics["validation_mae"] + 1e-12 < best_metrics["validation_mae"]: best_metrics = metrics best_epoch = epoch @@ -701,6 +987,66 @@ def _fit_candidate( } +def _ensemble_candidate(members: list[dict[str, Any]], seeds: list[int]) -> dict[str, Any]: + if not members: + raise ValueError("ensemble requires at least one member") + result = dict(members[0]) + metric_names = ( + "validation_mae", + "directional_accuracy", + "buy_precision", + "probability_brier", + "holdout_mae", + "holdout_baseline_mae", + "holdout_skill", + "holdout_directional_accuracy", + "holdout_buy_precision", + "holdout_probability_brier", + "validation_fold_mae_std", + "validation_fold_mae_worst", + "validation_trade_mean", + "validation_trade_win_rate", + ) + for name in metric_names: + values = [float(member[name]) for member in members if isinstance(member.get(name), (int, float))] + if values: + result[name] = sum(values) / len(values) + export_names = ( + "state_dict", + "head_weight", + "head_bias", + "attention_weight", + "attention_bias", + "context_norm_weight", + "context_norm_bias", + ) + result["ensemble_members"] = [ + {name: member[name] for name in export_names if name in member} + | {"seed": seeds[index] if index < len(seeds) else index} + for index, member in enumerate(members) + ] + result["ensemble_size"] = len(members) + symbol_names = sorted( + { + symbol + for member in members + for symbol in (member.get("symbol_metrics") or {}) + } + ) + if symbol_names: + result["symbol_metrics"] = {} + for symbol in symbol_names: + rows = [member.get("symbol_metrics", {}).get(symbol, {}) for member in members] + keys = {key for row in rows if isinstance(row, dict) for key in row} + averaged: dict[str, float] = {} + for key in keys: + values = [float(row[key]) for row in rows if isinstance(row.get(key), (int, float))] + if values: + averaged[key] = sum(values) / len(values) + result["symbol_metrics"][symbol] = averaged + return result + + def _validation_metrics(model: nn.Module, prepared: PreparedData, clip: float) -> dict[str, float]: return _evaluation_metrics( model, @@ -712,6 +1058,39 @@ def _validation_metrics(model: nn.Module, prepared: PreparedData, clip: float) - ) +def _validation_stability_metrics( + model: nn.Module, + prepared: PreparedData, + clip: float, + folds: int, +) -> dict[str, float]: + fold_count = max(1, min(int(folds), len(prepared.validation_targets))) + fold_size = max(1, len(prepared.validation_targets) // fold_count) + maes: list[float] = [] + for fold in range(fold_count): + start = fold * fold_size + end = len(prepared.validation_targets) if fold == fold_count - 1 else min(len(prepared.validation_targets), start + fold_size) + if end <= start: + continue + metrics = _evaluation_metrics( + model, + values=prepared.validation_x[start:end], + targets=prepared.validation_targets[start:end], + volatility_scales=prepared.validation_volatility_scales[start:end], + prepared=prepared, + clip=clip, + ) + maes.append(float(metrics["validation_mae"])) + if not maes: + return {"validation_fold_mae_std": 0.0, "validation_fold_mae_worst": math.inf} + mean = sum(maes) / len(maes) + variance = sum((value - mean) ** 2 for value in maes) / len(maes) + return { + "validation_fold_mae_std": math.sqrt(variance), + "validation_fold_mae_worst": max(maes), + } + + def _holdout_metrics(model: nn.Module, prepared: PreparedData, clip: float) -> dict[str, Any]: metrics = _evaluation_metrics( model, @@ -790,6 +1169,13 @@ def _evaluation_metrics( if prediction > 0 ] buy_wins = [actual for actual in buy_predictions if actual > 0] + ranked = sorted( + zip(decision_predictions, [row[decision] for row in probabilities], decision_targets), + key=lambda item: item[0] * max(0.0, item[1] - 0.5), + reverse=True, + ) + selected = ranked[: max(8, len(ranked) // 5)] + selected_targets = [row[2] for row in selected] by_horizon = {} baseline_by_horizon = {} for horizon_index, horizon in enumerate(prepared.target_horizons): @@ -815,6 +1201,12 @@ def _evaluation_metrics( "directional_accuracy": len(correct) / len(non_zero) if non_zero else 0.0, "buy_precision": len(buy_wins) / len(buy_predictions) if buy_predictions else 0.0, "probability_brier": sum(probability_errors) / len(probability_errors) if probability_errors else 1.0, + "validation_trade_mean": sum(selected_targets) / len(selected_targets) if selected_targets else 0.0, + "validation_trade_win_rate": ( + sum(1 for value in selected_targets if value > 0) / len(selected_targets) + if selected_targets + else 0.0 + ), } @@ -824,9 +1216,13 @@ def _candidate_score(row: dict[str, Any]) -> float: directional = float(row.get("directional_accuracy", 0.0)) buy_precision = float(row.get("buy_precision", 0.0)) probability_brier = float(row.get("probability_brier", 1.0)) - return mae * (1.0 - max(0.0, skill) * 0.05) * (1.0 - max(0.0, directional - 0.5) * 0.03) * ( + fold_std = max(0.0, float(row.get("validation_fold_mae_std", 0.0))) + stability_penalty = 1.0 + min(1.0, fold_std / max(mae, 1e-9)) * 0.25 + trade_mean = float(row.get("validation_trade_mean", 0.0)) + trade_penalty = max(0.0, -trade_mean) * 2.0 - max(0.0, trade_mean) * 0.5 + return mae * stability_penalty * (1.0 - max(0.0, skill) * 0.05) * (1.0 - max(0.0, directional - 0.5) * 0.03) * ( 1.0 - max(0.0, buy_precision - 0.5) * 0.02 - ) * (1.0 + max(0.0, probability_brier - 0.25) * 0.02) + ) * (1.0 + max(0.0, probability_brier - 0.25) * 0.02) + trade_penalty def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: torch.Tensor, horizon_count: int) -> torch.Tensor: @@ -842,7 +1238,19 @@ def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: tor probabilities = torch.sigmoid(logits) pt = probabilities * up_targets + (1.0 - probabilities) * (1.0 - up_targets) focal = ((1.0 - pt) ** 2.0 * bce).mean() - return mean_loss + 0.35 * sum(quantile_losses) / len(quantile_losses) + 0.15 * focal + soft_long = torch.sigmoid(values[:, :, 0] * 2.0) * probabilities + after_cost_utility = -(soft_long * targets).mean() + prediction_centered = values[:, :, 0] - values[:, :, 0].mean(dim=0, keepdim=True) + target_centered = targets - targets.mean(dim=0, keepdim=True) + cosine = nn.functional.cosine_similarity(prediction_centered, target_centered, dim=0).mean() + ranking_loss = 1.0 - cosine + return ( + mean_loss + + 0.35 * sum(quantile_losses) / len(quantile_losses) + + 0.15 * focal + + 0.10 * after_cost_utility + + 0.05 * ranking_loss + ) def _export_recurrent_state(model: RecurrentReturnModel) -> dict[str, Any]: diff --git a/tools/windows_training_agent.py b/tools/windows_training_agent.py index f949d7a..a9f00b8 100644 --- a/tools/windows_training_agent.py +++ b/tools/windows_training_agent.py @@ -64,19 +64,20 @@ def poll_once(args: argparse.Namespace, repo_root: Path, runtime_dir: Path, log_ try: run_retrain(args, job_id, job, repo_root, log_path) summary = read_json(runtime_dir / "torch_retrain_guard.json") - if summary.get("accepted") is not True: - raise RuntimeError( - "candidate rejected by untouched-holdout guard: " - + str(summary.get("reason") or "validation failed") - ) - report_progress(args, job_id, "running", "uploading", 72, "Обучение завершено, загружаю артефакты") - for name in ARTIFACT_NAMES: - path = runtime_dir / name - if path.is_file(): - upload_artifact(args, job_id, path, log_path) + accepted = summary.get("accepted") is True + if accepted: + report_progress(args, job_id, "running", "uploading", 72, "Обучение завершено, загружаю артефакты") + for name in ARTIFACT_NAMES: + path = runtime_dir / name + if path.is_file(): + upload_artifact(args, job_id, path, log_path) + message = "training completed; candidate accepted" + log(log_path, f"Completed retrain job {job_id}; candidate accepted") + else: + reason = str(summary.get("reason") or "validation failed") + message = f"training completed; candidate rejected by quality gate: {reason}" + log(log_path, f"Completed retrain job {job_id}; candidate rejected: {reason}") 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}") @@ -108,6 +109,10 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo "dropouts": "-Dropouts", "epochs": "-Epochs", "holdout_window": "-HoldoutWindow", + "ensemble_seeds": "-EnsembleSeeds", + "selection_folds": "-SelectionFolds", + "learning_rate": "-LearningRate", + "weight_decay": "-WeightDecay", } for key, ps_arg in arg_map.items(): value = parameters.get(key)