diff --git a/__init__.py b/__init__.py index 7170e90..23e836a 100644 --- a/__init__.py +++ b/__init__.py @@ -85,6 +85,7 @@ GridDCATrialOutput, JsonlOptimizationLogger, MissingOptimizationMetricError, + MultiSeedOptimization, ObjectiveResult, OptionPackageGenericEvaluator, OptionTrialOutput, @@ -96,6 +97,7 @@ PreparedPortfolioEvaluator, PreparedSignalEvaluator, ReportMetricObjective, + RobustSelectionConfig, SamplerConfig, SearchSpaceInfo, SelectedCandidate, @@ -584,11 +586,13 @@ "DuplicatePruner", "CONSTRAINTS_USER_ATTR", "JsonlOptimizationLogger", + "MultiSeedOptimization", "ObjectiveResult", "OptimizationConfig", "OptimizationResult", "OptimizationTrialRecord", "OptunaOptimizer", + "RobustSelectionConfig", "SamplerConfig", "SearchSpaceInfo", "SingleObjectiveEarlyStopping", diff --git a/docs/optimization.md b/docs/optimization.md index 5352e8d..919b407 100644 --- a/docs/optimization.md +++ b/docs/optimization.md @@ -25,6 +25,8 @@ from quantbt import ( ReportMetricObjective, SharpeObjective, CandidateSelector, + RobustSelectionConfig, + MultiSeedOptimization, ) ``` @@ -195,6 +197,72 @@ unseeded sampler behavior, matching `optuna.samplers.TPESampler()` defaults. This is useful for exploratory legacy-style searches. Keep an explicit integer seed for reproducible studies and stakeholder audit runs. +## Search Assurance + +Use `initial_trials` to force historical champions or hand-picked baselines to +be evaluated by the current evaluator before sampled trials: + +```python +result = optimizer.optimize( + param_ranges=param_ranges, + fixed_params={"issl": True}, + initial_trials=[ + hyhy, + hrhr, + hyhy_migrate_quantbt, + ], + candidate_selector=CandidateSelector(mode="feasible_best"), +) +``` + +Warm-start trials are tagged as `quantbt_source="warm_start"` and are exposed +through: + +```python +result.baseline_trials +result.search_diagnostics["baseline_rank"] +result.search_regression +``` + +For single-objective studies, QuantBT applies a baseline floor: if the selected +candidate is worse than the best feasible warm-start on the primary objective, +`selected_params` is reset to the warm-start baseline and +`search_regression=True`. + +When strategy params contain inactive/noisy branches, pass an +`effective_params_builder` so duplicate detection and diagnostics can use the +semantic parameter set: + +```python +def effective(params): + out = dict(params) + if not out["istp"]: + out.pop("tppercent", None) + if not out["usevol"]: + out.pop("rvol", None) + out.pop("len_vol", None) + return out + +result = optimizer.optimize( + param_ranges=param_ranges, + fixed_params=fixed, + initial_trials=[hyhy], + effective_params_builder=effective, +) +``` + +Use `early_stopping_min_trials` when early stopping is enabled so the study +cannot stop before a minimum exploration floor: + +```python +OptimizationConfig( + study_name="delta_rsi", + n_trials=1_500, + early_stopping_rounds=300, + early_stopping_min_trials=800, +) +``` + This fallback is intentionally used for early arbitrage, grid/DCA, and option package workflows until a specialized prepared evaluator is worth adding. @@ -335,6 +403,111 @@ selector when production params are required. `CandidateSelector(mode="pareto_first")` filters infeasible Pareto trials before selection. +### Robust Plateau Selection + +For practical alpha research, the highest Optuna trial can be an isolated +sample. QuantBT therefore exposes a post-search plateau selector: + +```python +selector = CandidateSelector( + mode="robust_plateau", + config=RobustSelectionConfig( + top_quantile=0.10, + min_trades=100, + max_drawdown_pct=25.0, + neighborhood_radius=0.10, + min_neighbor_count=8, + seed_consensus=3, + instability_penalty=0.25, + worst_weight=0.25, + drawdown_penalty=0.0, + ), +) + +result = optimizer.optimize( + param_ranges=param_ranges, + candidate_selector=selector, +) +``` + +The sampler still optimizes the raw objective. The selector runs only after the +study is complete: + +```text +completed trials +-> formal feasibility filter +-> optional metric filters such as min trades / max drawdown +-> top objective quantile +-> parameter-neighborhood scoring +-> medoid candidate from the best plateau +``` + +The robust score is selection-only: + +```text +score = + median(objective in neighborhood) + + worst_weight * worst(objective in neighborhood) + - instability_penalty * std(objective in neighborhood) + - drawdown_penalty * median(max_drawdown_pct) + + size_bonus * log(1 + neighbor_count) +``` + +This is designed to avoid selecting a single lucky spike that does not survive +nearby parameter perturbation. It does not change the objective surface seen by +Optuna. + +`result.robust_candidates` records the ranked plateau candidates, including +neighbor count, seed consensus count, objective dispersion, and selected medoid +trial number. + +### Multi-Seed Search + +One random TPE trajectory is not enough evidence that a parameter region is +stable. `MultiSeedOptimization` reruns the same evaluator under several sampler +seeds, aggregates the completed trial records, then applies the same selector: + +```python +multi = MultiSeedOptimization( + evaluator=evaluator, + config=OptimizationConfig( + study_name="delta_rsi_intrabar_multiseed", + n_trials=600, + seed=None, + show_progress_bar=False, + early_stopping_rounds=None, + ), + sampler_config=SamplerConfig( + name="tpe", + kwargs={ + "n_startup_trials": 120, + "multivariate": False, + "group": False, + }, + ), + seeds=(None, 41, 42, 43, 44), +) + +result = multi.optimize( + param_ranges=param_ranges, + fixed_params={"issl": True}, + initial_trials=[known_good_params], + candidate_selector=selector, +) +``` + +`result.seed_results` stores best/selected params per seed. Trial metadata also +contains `quantbt_seed`, so robust plateau selection can require a region to be +seen across several seeds via `seed_consensus`. + +For a normal single-study `OptunaOptimizer` result without seed metadata, the +selector treats the study as one consensus group. Set `seed_consensus > 1` only +when using aggregated multi-seed trial records. + +Warm-start baseline floor still applies: if the robust candidate is worse than +the best feasible historical baseline on the primary objective, QuantBT returns +the baseline and sets `search_regression=True`. + ## Reproducibility Safety Phase 32 final merge rules are conservative: diff --git a/optimization/__init__.py b/optimization/__init__.py index c0dab0b..399b0b9 100644 --- a/optimization/__init__.py +++ b/optimization/__init__.py @@ -1,7 +1,7 @@ """Domain-agnostic optimization API for QuantBT.""" from .callbacks import JsonlOptimizationLogger, SingleObjectiveEarlyStopping -from .candidate_selection import CandidateSelector, SelectedCandidate, constraints_feasible +from .candidate_selection import CandidateSelector, RobustSelectionConfig, SelectedCandidate, constraints_feasible from .config import OptimizationConfig, SamplerConfig from .constraints import CONSTRAINTS_USER_ATTR, constraints_from_trial, set_trial_constraints from .evaluator import TrialEvaluator @@ -31,6 +31,7 @@ result_full_report, ) from .optimizer import OptunaOptimizer +from .multiseed import MultiSeedOptimization from .result import ObjectiveResult, OptimizationResult, OptimizationTrialRecord from .samplers import build_sampler from .space import ( @@ -52,6 +53,7 @@ "GridDCATrialOutput", "JsonlOptimizationLogger", "MissingOptimizationMetricError", + "MultiSeedOptimization", "ObjectiveResult", "OptionPackageGenericEvaluator", "OptionTrialOutput", @@ -63,6 +65,7 @@ "PreparedPortfolioEvaluator", "PreparedSignalEvaluator", "ReportMetricObjective", + "RobustSelectionConfig", "SamplerConfig", "SearchSpaceInfo", "SelectedCandidate", diff --git a/optimization/callbacks.py b/optimization/callbacks.py index 2c6997b..bed2372 100644 --- a/optimization/callbacks.py +++ b/optimization/callbacks.py @@ -11,7 +11,7 @@ class SingleObjectiveEarlyStopping: """Stop a single-objective Optuna study after best-value stagnation.""" - def __init__(self, patience: int, direction: str, min_delta: float = 1e-4): + def __init__(self, patience: int, direction: str, min_delta: float = 1e-4, min_trials: int = 0): if patience <= 0: raise ValueError("patience must be positive") direction = str(direction).lower().strip() @@ -19,11 +19,15 @@ def __init__(self, patience: int, direction: str, min_delta: float = 1e-4): raise ValueError("direction must be maximize or minimize") if min_delta < 0.0: raise ValueError("min_delta must be >= 0") + if min_trials < 0: + raise ValueError("min_trials must be >= 0") self.patience = int(patience) self.direction = direction self.min_delta = float(min_delta) + self.min_trials = int(min_trials) self._best: Optional[float] = None self._stale = 0 + self._completed = 0 def __call__(self, study, trial) -> None: try: @@ -36,12 +40,13 @@ def __call__(self, study, trial) -> None: current = float(study.best_value) except Exception: return + self._completed += 1 if self._is_improved(current): self._best = current self._stale = 0 else: self._stale += 1 - if self._stale >= self.patience: + if self._completed >= self.min_trials and self._stale >= self.patience: study.stop() def _is_improved(self, current: float) -> bool: diff --git a/optimization/candidate_selection.py b/optimization/candidate_selection.py index f7c07a8..2583941 100644 --- a/optimization/candidate_selection.py +++ b/optimization/candidate_selection.py @@ -2,8 +2,10 @@ from __future__ import annotations +import math from dataclasses import dataclass, field -from typing import Any, Optional +from statistics import median +from typing import Any, Iterable, Optional from .result import OptimizationResult, OptimizationTrialRecord @@ -25,6 +27,39 @@ class SelectedCandidate: metadata: dict[str, Any] = field(default_factory=dict) +@dataclass(frozen=True) +class RobustSelectionConfig: + """Configuration for plateau-based production candidate selection. + + The selector is deliberately post-optimization: the sampler still learns + from the raw objective surface, while the final production params are + selected from a stable feasible neighborhood instead of a single spike. + """ + + top_quantile: float = 0.10 + min_trades: Optional[float] = None + max_drawdown_pct: Optional[float] = None + neighborhood_radius: float = 0.10 + min_neighbor_count: int = 3 + seed_consensus: int = 1 + instability_penalty: float = 0.25 + worst_weight: float = 0.25 + drawdown_penalty: float = 0.0 + size_bonus: float = 0.01 + ignore_params: tuple[str, ...] = () + + def __post_init__(self) -> None: + if not (0.0 < float(self.top_quantile) <= 1.0): + raise ValueError("top_quantile must be in (0, 1]") + if float(self.neighborhood_radius) < 0.0: + raise ValueError("neighborhood_radius must be non-negative") + if int(self.min_neighbor_count) <= 0: + raise ValueError("min_neighbor_count must be positive") + if int(self.seed_consensus) <= 0: + raise ValueError("seed_consensus must be positive") + object.__setattr__(self, "ignore_params", tuple(str(name) for name in self.ignore_params)) + + @dataclass(frozen=True) class CandidateSelector: """Small public selector interface. @@ -36,6 +71,7 @@ class CandidateSelector: mode: str = "feasible_best" objective_index: int = 0 + config: Optional[RobustSelectionConfig] = None def select(self, result: OptimizationResult) -> SelectedCandidate: mode = str(self.mode).lower().strip() @@ -45,6 +81,8 @@ def select(self, result: OptimizationResult) -> SelectedCandidate: return self._single_best(result, require_feasible=True) if mode in {"pareto_first", "first_pareto"}: return self._pareto_first(result) + if mode in {"robust_plateau", "plateau_robust"}: + return self._robust_plateau(result) raise ValueError(f"unsupported candidate selector mode={self.mode!r}") def _single_best(self, result: OptimizationResult, *, require_feasible: bool) -> SelectedCandidate: @@ -84,6 +122,95 @@ def _pareto_first(self, result: OptimizationResult) -> SelectedCandidate: }, ) + def _robust_plateau(self, result: OptimizationResult) -> SelectedCandidate: + config = self.config or RobustSelectionConfig() + objective_index = int(self.objective_index) + direction = _direction(result, objective_index) + feasible = [ + record + for record in result.trials + if _record_feasible_for_robust(record, objective_index=objective_index, config=config) + ] + if not feasible: + raise ValueError("no completed feasible optimization trials for robust plateau selection") + + ranked = sorted( + feasible, + key=lambda record: _signed_objective(record, objective_index, direction), + reverse=True, + ) + top_n = max( + 1, + int(math.ceil(len(ranked) * float(config.top_quantile))), + min(int(config.min_neighbor_count), len(ranked)), + ) + top_n = min(top_n, len(ranked)) + top = ranked[:top_n] + param_names = _param_names(feasible, ignore=config.ignore_params) + fallback_reasons: list[str] = [] + scored: list[dict[str, Any]] = [] + for record in top: + neighbors = [ + neighbor + for neighbor in top + if _param_distance(record.params, neighbor.params, feasible, param_names) <= float(config.neighborhood_radius) + ] + if not neighbors: + neighbors = [record] + seed_count = _seed_consensus_count(neighbors) + meets_count = len(neighbors) >= int(config.min_neighbor_count) + meets_seed = seed_count >= int(config.seed_consensus) + if meets_count and meets_seed: + scored.append(_score_neighborhood(record, neighbors, objective_index, direction, config, feasible, param_names)) + + if not scored: + fallback_reasons.append("no_candidate_met_neighbor_or_seed_consensus") + for record in top: + neighbors = [ + neighbor + for neighbor in top + if _param_distance(record.params, neighbor.params, feasible, param_names) <= float(config.neighborhood_radius) + ] or [record] + scored.append(_score_neighborhood(record, neighbors, objective_index, direction, config, feasible, param_names)) + + best_cluster = sorted(scored, key=lambda row: row["plateau_score"], reverse=True)[0] + selected_record = _medoid_record(best_cluster["neighbors"], feasible, param_names, objective_index, direction) + result.robust_candidates = [ + { + "trial_number": int(row["center"].number), + "plateau_score": float(row["plateau_score"]), + "neighbor_count": int(len(row["neighbors"])), + "seed_consensus_count": int(row["seed_consensus_count"]), + "median_objective": float(row["median_objective"]), + "worst_objective": float(row["worst_objective"]), + "objective_std": float(row["objective_std"]), + "params": dict(row["center"].params), + } + for row in sorted(scored, key=lambda item: item["plateau_score"], reverse=True) + ] + metadata = { + "selector": self.mode, + "selected_by": "robust_plateau", + "objective_index": objective_index, + "top_quantile": float(config.top_quantile), + "top_trials": int(top_n), + "feasible_trials": int(len(feasible)), + "neighborhood_radius": float(config.neighborhood_radius), + "min_neighbor_count": int(config.min_neighbor_count), + "seed_consensus": int(config.seed_consensus), + "seed_consensus_count": int(best_cluster["seed_consensus_count"]), + "neighbor_count": int(len(best_cluster["neighbors"])), + "plateau_score": float(best_cluster["plateau_score"]), + "median_objective": float(best_cluster["median_objective"]), + "worst_objective": float(best_cluster["worst_objective"]), + "objective_std": float(best_cluster["objective_std"]), + "cluster_center_trial": int(best_cluster["center"].number), + "medoid_trial_number": int(selected_record.number), + "fallback_reasons": fallback_reasons, + "param_names": param_names, + } + return _selected_from_record(selected_record, metadata=metadata) + def _selected_from_record(record: OptimizationTrialRecord, *, metadata: Optional[dict[str, Any]] = None) -> SelectedCandidate: merged_metadata = dict(record.metadata) @@ -106,3 +233,165 @@ def _direction(result: OptimizationResult, objective_index: int) -> str: if objective_index < 0 or objective_index >= len(directions): raise ValueError("objective_index out of range for optimization directions") return directions[objective_index] + + +def _record_feasible_for_robust( + record: OptimizationTrialRecord, + *, + objective_index: int, + config: RobustSelectionConfig, +) -> bool: + if record.state != "COMPLETE" or len(record.values) <= int(objective_index): + return False + if not constraints_feasible(record.constraints): + return False + if config.min_trades is not None: + trades = _metric(record, ("num_trades", "trades", "trade_count")) + if trades is None or float(trades) < float(config.min_trades): + return False + if config.max_drawdown_pct is not None: + mdd = _metric(record, ("max_drawdown_pct", "mdd_pct", "max_dd_pct")) + if mdd is None or float(mdd) > float(config.max_drawdown_pct): + return False + return True + + +def _metric(record: OptimizationTrialRecord, names: Iterable[str]) -> Optional[float]: + for name in names: + if name in record.metrics: + return float(record.metrics[name]) + return None + + +def _signed_objective(record: OptimizationTrialRecord, objective_index: int, direction: str) -> float: + value = float(record.values[int(objective_index)]) + if direction == "minimize": + return -value + return value + + +def _param_names(records: Iterable[OptimizationTrialRecord], *, ignore: tuple[str, ...]) -> list[str]: + ignored = set(ignore) + names: set[str] = set() + for record in records: + names.update(str(name) for name in record.params if str(name) not in ignored) + return sorted(names) + + +def _param_distance( + left: dict[str, Any], + right: dict[str, Any], + records: Iterable[OptimizationTrialRecord], + param_names: list[str], +) -> float: + if not param_names: + return 0.0 + total = 0.0 + for name in param_names: + lv = left.get(name) + rv = right.get(name) + if _is_numeric(lv) and _is_numeric(rv): + span = _numeric_span(records, name) + diff = 0.0 if span <= 0.0 else abs(float(lv) - float(rv)) / span + else: + diff = 0.0 if lv == rv else 1.0 + total += diff * diff + return math.sqrt(total / len(param_names)) + + +def _numeric_span(records: Iterable[OptimizationTrialRecord], name: str) -> float: + values = [float(record.params[name]) for record in records if name in record.params and _is_numeric(record.params[name])] + if not values: + return 0.0 + return float(max(values) - min(values)) + + +def _is_numeric(value: Any) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)) + + +def _seed_labels(records: Iterable[OptimizationTrialRecord]) -> set[str]: + labels = set() + for record in records: + for key in ("quantbt_seed", "seed"): + if key in record.metadata: + labels.add(str(record.metadata[key])) + break + return labels + + +def _seed_consensus_count(records: Iterable[OptimizationTrialRecord]) -> int: + records = list(records) + labels = _seed_labels(records) + if labels: + return len(labels) + return 1 if records else 0 + + +def _score_neighborhood( + center: OptimizationTrialRecord, + neighbors: list[OptimizationTrialRecord], + objective_index: int, + direction: str, + config: RobustSelectionConfig, + all_records: list[OptimizationTrialRecord], + param_names: list[str], +) -> dict[str, Any]: + signed = [_signed_objective(record, objective_index, direction) for record in neighbors] + med = float(median(signed)) + worst = float(min(signed)) + std = _std(signed) + mdds = [_metric(record, ("max_drawdown_pct", "mdd_pct", "max_dd_pct")) for record in neighbors] + mdd_penalty = float(median([float(value) for value in mdds if value is not None])) if any(value is not None for value in mdds) else 0.0 + score = ( + med + + float(config.worst_weight) * worst + - float(config.instability_penalty) * std + - float(config.drawdown_penalty) * mdd_penalty + + float(config.size_bonus) * math.log1p(len(neighbors)) + ) + return { + "center": center, + "neighbors": neighbors, + "plateau_score": float(score), + "median_objective": med, + "worst_objective": worst, + "objective_std": std, + "seed_consensus_count": _seed_consensus_count(neighbors), + "mean_distance": _mean_distance(center, neighbors, all_records, param_names), + } + + +def _std(values: list[float]) -> float: + if len(values) <= 1: + return 0.0 + mean = sum(values) / len(values) + return math.sqrt(sum((value - mean) ** 2 for value in values) / len(values)) + + +def _mean_distance( + center: OptimizationTrialRecord, + neighbors: list[OptimizationTrialRecord], + records: list[OptimizationTrialRecord], + param_names: list[str], +) -> float: + if not neighbors: + return 0.0 + return sum(_param_distance(center.params, record.params, records, param_names) for record in neighbors) / len(neighbors) + + +def _medoid_record( + neighbors: list[OptimizationTrialRecord], + all_records: list[OptimizationTrialRecord], + param_names: list[str], + objective_index: int, + direction: str, +) -> OptimizationTrialRecord: + return sorted( + neighbors, + key=lambda record: ( + _mean_distance(record, neighbors, all_records, param_names), + -_signed_objective(record, objective_index, direction), + int(record.number), + ), + )[0] diff --git a/optimization/config.py b/optimization/config.py index da9b345..ab61560 100644 --- a/optimization/config.py +++ b/optimization/config.py @@ -25,6 +25,7 @@ class OptimizationConfig: seed: Optional[int] = 42 n_jobs: int = 1 early_stopping_rounds: Optional[int] = None + early_stopping_min_trials: int = 0 early_stopping_min_delta: float = 1e-4 show_progress_bar: bool = True storage: Optional[str] = None @@ -49,6 +50,8 @@ def __post_init__(self) -> None: raise ValueError("n_jobs must be positive") if self.early_stopping_rounds is not None and self.early_stopping_rounds <= 0: raise ValueError("early_stopping_rounds must be positive when provided") + if self.early_stopping_min_trials < 0: + raise ValueError("early_stopping_min_trials must be >= 0") if self.early_stopping_min_delta < 0.0: raise ValueError("early_stopping_min_delta must be >= 0") duplicate_policy = str(self.duplicate_policy).lower().strip() diff --git a/optimization/multiseed.py b/optimization/multiseed.py new file mode 100644 index 0000000..e3e42f4 --- /dev/null +++ b/optimization/multiseed.py @@ -0,0 +1,176 @@ +"""Multi-seed optimization orchestration.""" + +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from typing import Any, Callable, Mapping, Optional, Sequence + +from .candidate_selection import CandidateSelector, RobustSelectionConfig +from .config import OptimizationConfig, SamplerConfig +from .evaluator import TrialEvaluator +from .optimizer import OptunaOptimizer, _apply_baseline_floor, _is_better +from .result import OptimizationResult, OptimizationTrialRecord + + +@dataclass(frozen=True) +class MultiSeedOptimization: + """Run the same search across several sampler seeds and aggregate trials. + + This is a search-quality tool, not a different objective. Each seed still + optimizes the same evaluator; the aggregate result then selects production + params from regions that survive multiple random trajectories. + """ + + evaluator: TrialEvaluator + config: OptimizationConfig + sampler_config: SamplerConfig = field(default_factory=SamplerConfig) + seeds: Sequence[Optional[int]] = (None, 41, 42, 43, 44) + trials_per_seed: Optional[int] = None + + def optimize( + self, + *, + param_ranges: Mapping[str, Any], + fixed_params: Optional[Mapping[str, Any]] = None, + initial_trials: Optional[Sequence[Mapping[str, Any]]] = None, + effective_params_builder: Optional[Callable[[Mapping[str, Any]], Mapping[str, Any]]] = None, + candidate_selector: Optional[CandidateSelector] = None, + ) -> OptimizationResult: + if not self.seeds: + raise ValueError("MultiSeedOptimization.seeds must be non-empty") + + seed_results: list[OptimizationResult] = [] + combined_trials: list[OptimizationTrialRecord] = [] + seed_summaries: list[dict[str, Any]] = [] + global_number = 0 + for seed_index, seed in enumerate(self.seeds): + seed_label = "none" if seed is None else str(seed) + config = replace( + self.config, + seed=seed, + n_trials=int(self.trials_per_seed or self.config.n_trials), + study_name=f"{self.config.study_name}_seed_{seed_label}", + ) + result = OptunaOptimizer( + evaluator=self.evaluator, + config=config, + sampler_config=self.sampler_config, + ).optimize( + param_ranges=param_ranges, + fixed_params=fixed_params, + initial_trials=initial_trials, + effective_params_builder=effective_params_builder, + ) + seed_results.append(result) + seed_summaries.append(_seed_summary(result, seed=seed, seed_index=seed_index)) + for record in result.trials: + metadata = dict(record.metadata) + metadata.update( + { + "quantbt_seed": seed_label, + "quantbt_seed_index": int(seed_index), + "quantbt_original_trial_number": int(record.number), + } + ) + combined_trials.append( + OptimizationTrialRecord( + number=int(global_number), + state=str(record.state), + params=dict(record.params), + values=tuple(record.values), + metrics=dict(record.metrics), + constraints=tuple(record.constraints), + metadata=metadata, + ) + ) + global_number += 1 + + study_view = _StudyDirectionsView(seed_results[0].study.directions) + aggregate = OptimizationResult( + study=study_view, + best_params=None, + best_values=None, + pareto_trials=[], + trials=combined_trials, + trials_frame=None, + ) + aggregate.baseline_trials = [ + record + for record in combined_trials + if record.metadata.get("quantbt_source") == "warm_start" + ] + aggregate.seed_results = seed_summaries + _set_best_from_trials(aggregate) + + selector = candidate_selector or CandidateSelector( + mode="robust_plateau", + config=RobustSelectionConfig(seed_consensus=min(2, len(self.seeds))), + ) + selected = selector.select(aggregate) + aggregate.selected_params = dict(selected.params) + aggregate.selection_metadata = dict(selected.metadata) + aggregate.selection_metadata.update( + { + "selected_by_multiseed": True, + "seed_count": int(len(self.seeds)), + } + ) + aggregate.search_diagnostics = { + "seed_count": int(len(self.seeds)), + "seed_results": seed_summaries, + "completed_trials": int(sum(1 for record in combined_trials if record.state == "COMPLETE")), + "pruned_trials": int(sum(1 for record in combined_trials if record.state == "PRUNED")), + "failed_trials": int(sum(1 for record in combined_trials if record.state == "FAIL")), + "top_parameter_frequency": _top_parameter_frequency(combined_trials), + } + _apply_baseline_floor(aggregate) + return aggregate + + +def _set_best_from_trials(result: OptimizationResult) -> None: + try: + direction = str(result.study.directions[0].name).lower() + except Exception: + direction = "maximize" + completed = [record for record in result.trials if record.state == "COMPLETE" and record.values] + if not completed: + return + best = completed[0] + for record in completed[1:]: + if _is_better(record.values[0], best.values[0], direction): + best = record + result.best_params = dict(best.params) + result.best_values = tuple(best.values) + + +def _seed_summary(result: OptimizationResult, *, seed: Optional[int], seed_index: int) -> dict[str, Any]: + return { + "seed": None if seed is None else int(seed), + "seed_index": int(seed_index), + "best_params": None if result.best_params is None else dict(result.best_params), + "best_values": None if result.best_values is None else tuple(float(value) for value in result.best_values), + "selected_params": None if result.selected_params is None else dict(result.selected_params), + "search_regression": bool(result.search_regression), + "baseline_rank": list(result.search_diagnostics.get("baseline_rank", [])), + "completed_trials": int(result.search_diagnostics.get("completed_trials", 0)), + } + + +def _top_parameter_frequency(records: Sequence[OptimizationTrialRecord]) -> dict[str, dict[str, int]]: + completed = [record for record in records if record.state == "COMPLETE" and record.values] + if not completed: + return {} + ranked = sorted(completed, key=lambda record: record.values[0], reverse=True) + top_n = max(1, len(ranked) // 10) + counts: dict[str, dict[str, int]] = {} + for record in ranked[:top_n]: + for name, value in record.params.items(): + bucket = counts.setdefault(str(name), {}) + label = str(value) + bucket[label] = bucket.get(label, 0) + 1 + return counts + + +class _StudyDirectionsView: + def __init__(self, directions: Sequence[Any]): + self.directions = tuple(directions) diff --git a/optimization/optimizer.py b/optimization/optimizer.py index 4fbd276..b420320 100644 --- a/optimization/optimizer.py +++ b/optimization/optimizer.py @@ -3,16 +3,16 @@ from __future__ import annotations import math -from typing import Any, Mapping, Optional +from typing import Any, Callable, Mapping, Optional, Sequence from .callbacks import JsonlOptimizationLogger, SingleObjectiveEarlyStopping -from .candidate_selection import CandidateSelector +from .candidate_selection import constraints_feasible from .config import OptimizationConfig, SamplerConfig from .constraints import constraints_from_trial, set_trial_constraints from .evaluator import TrialEvaluator from .result import ObjectiveResult, OptimizationResult, OptimizationTrialRecord from .samplers import build_sampler -from .space import stable_params_key, suggest_params +from .space import search_space_info, stable_params_key, suggest_params class OptunaOptimizer: @@ -35,6 +35,8 @@ def optimize( *, param_ranges: Mapping[str, Any], fixed_params: Optional[Mapping[str, Any]] = None, + initial_trials: Optional[Sequence[Mapping[str, Any]]] = None, + effective_params_builder: Optional[Callable[[Mapping[str, Any]], Mapping[str, Any]]] = None, candidate_selector=None, ) -> OptimizationResult: """Run an Optuna study and return a QuantBT result schema.""" @@ -69,6 +71,12 @@ def optimize( pruner=optuna.pruners.NopPruner(), ) self._preload_seen_params(study) + self._enqueue_initial_trials( + study, + param_ranges=param_ranges, + fixed_params=fixed_params, + initial_trials=initial_trials, + ) callbacks = [] if self.config.early_stopping_rounds is not None: if objective_count != 1: @@ -78,6 +86,7 @@ def optimize( self.config.early_stopping_rounds, self.config.directions[0], min_delta=float(self.config.early_stopping_min_delta), + min_trials=int(self.config.early_stopping_min_trials), ) ) if self.config.log_path is not None: @@ -85,7 +94,13 @@ def optimize( catch = (Exception,) if self.config.exception_policy == "fail_trial" else () study.optimize( - lambda trial: self._objective(trial, param_ranges, fixed_params, objective_count), + lambda trial: self._objective( + trial, + param_ranges, + fixed_params, + objective_count, + effective_params_builder=effective_params_builder, + ), n_trials=int(self.config.n_trials), n_jobs=int(self.config.n_jobs), callbacks=callbacks, @@ -93,6 +108,17 @@ def optimize( catch=catch, ) result = _build_result(study, objective_count) + result.baseline_trials = [ + record + for record in result.trials + if record.metadata.get("quantbt_source") == "warm_start" + ] + result.search_diagnostics = _search_diagnostics( + param_ranges=param_ranges, + fixed_params=fixed_params, + result=result, + objective_index=0, + ) if candidate_selector is not None: selected = candidate_selector.select(result) result.selected_params = dict(getattr(selected, "params", selected)) @@ -103,8 +129,27 @@ def optimize( result.selection_metadata = {"selected_by": None, "reason": "constraints_require_explicit_candidate_selector"} else: result.selected_params = dict(result.best_params or {}) + _apply_baseline_floor(result) return result + def _enqueue_initial_trials(self, study, *, param_ranges, fixed_params, initial_trials) -> None: + if not initial_trials: + return + fixed = dict(fixed_params or {}) + for idx, payload in enumerate(initial_trials): + full_params = dict(payload or {}) + full_params.update(fixed) + trial_params = _trial_params_for_enqueue(full_params, param_ranges, fixed) + study.enqueue_trial( + trial_params, + user_attrs={ + "quantbt_source": "warm_start", + "quantbt_initial_trial_id": int(idx), + "quantbt_initial_full_params": dict(full_params), + }, + skip_if_exists=True, + ) + def _preload_seen_params(self, study) -> None: if not self.config.load_if_exists: return @@ -117,15 +162,21 @@ def _preload_seen_params(self, study) -> None: if key: self._seen_params.add(str(key)) - def _objective(self, trial, param_ranges, fixed_params, objective_count: int): + def _objective(self, trial, param_ranges, fixed_params, objective_count: int, *, effective_params_builder=None): try: import optuna except Exception as exc: # pragma: no cover raise ImportError("QuantBT optimization requires optuna") from exc params = suggest_params(trial, param_ranges, fixed_params=fixed_params) - params_key = stable_params_key(params) + source = str(trial.user_attrs.get("quantbt_source", "sampled")) + raw_params_key = stable_params_key(params) + effective_params = dict(effective_params_builder(params)) if effective_params_builder is not None else dict(params) + params_key = stable_params_key(effective_params) trial.set_user_attr("quantbt_full_params", dict(params)) + trial.set_user_attr("quantbt_source", source) trial.set_user_attr("quantbt_params_key", params_key) + trial.set_user_attr("quantbt_raw_params_key", raw_params_key) + trial.set_user_attr("quantbt_effective_params", dict(effective_params)) if params_key in self._seen_params: if self.config.duplicate_policy == "prune": raise optuna.TrialPruned("duplicate parameter set") @@ -154,7 +205,11 @@ def _objective(self, trial, param_ranges, fixed_params, objective_count: int): raise optuna.TrialPruned("non-finite objective value") trial.set_user_attr("quantbt_metrics", dict(objective.metrics)) - trial.set_user_attr("quantbt_metadata", dict(objective.metadata)) + metadata = dict(objective.metadata) + metadata.setdefault("quantbt_source", source) + metadata.setdefault("quantbt_params_key", params_key) + metadata.setdefault("quantbt_raw_params_key", raw_params_key) + trial.set_user_attr("quantbt_metadata", metadata) set_trial_constraints(trial, objective.constraints) if objective_count == 1: @@ -197,6 +252,15 @@ def _result_has_constraints(result: OptimizationResult) -> bool: def _trial_record(trial) -> OptimizationTrialRecord: values = tuple(float(value) for value in (trial.values or ())) + metadata = dict(trial.user_attrs.get("quantbt_metadata", {})) + for key in ( + "quantbt_source", + "quantbt_params_key", + "quantbt_raw_params_key", + "quantbt_initial_trial_id", + ): + if key in trial.user_attrs: + metadata.setdefault(key, trial.user_attrs[key]) return OptimizationTrialRecord( number=int(trial.number), state=str(trial.state.name), @@ -204,5 +268,233 @@ def _trial_record(trial) -> OptimizationTrialRecord: values=values, metrics=dict(trial.user_attrs.get("quantbt_metrics", {})), constraints=tuple(float(value) for value in trial.user_attrs.get("quantbt_constraints", ())), - metadata=dict(trial.user_attrs.get("quantbt_metadata", {})), + metadata=metadata, + ) + + +def _trial_params_for_enqueue(params: Mapping[str, Any], param_ranges: Mapping[str, Any], fixed_params: Mapping[str, Any]) -> dict[str, Any]: + """Return only Optuna-suggested params for `study.enqueue_trial`. + + Scalar constants and fixed params are merged inside `suggest_params`, so + enqueuing them would create confusing Optuna distributions. Missing active + search params are rejected because a warm-start baseline must be evaluated + exactly, not partially sampled. + """ + + queued: dict[str, Any] = {} + missing: list[str] = [] + fixed = set(dict(fixed_params or {})) + for name, spec in dict(param_ranges or {}).items(): + if name in fixed or not _is_suggested_spec(spec): + continue + if name not in params: + missing.append(str(name)) + else: + queued[str(name)] = params[name] + if missing: + joined = ", ".join(missing[:10]) + raise ValueError(f"initial trial is missing search params: {joined}") + return queued + + +def _is_suggested_spec(spec: Any) -> bool: + if isinstance(spec, range): + return True + if isinstance(spec, tuple) and len(spec) in (2, 3): + return True + if isinstance(spec, list): + return True + return False + + +def _apply_baseline_floor(result: OptimizationResult) -> None: + """Keep the best feasible warm-start when selected candidate regresses.""" + + try: + directions = tuple(str(direction.name).lower() for direction in result.study.directions) + except Exception: + directions = ("maximize",) + if len(directions) != 1: + return + baselines = [ + record + for record in result.baseline_trials + if record.state == "COMPLETE" and record.values and constraints_feasible(record.constraints) + ] + if not baselines: + result.search_regression = False + result.selection_metadata.setdefault("best_baseline_trial", None) + return + best_baseline = sorted( + baselines, + key=lambda record: record.values[0], + reverse=directions[0] == "maximize", + )[0] + selected = _selected_record(result) + if selected is None: + selected = best_baseline + selected_value = selected.values[0] if selected.values else float("-inf") + baseline_better = _is_better(best_baseline.values[0], selected_value, directions[0]) + result.selection_metadata.setdefault( + "best_baseline_trial", + { + "trial_number": int(best_baseline.number), + "value": float(best_baseline.values[0]), + "params": dict(best_baseline.params), + }, ) + if not baseline_better: + result.search_regression = False + result.selection_metadata.setdefault("search_regression", False) + return + result.selected_params = dict(best_baseline.params) + result.search_regression = True + result.selection_metadata.update( + { + "selected_by": "warm_start_baseline_floor", + "search_regression": True, + "previous_selected_trial": None if selected is None else int(selected.number), + "previous_selected_value": None if selected is None or not selected.values else float(selected.values[0]), + "trial_number": int(best_baseline.number), + "value": float(best_baseline.values[0]), + } + ) + + +def _selected_record(result: OptimizationResult) -> Optional[OptimizationTrialRecord]: + trial_number = result.selection_metadata.get("trial_number") + if trial_number is not None: + for record in result.trials: + if int(record.number) == int(trial_number): + return record + if result.selected_params is not None: + selected_key = stable_params_key(result.selected_params) + for record in result.trials: + if stable_params_key(record.params) == selected_key and record.state == "COMPLETE": + return record + if result.best_params is not None: + best_key = stable_params_key(result.best_params) + for record in result.trials: + if stable_params_key(record.params) == best_key and record.state == "COMPLETE": + return record + return None + + +def _is_better(candidate: float, incumbent: float, direction: str) -> bool: + if direction == "minimize": + return float(candidate) < float(incumbent) + return float(candidate) > float(incumbent) + + +def _search_diagnostics( + *, + param_ranges: Mapping[str, Any], + fixed_params: Optional[Mapping[str, Any]], + result: OptimizationResult, + objective_index: int, +) -> dict[str, Any]: + info = search_space_info(param_ranges, fixed_params=fixed_params) + variable_names = list(info.variable_names) + completed = [ + record + for record in result.trials + if record.state == "COMPLETE" and len(record.values) > int(objective_index) + ] + try: + direction = str(result.study.directions[int(objective_index)].name).lower() + except Exception: + direction = "maximize" + ranked = sorted( + completed, + key=lambda record: record.values[int(objective_index)], + reverse=direction == "maximize", + ) + top_n = max(1, int(math.ceil(len(ranked) * 0.10))) if ranked else 0 + top = ranked[:top_n] + source_counts: dict[str, int] = {} + effective_keys: list[str] = [] + for record in result.trials: + source = str(record.metadata.get("quantbt_source", "sampled")) + source_counts[source] = source_counts.get(source, 0) + 1 + key = record.metadata.get("quantbt_params_key") + if key is not None: + effective_keys.append(str(key)) + coverage = { + name: len({record.params.get(name) for record in completed if name in record.params}) + for name in variable_names + } + return { + "nominal_dimension": int(len(variable_names)), + "variable_names": variable_names, + "grid_size_estimate": info.grid_size, + "has_categorical": bool(info.has_categorical), + "has_continuous": bool(info.has_continuous), + "has_dynamic_float": bool(info.has_dynamic_float), + "param_kind_counts": _param_kind_counts(param_ranges, fixed_params), + "completed_trials": int(len(completed)), + "pruned_trials": int(sum(1 for record in result.trials if record.state == "PRUNED")), + "failed_trials": int(sum(1 for record in result.trials if record.state == "FAIL")), + "source_counts": source_counts, + "effective_duplicate_count": int(len(effective_keys) - len(set(effective_keys))), + "param_coverage": coverage, + "top_decile_size": int(top_n), + "top_decile_distributions": _top_distributions(top, variable_names), + "baseline_rank": _baseline_rank(ranked), + } + + +def _param_kind_counts(param_ranges: Mapping[str, Any], fixed_params: Optional[Mapping[str, Any]]) -> dict[str, int]: + fixed = set(dict(fixed_params or {})) + counts = {"fixed": 0, "categorical": 0, "int": 0, "float": 0, "constant": 0} + for name, spec in dict(param_ranges or {}).items(): + if name in fixed: + counts["fixed"] += 1 + continue + if isinstance(spec, range) or isinstance(spec, list): + counts["categorical"] += 1 + elif isinstance(spec, tuple) and len(spec) in (2, 3): + numeric = all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in spec) + looks_int = numeric and all(isinstance(value, int) and not isinstance(value, bool) for value in spec) + counts["int" if looks_int else "float"] += 1 + else: + counts["constant"] += 1 + return counts + + +def _top_distributions(records: Sequence[OptimizationTrialRecord], variable_names: Sequence[str]) -> dict[str, dict[str, Any]]: + distributions: dict[str, dict[str, Any]] = {} + for name in variable_names: + values = [record.params.get(name) for record in records if name in record.params] + counts: dict[str, int] = {} + numeric: list[float] = [] + for value in values: + counts[str(value)] = counts.get(str(value), 0) + 1 + if isinstance(value, (int, float)) and not isinstance(value, bool): + numeric.append(float(value)) + payload: dict[str, Any] = {"counts": counts} + if numeric: + payload.update( + { + "min": float(min(numeric)), + "max": float(max(numeric)), + "mean": float(sum(numeric) / len(numeric)), + } + ) + distributions[str(name)] = payload + return distributions + + +def _baseline_rank(ranked: Sequence[OptimizationTrialRecord]) -> list[dict[str, Any]]: + rows = [] + for rank, record in enumerate(ranked, start=1): + if record.metadata.get("quantbt_source") != "warm_start": + continue + rows.append( + { + "rank": int(rank), + "trial_number": int(record.number), + "value": None if not record.values else float(record.values[0]), + "params": dict(record.params), + } + ) + return rows diff --git a/optimization/result.py b/optimization/result.py index 4a38a53..d1ecdd1 100644 --- a/optimization/result.py +++ b/optimization/result.py @@ -71,3 +71,10 @@ class OptimizationResult: trials_frame: Any selected_params: Optional[dict[str, Any]] = None selection_metadata: dict[str, Any] = field(default_factory=dict) + baseline_trials: list[OptimizationTrialRecord] = field(default_factory=list) + phase_results: list[Any] = field(default_factory=list) + seed_results: list[Any] = field(default_factory=list) + robust_candidates: list[Any] = field(default_factory=list) + selected_validation: dict[str, Any] = field(default_factory=dict) + search_regression: bool = False + search_diagnostics: dict[str, Any] = field(default_factory=dict) diff --git a/tests/test_optimization_core.py b/tests/test_optimization_core.py index 55938a5..fdc71d5 100644 --- a/tests/test_optimization_core.py +++ b/tests/test_optimization_core.py @@ -123,6 +123,84 @@ def test_optuna_optimizer_accepts_unseeded_sampler(): assert len(result.trials) == 4 +def test_initial_trials_are_evaluated_first_and_recorded_as_baselines(): + evaluator = QuadraticEvaluator() + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="initial_trials_core", n_trials=4, seed=7, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + result = optimizer.optimize( + param_ranges={"x": (0, 6, 1)}, + initial_trials=[{"x": 3}], + ) + + assert evaluator.calls[0]["x"] == 3 + assert result.baseline_trials + assert result.baseline_trials[0].metadata["quantbt_source"] == "warm_start" + assert result.search_diagnostics["source_counts"]["warm_start"] == 1 + assert result.search_diagnostics["baseline_rank"][0]["rank"] == 1 + + +def test_baseline_floor_keeps_warm_start_when_selector_prefers_worse_sample(): + class FirstSampleSelector: + def select(self, result): + from quantbt.optimization import SelectedCandidate + + sampled = next(record for record in result.trials if record.metadata.get("quantbt_source") != "warm_start" and record.state == "COMPLETE") + return SelectedCandidate( + params=dict(sampled.params), + values=tuple(sampled.values), + metrics=dict(sampled.metrics), + constraints=tuple(sampled.constraints), + metadata={"selector": "first_sample", "trial_number": sampled.number}, + ) + + evaluator = QuadraticEvaluator() + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="baseline_floor_core", n_trials=3, seed=2, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + result = optimizer.optimize( + param_ranges={"x": (0, 6, 1)}, + initial_trials=[{"x": 3}], + candidate_selector=FirstSampleSelector(), + ) + + assert result.selected_params == {"x": 3} + assert result.search_regression is True + assert result.selection_metadata["selected_by"] == "warm_start_baseline_floor" + + +def test_effective_params_builder_prunes_semantic_duplicates(): + class ToggleEvaluator: + def __init__(self): + self.calls = [] + + def evaluate(self, params): + self.calls.append(dict(params)) + return ObjectiveResult.scalar(float(params["x"]), metrics={"x": params["x"]}) + + evaluator = ToggleEvaluator() + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="effective_duplicate_core", n_trials=3, seed=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + result = optimizer.optimize( + param_ranges={"x": [1], "use_filter": [False], "len_filter": [10, 20]}, + effective_params_builder=lambda params: {"x": params["x"], "use_filter": params["use_filter"]}, + ) + + assert len(evaluator.calls) == 1 + assert [record.state for record in result.trials].count("PRUNED") == 2 + assert result.search_diagnostics["effective_duplicate_count"] == 2 + + def test_constraint_storage(): class ConstraintEvaluator: def evaluate(self, params): @@ -209,6 +287,27 @@ def test_single_objective_early_stopping_and_jsonl_logger(tmp_path): assert rows[0]["values"] == [1.0] +def test_early_stopping_respects_min_trials_floor(): + optimizer = OptunaOptimizer( + evaluator=ConstantEvaluator(1.0), + config=OptimizationConfig( + study_name="early_stop_min_trials_core", + n_trials=10, + seed=1, + early_stopping_rounds=1, + early_stopping_min_trials=4, + early_stopping_min_delta=0.0, + show_progress_bar=False, + ), + sampler_config=SamplerConfig(name="random"), + ) + + result = optimizer.optimize(param_ranges={"x": (1, 10, 1)}) + + assert len(result.trials) >= 4 + assert len(result.trials) < 10 + + def test_pruned_trials_do_not_consume_patience(): callback = SingleObjectiveEarlyStopping(patience=1, direction="maximize") study = optuna.create_study(direction="maximize") diff --git a/tests/test_optimization_phase33b.py b/tests/test_optimization_phase33b.py new file mode 100644 index 0000000..129f793 --- /dev/null +++ b/tests/test_optimization_phase33b.py @@ -0,0 +1,194 @@ +import pytest + +from quantbt.optimization import ( + CandidateSelector, + MultiSeedOptimization, + ObjectiveResult, + OptimizationConfig, + OptimizationResult, + OptimizationTrialRecord, + OptunaOptimizer, + RobustSelectionConfig, + SamplerConfig, +) + + +class _Direction: + name = "MAXIMIZE" + + +class _Study: + directions = (_Direction(),) + + +def _trial(number, x, value, *, metrics=None, constraints=(), state="COMPLETE", seed=None): + metadata = {} + if seed is not None: + metadata["quantbt_seed"] = seed + return OptimizationTrialRecord( + number=number, + state=state, + params={"x": x}, + values=(float(value),) if state == "COMPLETE" else (), + metrics=dict(metrics or {"num_trades": 200, "max_drawdown_pct": 8.0}), + constraints=tuple(constraints), + metadata=metadata, + ) + + +def _result(records): + return OptimizationResult( + study=_Study(), + best_params=dict(records[0].params), + best_values=tuple(records[0].values), + pareto_trials=[], + trials=list(records), + trials_frame=None, + ) + + +def test_robust_plateau_selector_avoids_isolated_spike(): + result = _result( + [ + _trial(0, 100.0, 10.0, seed=1), + _trial(1, -0.02, 8.00, seed=1), + _trial(2, 0.00, 7.95, seed=2), + _trial(3, 0.02, 7.90, seed=3), + _trial(4, 0.05, 7.85, seed=4), + ] + ) + + selected = CandidateSelector( + mode="robust_plateau", + config=RobustSelectionConfig( + top_quantile=1.0, + neighborhood_radius=0.05, + min_neighbor_count=3, + seed_consensus=2, + instability_penalty=0.5, + worst_weight=0.2, + ), + ).select(result) + + assert abs(float(selected.params["x"])) < 0.06 + assert selected.metadata["selected_by"] == "robust_plateau" + assert selected.metadata["neighbor_count"] >= 3 + assert selected.metadata["seed_consensus_count"] >= 2 + assert result.robust_candidates[0]["params"]["x"] != 100.0 + + +def test_robust_plateau_selector_respects_constraints_and_metric_filters(): + result = _result( + [ + _trial(0, 1.0, 12.0, metrics={"num_trades": 20, "max_drawdown_pct": 4.0}), + _trial(1, 2.0, 11.0, metrics={"num_trades": 200, "max_drawdown_pct": 40.0}), + _trial(2, 3.0, 10.0, constraints=(1.0,), metrics={"num_trades": 200, "max_drawdown_pct": 4.0}), + _trial(3, 4.0, 8.0, metrics={"num_trades": 200, "max_drawdown_pct": 4.0}), + _trial(4, 4.1, 7.9, metrics={"num_trades": 210, "max_drawdown_pct": 4.1}), + _trial(5, 4.2, 7.8, metrics={"num_trades": 220, "max_drawdown_pct": 4.2}), + ] + ) + + selected = CandidateSelector( + mode="robust_plateau", + config=RobustSelectionConfig( + top_quantile=1.0, + min_trades=100, + max_drawdown_pct=10.0, + neighborhood_radius=0.04, + min_neighbor_count=3, + ), + ).select(result) + + assert selected.params["x"] in {4.0, 4.1, 4.2} + assert selected.metrics["num_trades"] >= 100 + assert selected.metrics["max_drawdown_pct"] <= 10.0 + assert selected.metadata["feasible_trials"] == 3 + + +def test_robust_plateau_selector_raises_when_no_feasible_trials(): + result = _result([_trial(0, 1.0, 1.0, constraints=(1.0,))]) + + with pytest.raises(ValueError, match="no completed feasible"): + CandidateSelector(mode="robust_plateau").select(result) + + +class PlateauEvaluator: + def evaluate(self, params): + x = float(params["x"]) + if x == 10: + score = 10.0 + elif x in {1.0, 2.0, 3.0}: + score = 8.0 - abs(x - 2.0) * 0.05 + else: + score = 4.0 + return ObjectiveResult.scalar( + score, + metrics={"num_trades": 200 + x, "max_drawdown_pct": 5.0 + abs(x - 2.0)}, + ) + + +def test_multi_seed_optimization_selects_consensus_plateau(): + optimizer = MultiSeedOptimization( + evaluator=PlateauEvaluator(), + config=OptimizationConfig( + study_name="phase33b_multiseed", + n_trials=4, + seed=None, + show_progress_bar=False, + duplicate_policy="allow", + ), + sampler_config=SamplerConfig(name="random"), + seeds=(11, 22), + trials_per_seed=4, + ) + + result = optimizer.optimize( + param_ranges={"x": [1, 2, 3, 4, 10]}, + initial_trials=[{"x": 1}, {"x": 2}, {"x": 3}, {"x": 4}], + candidate_selector=CandidateSelector( + mode="robust_plateau", + config=RobustSelectionConfig( + top_quantile=1.0, + neighborhood_radius=0.12, + min_neighbor_count=3, + seed_consensus=2, + ), + ), + ) + + assert result.selected_params is not None + assert result.selected_params["x"] in {1, 2, 3} + assert result.selection_metadata["selected_by_multiseed"] is True + assert result.selection_metadata["seed_consensus_count"] == 2 + assert len(result.seed_results) == 2 + assert {record.metadata["quantbt_seed"] for record in result.trials if record.state == "COMPLETE"} == {"11", "22"} + + +def test_robust_selection_cannot_replace_better_warm_start_baseline(): + result = OptunaOptimizer( + evaluator=PlateauEvaluator(), + config=OptimizationConfig( + study_name="phase33b_baseline_floor", + n_trials=4, + seed=1, + show_progress_bar=False, + duplicate_policy="allow", + ), + sampler_config=SamplerConfig(name="random"), + ).optimize( + param_ranges={"x": [1, 2, 3, 10]}, + initial_trials=[{"x": 10}, {"x": 1}, {"x": 2}, {"x": 3}], + candidate_selector=CandidateSelector( + mode="robust_plateau", + config=RobustSelectionConfig( + top_quantile=1.0, + neighborhood_radius=0.12, + min_neighbor_count=3, + ), + ), + ) + + assert result.selected_params == {"x": 10} + assert result.search_regression is True + assert result.selection_metadata["selected_by"] == "warm_start_baseline_floor" diff --git a/upgrade/implement.md b/upgrade/implement.md index c16f0fd..68e9641 100644 --- a/upgrade/implement.md +++ b/upgrade/implement.md @@ -6284,6 +6284,131 @@ PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q # 513 passed, 1 skipped ``` +### Phase 33 - Optimization Search Quality And Robust Selection + +Guide: + +- Detailed source plan: + `upgrade/quantbt_optimization_search_quality_upgrade.md`. +- Scope is intentionally split into 2 phases: + - Phase 33A: Search Assurance Core. + - Phase 33B: Multi-seed search and robust plateau candidate selection. + +Reason: + +- A single TPE trajectory over mixed/conditional alpha spaces can miss known + good regions such as historical Delta-RSI champions. +- Search quality should guarantee that known baselines are evaluated by the + current evaluator and cannot be silently replaced by a worse sampled trial. +- This does not claim global optimality; it raises the optimizer from + best-trial hunting to baseline-aware, diagnostic, reproducible research. + +### Phase 33A - Search Assurance Core + +Status: implemented on `feat/domain-agnostic-optimization`. + +Implemented: + +- Added `initial_trials` to `OptunaOptimizer.optimize(...)`. + - Historical champions are enqueued with `study.enqueue_trial(...)`. + - Warm-start trials are tagged as `quantbt_source="warm_start"`. + - Fixed params are merged before enqueue validation. + - Missing active search params in a warm-start raise instead of silently + sampling a partial baseline. +- Added baseline floor for single-objective studies. + - `result.baseline_trials` records completed warm-start trials. + - If the selected candidate is worse than the best feasible warm-start, + QuantBT resets `selected_params` to that warm-start and sets + `search_regression=True`. +- Added `effective_params_builder`. + - Duplicate detection can use semantic/effective params rather than raw + noisy params. + - This is designed for alpha spaces where toggles make params inactive. +- Added `early_stopping_min_trials`. + - Early stopping cannot stop before the configured completed-trial floor. +- Added `OptimizationConfig(seed=None)`. + - This restores true Optuna unseeded behavior for legacy-style exploratory + searches while keeping integer seeds for audit runs. +- Added search diagnostics: + - nominal variable dimension; + - estimated grid size; + - param kind counts; + - source counts; + - effective duplicate count; + - per-param coverage; + - top-decile parameter distributions; + - baseline rank. +- Added docs in `docs/optimization.md`. + +Validation: + +```bash +MPLCONFIGDIR=/tmp PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q quantbt/tests/test_optimization_core.py quantbt/tests/test_optimization_samplers.py +# 23 passed + +MPLCONFIGDIR=/tmp PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q quantbt/tests/test_optimization_core.py quantbt/tests/test_optimization_samplers.py quantbt/tests/test_optimization_evaluators.py quantbt/tests/test_optimization_integration.py +# 47 passed +``` + +Phase 33A merge gates: + +- Warm-start trials are evaluated before sampled trials. +- Best feasible warm-start baseline cannot be silently lost. +- Effective duplicate detection prunes semantic duplicates. +- Early stopping respects `early_stopping_min_trials`. +- `seed=None` runs Optuna's unseeded sampler path. +- Search diagnostics persist baseline/source/coverage metadata. + +### Phase 33B - Multi-Seed Robust Plateau Candidate Selection + +Status: implemented on `feat/domain-agnostic-optimization`. + +Implemented: + +- Added `RobustSelectionConfig`. + - Controls top objective quantile, metric feasibility filters, parameter + neighborhood radius, minimum neighbor count, seed consensus, instability + penalty, worst-neighbor weight, drawdown penalty, and size bonus. +- Added `CandidateSelector(mode="robust_plateau", config=...)`. + - Filters failed/pruned/infeasible trials. + - Applies optional `min_trades` and `max_drawdown_pct` filters. + - Takes a top objective quantile instead of only the best trial. + - Scores local parameter neighborhoods by median objective, worst-neighbor + objective, objective dispersion, drawdown penalty, plateau size, and seed + consistency. + - Selects the medoid record from the best plateau rather than an isolated + spike. + - Writes ranked `result.robust_candidates` metadata. +- Added `MultiSeedOptimization`. + - Runs the same evaluator across several sampler seeds. + - Aggregates trial records with `quantbt_seed` and original trial metadata. + - Stores `result.seed_results` and seed-level diagnostics. + - Applies a robust selector over the aggregate search surface. + - Keeps the Phase 33A warm-start baseline floor, so a worse new candidate + cannot silently replace a better feasible historical baseline. +- Exported the new API from both `quantbt.optimization` and top-level + `quantbt`. +- Updated `docs/optimization.md` with robust selector and multi-seed examples. + +Validation: + +```bash +MPLCONFIGDIR=/tmp PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q quantbt/tests/test_optimization_phase33b.py quantbt/tests/test_optimization_core.py quantbt/tests/test_optimization_samplers.py quantbt/tests/test_optimization_evaluators.py quantbt/tests/test_optimization_integration.py +# 52 passed +``` + +Phase 33B merge gates: + +- Robust plateau selector does not choose an isolated spike in deterministic + mock data. +- Feasibility constraints and metric filters are respected before selection. +- Multi-seed aggregation records seed metadata and selects from a consensus + plateau. +- Historical warm-start baseline floor remains active after robust selection. +- Validation/stress gate is available through selector metadata and baseline + floor, but real alpha WFO/stress bundle validation remains a strategy-level + certification step, not a generic optimizer-core guarantee. + ## Merge Gates Do not merge unless all are true: