0f9d756caa
Extend ml_training_service to dispatch GPU training jobs as K8s batch/v1 Jobs, collect results via a Rust sidecar uploader, and support model promotion with operator approval via fxt CLI. - K8s dispatcher creates Jobs on gpu-training pool with native sidecar - training_uploader crate: watches DONE/FAILED marker, uploads to S3, reports completion via ReportJobCompletion gRPC - PromotionManager compares metrics, queues better models for approval - 4 new proto RPCs: ReportJobCompletion, ListPendingPromotions, ApprovePromotion, RejectPromotion - fxt commands: train start, model list/approve/reject - Training binaries write DONE/FAILED markers + metrics.json - Dockerfile, K8s job template, and CI pipeline updated - StartTraining gracefully falls back to in-process when outside K8s - 27 new tests (16 service + 11 promotion), 141 total service tests pass Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
80 lines
2.8 KiB
Rust
80 lines
2.8 KiB
Rust
//! Training completion marker and metrics writer.
|
|
//!
|
|
//! After training finishes (successfully or not), the binary writes a marker
|
|
//! file (`DONE` or `FAILED`) plus an optional `metrics.json` into the output
|
|
//! directory. A K8s sidecar can watch for these files to detect completion
|
|
//! and trigger result uploads.
|
|
|
|
use std::path::Path;
|
|
|
|
use tracing::{error, info};
|
|
|
|
/// Summary metrics written alongside the `DONE` marker on success.
|
|
#[derive(serde::Serialize)]
|
|
pub struct CompletionMetrics {
|
|
/// Model name (e.g. "kan", "dqn", "ppo")
|
|
pub model: String,
|
|
/// Symbol trained on (e.g. "ES.FUT")
|
|
pub symbol: String,
|
|
/// Best validation loss across all folds (if available)
|
|
pub best_val_loss: Option<f64>,
|
|
/// Sharpe ratio from evaluation (if available)
|
|
pub sharpe_ratio: Option<f64>,
|
|
/// Total training epochs completed across all folds
|
|
pub epochs_completed: usize,
|
|
/// Number of walk-forward folds completed
|
|
pub folds_completed: usize,
|
|
}
|
|
|
|
/// Write a `DONE` marker and `metrics.json` into `output_dir`.
|
|
///
|
|
/// Errors are logged but never cause a panic -- the training result is
|
|
/// already persisted in checkpoints; the marker is best-effort.
|
|
pub fn write_success_marker(output_dir: &Path, metrics: &CompletionMetrics) {
|
|
// Write metrics.json
|
|
match serde_json::to_string_pretty(metrics) {
|
|
Ok(json) => {
|
|
let metrics_path = output_dir.join("metrics.json");
|
|
if let Err(e) = std::fs::write(&metrics_path, json) {
|
|
error!("Failed to write {}: {}", metrics_path.display(), e);
|
|
} else {
|
|
info!("Wrote {}", metrics_path.display());
|
|
}
|
|
}
|
|
Err(e) => {
|
|
error!("Failed to serialize CompletionMetrics: {}", e);
|
|
}
|
|
}
|
|
|
|
// Write empty DONE marker
|
|
let done_path = output_dir.join("DONE");
|
|
if let Err(e) = std::fs::write(&done_path, "") {
|
|
error!("Failed to write {}: {}", done_path.display(), e);
|
|
} else {
|
|
info!("Wrote {}", done_path.display());
|
|
}
|
|
}
|
|
|
|
/// Write a `FAILED` marker containing the error message.
|
|
///
|
|
/// Errors are logged but never cause a panic.
|
|
pub fn write_failure_marker(output_dir: &Path, error_msg: &str) {
|
|
// Ensure the output directory exists so the marker can be written even
|
|
// when training fails before creating any checkpoints.
|
|
if let Err(e) = std::fs::create_dir_all(output_dir) {
|
|
error!(
|
|
"Failed to create output dir {}: {} -- cannot write FAILED marker",
|
|
output_dir.display(),
|
|
e
|
|
);
|
|
return;
|
|
}
|
|
|
|
let failed_path = output_dir.join("FAILED");
|
|
if let Err(e) = std::fs::write(&failed_path, error_msg) {
|
|
error!("Failed to write {}: {}", failed_path.display(), e);
|
|
} else {
|
|
info!("Wrote {} ({})", failed_path.display(), error_msg);
|
|
}
|
|
}
|