refactor(quickadapter): share Optuna-artifact quarantine path helper

The best-params quarantine in Utils re-inlined the timestamp + tie-break
path computation that the journal quarantine already implemented in
QuickAdapterRegressorV3, duplicating the algorithm across the two
modules.

- Add a shared, parameterized _optuna_quarantine_path(path, now, *, tag,
  limit) in Utils (the module both call sites can import).
- Delegate from _quarantine_corrupt_optuna_best_params (Utils) and from
  _optuna_quarantine_journal (regressor); each keeps its own domain
  constants (journal vs best-params), logging and rename handling, so
  quarantine names, log output and error semantics are unchanged.
- Remove the regressor's private _optuna_quarantine_path copy.

diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
index 230a594a651c0053fa060e74922fea62b3f73166..6146913e6c90afab099e29a9b58669485749c6f8 100644 (file)
--- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
+++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
@@ -103,6 +103,7 @@ from Utils import (
     label_weight_column_name,
     label_weight_known_at_lookahead_column_name,
     migrate_config,
+    _optuna_quarantine_path,
     optuna_load_best_params,
     optuna_save_best_params,
     require_bool,
@@ -4783,26 +4784,6 @@ class QuickAdapterRegressorV3(BaseRegressionModel):
                 self.optuna_save_best_params(pair, namespace)
         return study
 
-    @staticmethod
-    def _optuna_quarantine_path(journal_path: Path, now: datetime) -> Path:
-        """Quarantine target path for a corrupt Optuna journal.
-
-        The tag is appended *after* ``.log`` so the live-journal glob
-        ``optuna-*.log`` never matches quarantined artefacts. Collisions
-        are bounded by ``_OPTUNA_JOURNAL_QUARANTINE_TIE_BREAK_LIMIT``;
-        exhausted candidates raise instead of reusing a quarantine file.
-        """
-        stamp = now.strftime("%Y%m%dT%H%M%S%fZ")
-        tag = QuickAdapterRegressorV3._OPTUNA_JOURNAL_QUARANTINE_TAG
-        base_name = f"{journal_path.name}.{tag}-{stamp}"
-        limit = QuickAdapterRegressorV3._OPTUNA_JOURNAL_QUARANTINE_TIE_BREAK_LIMIT
-        for index in range(limit + 1):
-            suffix = "" if index == 0 else f"-{index}"
-            candidate = journal_path.with_name(f"{base_name}{suffix}")
-            if not candidate.exists():
-                return candidate
-        raise FileExistsError(journal_path)
-
     @staticmethod
     def _optuna_quarantine_journal(
         journal_path: Path, pair: str, cause: Exception
@@ -4817,8 +4798,11 @@ class QuickAdapterRegressorV3(BaseRegressionModel):
         """
         if not journal_path.exists():
             return None
-        quarantine_path = QuickAdapterRegressorV3._optuna_quarantine_path(
-            journal_path, datetime.now(timezone.utc)
+        quarantine_path = _optuna_quarantine_path(
+            journal_path,
+            datetime.now(timezone.utc),
+            tag=QuickAdapterRegressorV3._OPTUNA_JOURNAL_QUARANTINE_TAG,
+            limit=QuickAdapterRegressorV3._OPTUNA_JOURNAL_QUARANTINE_TIE_BREAK_LIMIT,
         )
         try:
             journal_path.rename(quarantine_path)

diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py
index e2d8c5ec630d18a0d3c1d820d2daa29be0c8234c..7ab52f5964f24f2c0cbde3884bace78a1a50ba62 100644 (file)
--- a/quickadapter/user_data/strategies/Utils.py
+++ b/quickadapter/user_data/strategies/Utils.py
@@ -5482,6 +5482,26 @@ def _validate_optuna_label_best_params(
     return params
 
 
+def _optuna_quarantine_path(
+    path: Path, now: datetime, *, tag: str, limit: int
+) -> Path:
+    """Quarantine target path for a corrupt Optuna artefact.
+
+    The tag and timestamp are appended after the complete filename
+    (extension included) so live-artefact globs never match quarantined
+    files. Collisions are bounded by ``limit``; exhausted candidates raise
+    instead of reusing a quarantine file.
+    """
+    stamp = now.strftime("%Y%m%dT%H%M%S%fZ")
+    base_name = f"{path.name}.{tag}-{stamp}"
+    for index in range(limit + 1):
+        suffix = "" if index == 0 else f"-{index}"
+        candidate = path.with_name(f"{base_name}{suffix}")
+        if not candidate.exists():
+            return candidate
+    raise FileExistsError(path)
+
+
 def _quarantine_corrupt_optuna_best_params(
     best_params_path: Path,
     pair: str,
@@ -5492,17 +5512,12 @@ def _quarantine_corrupt_optuna_best_params(
     """Atomically move corrupt persisted best params out of the live path."""
     if not best_params_path.exists():
         return None
-    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
-    quarantine_base = (
-        f"{best_params_path.name}.{_OPTUNA_BEST_PARAMS_QUARANTINE_TAG}-{stamp}"
+    quarantine_path = _optuna_quarantine_path(
+        best_params_path,
+        datetime.now(timezone.utc),
+        tag=_OPTUNA_BEST_PARAMS_QUARANTINE_TAG,
+        limit=_OPTUNA_BEST_PARAMS_QUARANTINE_TIE_BREAK_LIMIT,
     )
-    for index in range(_OPTUNA_BEST_PARAMS_QUARANTINE_TIE_BREAK_LIMIT + 1):
-        suffix = "" if index == 0 else f"-{index}"
-        quarantine_path = best_params_path.with_name(f"{quarantine_base}{suffix}")
-        if not quarantine_path.exists():
-            break
-    else:
-        raise FileExistsError(best_params_path)
     try:
         best_params_path.rename(quarantine_path)
     except OSError as quarantine_error: