# -*- coding: utf-8 -*-
"""
HPO Monitoring and Auditing SDK.
Provides high-level progress tracking and health checks for large-scale
Hyperparameter Optimization sweeps.
"""
#? Metadata
__author__ = "Gemini CLI"
__email__ = "adhisant@tnt.uni-hannover.de"
__license__ = "Clear BSD"
__version__ = "1.0.0"
# =============================================================================
# STANDARD LIBRARY IMPORTS
# =============================================================================
import typing as t
from datetime import datetime
# =============================================================================
# THIRD-PARTY IMPORTS
# =============================================================================
import optuna
from optuna.trial import TrialState
import pandas as pd
from loguru import logger
# =============================================================================
# CLASSES
# =============================================================================
[docs]
class AuditService:
"""Service for monitoring HPO progress and health."""
def __init__(self, optuna_db: str = "mysql+pymysql://optuna:optuna@juno:3311/optuna"):
self.optuna_db = optuna_db
[docs]
def get_study_progress(self, study_name: str) -> t.Dict[str, t.Any]:
"""Returns progress statistics for a single study."""
try:
study = optuna.load_study(study_name=study_name, storage=self.optuna_db)
trials = study.trials
states = [t.state for t in trials]
comp = states.count(TrialState.COMPLETE)
fail = states.count(TrialState.FAIL)
pruned = states.count(TrialState.PRUNED)
running = states.count(TrialState.RUNNING)
best_val = None
if comp > 0:
if len(study.best_trials[0].values) > 1:
# Multi-objective
best_val = study.best_trials[0].values
else:
best_val = study.best_trials[0].value
return {
"study_name": study_name,
"total": len(trials),
"complete": comp,
"fail": fail,
"pruned": pruned,
"running": running,
"best_value": best_val
}
except Exception as e:
logger.error(f"Failed to audit study {study_name}: {e}")
return {}
[docs]
def bulk_audit(self, pattern: str) -> pd.DataFrame:
"""Audits multiple studies matching a pattern."""
summaries = optuna.get_all_study_summaries(storage=self.optuna_db)
results = []
import re
p = re.compile(pattern, re.IGNORECASE)
for s in summaries:
if p.search(s.study_name):
results.append(self.get_study_progress(s.study_name))
return pd.DataFrame([r for r in results if r])