101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""Diverse top-N finalist selection (doc 06 §3).
|
|
|
|
Optuna converges: the top 10 trials by score are usually near-clones of one
|
|
peak. Verifying 3 clones in MT5 tells you nothing about robustness. Instead,
|
|
pick **meaningfully different** finalists via greedy max-distance selection,
|
|
lightly weighted by rank so strong scores are still preferred.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
|
|
|
|
def param_distance(
|
|
a: dict[str, float],
|
|
b: dict[str, float],
|
|
ranges: dict[str, tuple[float, float, float]],
|
|
) -> float:
|
|
"""Average normalized parameter distance between two trials.
|
|
|
|
Each numeric parameter is min-max normalized to ``[0, 1]`` using its
|
|
declared range; booleans map to 0/1; categoricals (here as ints) use
|
|
index distance. Returns the mean across all shared parameters.
|
|
"""
|
|
keys = [k for k in ranges if k in a and k in b]
|
|
if not keys:
|
|
return 0.0
|
|
dists = []
|
|
for k in keys:
|
|
lo, hi, _ = ranges[k]
|
|
span = hi - lo
|
|
if span <= 0:
|
|
dists.append(0.0)
|
|
continue
|
|
dists.append(abs(float(a[k]) - float(b[k])) / span)
|
|
return float(np.mean(dists))
|
|
|
|
|
|
def select_diverse_topn(
|
|
study,
|
|
n: int,
|
|
ranges: dict[str, tuple[float, float, float]],
|
|
*,
|
|
constraints_key: str = "violations",
|
|
max_constraint_violations: int = 0,
|
|
rank_weight: float = 0.3,
|
|
) -> list:
|
|
"""Select ``n`` diverse, high-scoring, constraint-passing trials.
|
|
|
|
Algorithm (doc 06 §3):
|
|
|
|
1. filter trials by the constraints (drop violators)
|
|
2. sort by value descending
|
|
3. start the selected set with the single best trial
|
|
4. repeatedly add the remaining trial with MAXIMUM parameter-distance to
|
|
the already-selected set, lightly weighted by its rank so strong
|
|
scores are still preferred
|
|
5. stop at ``n``
|
|
"""
|
|
completed = [t for t in study.trials if t.state == TrialState.COMPLETE]
|
|
# Filter by constraint violations.
|
|
passing = [
|
|
t for t in completed
|
|
if len(t.user_attrs.get(constraints_key, [])) <= max_constraint_violations
|
|
]
|
|
if not passing:
|
|
return []
|
|
# Sort by value (assume maximize; negate for minimize).
|
|
passing.sort(key=lambda t: (t.value if t.value is not None else float("-inf")), reverse=True)
|
|
|
|
selected = [passing[0]]
|
|
remaining = passing[1:]
|
|
|
|
while remaining and len(selected) < n:
|
|
# Score each remaining trial by min-distance to the selected set,
|
|
# blended with a rank bonus so we still prefer strong scores.
|
|
best_trial = None
|
|
best_score = float("-inf")
|
|
for rank, t in enumerate(remaining):
|
|
d = max(param_distance(t.params, s.params, ranges) for s in selected)
|
|
rank_bonus = (1.0 - rank / max(len(remaining), 1)) * rank_weight
|
|
score = d + rank_bonus
|
|
if score > best_score:
|
|
best_score = score
|
|
best_trial = t
|
|
if best_trial is None:
|
|
break
|
|
selected.append(best_trial)
|
|
remaining.remove(best_trial)
|
|
|
|
return selected
|
|
|
|
|
|
# Late import to avoid hard dependency at module load for type hints only.
|
|
try:
|
|
from optuna.trial import TrialState # type: ignore
|
|
except Exception: # pragma: no cover - optuna always installed in this stack
|
|
class TrialState: # type: ignore[no-redef]
|
|
COMPLETE = "COMPLETE"
|