From d58e20aa4dfdacc3ef6ce3c29f440fb4ab6cdea1 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 27 Jun 2026 17:52:49 +0300 Subject: [PATCH] Improve training agent progress reporting --- tools/train_torch_recurrent_forecaster.py | 26 ++++++++++++++++---- tools/windows_training_agent.py | 30 ++++++++++++++++++++++- 2 files changed, 50 insertions(+), 6 deletions(-) diff --git a/tools/train_torch_recurrent_forecaster.py b/tools/train_torch_recurrent_forecaster.py index 346e0fa..7876388 100644 --- a/tools/train_torch_recurrent_forecaster.py +++ b/tools/train_torch_recurrent_forecaster.py @@ -118,6 +118,10 @@ def main() -> None: target_horizons = _horizons(args.horizons, decision_horizon) feature_names = _feature_names_arg(args.features) 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} " + f"limit={args.limit} epochs={args.epochs}" + ) artifact: dict[str, Any] = { "version": 4, @@ -141,7 +145,9 @@ def main() -> None: "symbols": {}, } - for symbol in symbols: + total_symbols = len(symbols) + for index, symbol in enumerate(symbols, start=1): + _progress(f"{symbol}: training started ({index}/{total_symbols})") result = _train_symbol( client=client, symbol=symbol, @@ -170,10 +176,10 @@ def main() -> None: seed=args.seed, ) if result is None: - print(f"{symbol}: skipped, not enough candles or train/validation samples") + _progress(f"{symbol}: skipped, not enough candles or train/validation samples") continue artifact["symbols"][symbol] = result - print( + _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']))} " @@ -187,7 +193,11 @@ def main() -> None: 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) - print(f"saved {output}") + _progress(f"saved {output}") + + +def _progress(message: str) -> None: + print(message, flush=True) def _parse_args() -> argparse.Namespace: @@ -274,12 +284,13 @@ def _train_symbol( add_indicators(rows) market_candles[context_symbol] = rows except Exception as exc: - print(f"{symbol}: context {context_symbol} skipped: {exc}") + _progress(f"{symbol}: context {context_symbol} skipped: {exc}") trend_candles = _historical_klines(client, symbol, "D", min(max(260, limit // 24 + 260), 1000)) add_indicators(trend_candles) best: dict[str, Any] | None = None for lookback in lookbacks: + _progress(f"{symbol}: preparing lookback={lookback}") prepared = _prepare_data( candles=candles, feature_names=feature_names, @@ -307,6 +318,11 @@ def _train_symbol( for dropout in dropouts: if num_layers <= 1 and dropout != 0.0: continue + _progress( + f"{symbol}: fitting {architecture} " + f"lookback={lookback} hidden={hidden_size} " + f"layers={num_layers} dropout={dropout}" + ) candidate = _fit_candidate( prepared=prepared, architecture=architecture, diff --git a/tools/windows_training_agent.py b/tools/windows_training_agent.py index 7ae300c..16147be 100644 --- a/tools/windows_training_agent.py +++ b/tools/windows_training_agent.py @@ -128,12 +128,15 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo reader = threading.Thread(target=read_output, name="training-output-reader", daemon=True) reader.start() last_report_at = 0.0 + started_at = time.monotonic() + last_output_at = started_at last_message = "PyTorch retrain выполняется" while True: got_line = False try: message = output_queue.get(timeout=5) got_line = True + last_output_at = time.monotonic() log(log_path, message) line_count += 1 if message: @@ -144,7 +147,10 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo progress = min(70, 8 + line_count // 3) now = time.monotonic() if got_line or now - last_report_at >= 30: - safe_report_progress(args, job_id, "running", "training", progress, last_message, log_path) + report_message = last_message + if not got_line: + report_message = training_heartbeat_message(now, started_at, last_output_at, last_message) + safe_report_progress(args, job_id, "running", "training", progress, report_message, log_path) last_report_at = now if process.poll() is not None and output_queue.empty(): @@ -157,6 +163,28 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo report_progress(args, job_id, "running", "guard", 70, "Guard завершён, подготавливаю артефакты") +def training_heartbeat_message(now: float, started_at: float, last_output_at: float, last_message: str) -> str: + elapsed = format_duration(now - started_at) + idle_seconds = max(0.0, now - last_output_at) + if idle_seconds >= 45: + return ( + f"PyTorch обучает модель: процесс активен {elapsed}; " + f"последний лог {format_duration(idle_seconds)} назад: {last_message[:140]}" + ) + return last_message or f"PyTorch обучает модель: процесс активен {elapsed}" + + +def format_duration(seconds: float) -> str: + total_seconds = max(0, int(seconds)) + minutes, seconds_part = divmod(total_seconds, 60) + hours, minutes_part = divmod(minutes, 60) + if hours: + return f"{hours}ч {minutes_part}м" + if minutes: + return f"{minutes}м {seconds_part}с" + return f"{seconds_part}с" + + def upload_artifact(args: argparse.Namespace, job_id: str, path: Path, log_path: Path) -> None: digest = hashlib.sha256(path.read_bytes()).hexdigest() size = path.stat().st_size