diff --git a/README.md b/README.md index 9c6e42f..e0db2e0 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,8 @@ historical reproduction. - Walk-forward and train/test optimization designed to avoid leaking OOS data into parameter selection, plus full-sample robust calibration for final production parameter discovery. +- Domain-agnostic Optuna optimization adapters for prepared signal, intrabar, + portfolio, and generic endpoint workflows. ## Performance Philosophy @@ -141,6 +143,23 @@ about 23.3x faster than the readable Python oracle on the committed benchmark while preserving the oracle semantics through targeted parity tests and audit second-pass checks. +Latest Phase 32C optimization overhead benchmark: + +| Measurement | Result | +|---|---:| +| Optimizer overhead | 0.0174s for 24 trials | +| Optimizer overhead / trial | 0.000723s | +| Prepared signal evaluator | 2.03x faster than normal endpoint replay | +| Intrabar first vs warm run | 3.70x first/warm ratio | +| Parity | pass, final equity diff 0.0 | + +Phase 32C consolidates safe walk-forward optimization primitives with the new +domain-agnostic optimizer core while keeping WFO fold isolation and robust +selection semantics inside `walkforward.py`. Read +[`docs/optimization.md`](docs/optimization.md) and +`benchmarks/results/optimization_overhead.md` for signal, intrabar, portfolio, +arbitrage/grid/options fallback examples and benchmark details. + Ecosystem positioning: | Tool | Core strength | Runtime model | QuantBT role beside it | @@ -241,6 +260,9 @@ service creates a run. - Full-sample robust calibration selectors: `full_robust`, `full_plateau_robust`, `full_temporal_robust`, and `full_best`. - Optional trade-count penalty to avoid overfit low-trade Sharpe traps. +- Shared domain-agnostic optimizer primitives for search-space parsing, + duplicate detection, early stopping, objective helpers, constraints, and + candidate selection. ### Nautilus Validation Reports diff --git a/__init__.py b/__init__.py index caa4935..7170e90 100644 --- a/__init__.py +++ b/__init__.py @@ -75,6 +75,51 @@ validate_walkforward_strategy_output, walkforward_support_matrix, ) +from .optimization import ( + CONSTRAINTS_USER_ATTR, + ArbitrageGenericEvaluator, + ArbitrageTrialOutput, + CandidateSelector, + GenericEndpointEvaluator, + GridDCAGenericEvaluator, + GridDCATrialOutput, + JsonlOptimizationLogger, + MissingOptimizationMetricError, + ObjectiveResult, + OptionPackageGenericEvaluator, + OptionTrialOutput, + OptimizationConfig, + OptimizationResult, + OptimizationTrialRecord, + OptunaOptimizer, + PreparedIntrabarEvaluator, + PreparedPortfolioEvaluator, + PreparedSignalEvaluator, + ReportMetricObjective, + SamplerConfig, + SearchSpaceInfo, + SelectedCandidate, + SharpeObjective, + SingleObjectiveEarlyStopping, + TrialEvaluator, + build_grid_search_space, + build_sampler, + constraints_feasible, + constraints_from_trial, + max_drawdown_constraint, + max_margin_utilization_constraint, + max_rejection_rate_constraint, + max_turnover_constraint, + metric_from_result, + metrics_from_result, + min_trades_constraint, + result_full_report, + search_space_info, + set_trial_constraints, + stable_params_key, + suggest_parameter, + suggest_params, +) from .engines import BacktestEngineV2, EventDrivenBacktestEngine, OptionBacktestEngine, PortfolioBacktestEngine from .backends import ( NativeEventBackend, @@ -537,6 +582,25 @@ "WalkForwardCompatibilityEntry", "EarlyStoppingCallback", "DuplicatePruner", + "CONSTRAINTS_USER_ATTR", + "JsonlOptimizationLogger", + "ObjectiveResult", + "OptimizationConfig", + "OptimizationResult", + "OptimizationTrialRecord", + "OptunaOptimizer", + "SamplerConfig", + "SearchSpaceInfo", + "SingleObjectiveEarlyStopping", + "TrialEvaluator", + "build_grid_search_space", + "build_sampler", + "constraints_from_trial", + "search_space_info", + "set_trial_constraints", + "stable_params_key", + "suggest_parameter", + "suggest_params", "benchmark_walkforward_kernels", "logging_callback", "score_strategy_output", diff --git a/benchmarks/results/optimization_overhead.json b/benchmarks/results/optimization_overhead.json new file mode 100644 index 0000000..7dda9ee --- /dev/null +++ b/benchmarks/results/optimization_overhead.json @@ -0,0 +1,16 @@ +{ + "intrabar_compile_to_warm_ratio": 3.6954904749510424, + "intrabar_final_equity_diff": 0.0, + "intrabar_first_run_seconds": 0.01777169480919838, + "intrabar_warm_run_seconds": 0.004809021949768066, + "loops": 24, + "normal_signal_replay_seconds": 0.1651457599364221, + "optimizer_overhead_per_trial_seconds": 0.0007232134230434895, + "optimizer_overhead_seconds": 0.017357122153043747, + "prepared_signal_replay_seconds": 0.08149230107665062, + "prepared_signal_speedup": 2.026519778611824, + "rows": 360, + "signal_final_equity_diff": 0.0, + "status": "pass", + "trials": 24 +} diff --git a/benchmarks/results/optimization_overhead.md b/benchmarks/results/optimization_overhead.md new file mode 100644 index 0000000..0bfea86 --- /dev/null +++ b/benchmarks/results/optimization_overhead.md @@ -0,0 +1,21 @@ +# Phase 32C Optimization Overhead Benchmark + +Status: **pass** + +| Measurement | Value | +|---|---:| +| Optimizer overhead | `0.017357s` | +| Optimizer overhead / trial | `0.000723s` | +| Normal signal replays | `0.165146s` | +| Prepared signal replays | `0.081492s` | +| Prepared signal speedup | `2.027x` | +| Intrabar first run | `0.017772s` | +| Intrabar warm run | `0.004809s` | +| Intrabar first/warm ratio | `3.695x` | + +Parity checks: + +- Signal final equity diff: `0.0` +- Intrabar final equity diff: `0.0` + +This benchmark measures facade/optimizer overhead, not strategy quality. diff --git a/benchmarks/run_optimization_overhead.py b/benchmarks/run_optimization_overhead.py new file mode 100644 index 0000000..6f6156e --- /dev/null +++ b/benchmarks/run_optimization_overhead.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +"""Phase 32C optimization overhead and prepared-evaluator benchmark.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys +import time + +import numpy as np +import pandas as pd + +PACKAGE_DIR = Path(__file__).resolve().parents[1] +PROJECT_DIR = PACKAGE_DIR.parent +if str(PROJECT_DIR) not in sys.path: + sys.path.insert(0, str(PROJECT_DIR)) + +from quantbt import ( # noqa: E402 + GenericEndpointEvaluator, + IntrabarIntentTape, + ObjectiveResult, + OptimizationConfig, + OptunaOptimizer, + PreparedSignalEvaluator, + QuantBTEndpoint, + SamplerConfig, +) + + +def run_benchmark(rows: int = 360, trials: int = 24, loops: int = 24) -> dict: + df = _frame(rows) + optimizer_seconds = _optimizer_overhead(trials) + normal_seconds, prepared_seconds, signal_diff = _signal_replay_benchmark(df, loops) + first_intrabar, warm_intrabar, intrabar_diff = _intrabar_compile_benchmark(df) + status = "pass" if signal_diff <= 1e-9 and intrabar_diff <= 1e-9 else "fail" + return { + "status": status, + "rows": int(rows), + "trials": int(trials), + "loops": int(loops), + "optimizer_overhead_seconds": float(optimizer_seconds), + "optimizer_overhead_per_trial_seconds": float(optimizer_seconds / max(1, trials)), + "normal_signal_replay_seconds": float(normal_seconds), + "prepared_signal_replay_seconds": float(prepared_seconds), + "prepared_signal_speedup": float(normal_seconds / prepared_seconds) if prepared_seconds > 0 else 0.0, + "signal_final_equity_diff": float(signal_diff), + "intrabar_first_run_seconds": float(first_intrabar), + "intrabar_warm_run_seconds": float(warm_intrabar), + "intrabar_compile_to_warm_ratio": float(first_intrabar / warm_intrabar) if warm_intrabar > 0 else 0.0, + "intrabar_final_equity_diff": float(intrabar_diff), + } + + +def make_markdown(report: dict) -> str: + return "\n".join( + [ + "# Phase 32C Optimization Overhead Benchmark", + "", + f"Status: **{report['status']}**", + "", + "| Measurement | Value |", + "|---|---:|", + f"| Optimizer overhead | `{report['optimizer_overhead_seconds']:.6f}s` |", + f"| Optimizer overhead / trial | `{report['optimizer_overhead_per_trial_seconds']:.6f}s` |", + f"| Normal signal replays | `{report['normal_signal_replay_seconds']:.6f}s` |", + f"| Prepared signal replays | `{report['prepared_signal_replay_seconds']:.6f}s` |", + f"| Prepared signal speedup | `{report['prepared_signal_speedup']:.3f}x` |", + f"| Intrabar first run | `{report['intrabar_first_run_seconds']:.6f}s` |", + f"| Intrabar warm run | `{report['intrabar_warm_run_seconds']:.6f}s` |", + f"| Intrabar first/warm ratio | `{report['intrabar_compile_to_warm_ratio']:.3f}x` |", + "", + "Parity checks:", + "", + f"- Signal final equity diff: `{report['signal_final_equity_diff']}`", + f"- Intrabar final equity diff: `{report['intrabar_final_equity_diff']}`", + "", + "This benchmark measures facade/optimizer overhead, not strategy quality.", + ] + ) + "\n" + + +def _optimizer_overhead(trials: int) -> float: + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: value, + objective_builder=lambda result, params: ObjectiveResult.scalar(float(result), metrics={"score": float(result)}), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig( + study_name=f"phase32c_overhead_{time.time_ns()}", + n_trials=int(trials), + seed=42, + show_progress_bar=False, + duplicate_policy="allow", + ), + sampler_config=SamplerConfig(name="random"), + ) + start = time.perf_counter() + optimizer.optimize(param_ranges={"x": (0.0, 1.0)}) + return time.perf_counter() - start + + +def _signal_replay_benchmark(df: pd.DataFrame, loops: int): + endpoint = QuantBTEndpoint.signal_notional( + backend="native_vectorized", + initial_capital=20_000.0, + leverage=5.0, + alloc_per_trade=1_000.0, + fee_rate=0.0, + use_funding=False, + ) + signal = pd.Series(np.where(df["close"].diff().fillna(0.0) > 0.0, 1.0, 0.0), index=df.index) + normal = endpoint.backtest(data=df, signal=signal, symbols=["BTC"]) + prepared = endpoint.prepare_service_context(data=df, symbols=["BTC"]) + prepared_result = prepared.backtest(signal=signal) + diff = abs(float(normal.equity.iloc[-1]) - float(prepared_result.equity.iloc[-1])) + + start = time.perf_counter() + for _ in range(int(loops)): + endpoint.backtest(data=df, signal=signal, symbols=["BTC"]) + normal_seconds = time.perf_counter() - start + + evaluator = PreparedSignalEvaluator( + prepared_context=prepared, + strategy_func=lambda params: signal, + objective_builder=lambda result, params: ObjectiveResult.scalar(float(result.equity.iloc[-1])), + ) + start = time.perf_counter() + for _ in range(int(loops)): + evaluator.evaluate({}) + prepared_seconds = time.perf_counter() - start + return normal_seconds, prepared_seconds, diff + + +def _intrabar_compile_benchmark(df: pd.DataFrame): + endpoint = QuantBTEndpoint.intrabar_bracket( + initial_capital=20_000.0, + leverage=5.0, + fee_rate=0.0, + slippage_bps=0.0, + use_funding=False, + report_level="minimal", + ) + runner = endpoint.prepare_intrabar(data=df, symbols=["BTC"]) + entry = np.zeros(len(df)) + entry[0] = 1.0 + intent = IntrabarIntentTape.from_arrays(entry_side=entry, entry_size=np.abs(entry)) + + start = time.perf_counter() + first = runner.run(intent, report_level="minimal") + first_seconds = time.perf_counter() - start + start = time.perf_counter() + warm = runner.run(intent, report_level="minimal") + warm_seconds = time.perf_counter() - start + diff = abs(float(first.equity.iloc[-1]) - float(warm.equity.iloc[-1])) + return first_seconds, warm_seconds, diff + + +def _frame(rows: int) -> pd.DataFrame: + idx = pd.date_range("2024-01-01", periods=int(rows), freq="1h", tz="UTC") + x = np.linspace(0.0, 16.0, len(idx)) + close = 100.0 + np.sin(x) * 2.0 + np.arange(len(idx)) * 0.01 + return pd.DataFrame( + { + "open": close, + "high": close * 1.01, + "low": close * 0.99, + "close": close, + "volume": 1_000.0, + }, + index=idx, + ) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--rows", type=int, default=360) + parser.add_argument("--trials", type=int, default=24) + parser.add_argument("--loops", type=int, default=24) + parser.add_argument("--json", type=Path, default=PACKAGE_DIR / "benchmarks" / "results" / "optimization_overhead.json") + parser.add_argument("--markdown", type=Path, default=PACKAGE_DIR / "benchmarks" / "results" / "optimization_overhead.md") + args = parser.parse_args() + report = run_benchmark(rows=args.rows, trials=args.trials, loops=args.loops) + args.json.parent.mkdir(parents=True, exist_ok=True) + args.json.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n") + args.markdown.write_text(make_markdown(report)) + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if report["status"] == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/core/intrabar_reference.py b/core/intrabar_reference.py index 8679b4e..06de5bb 100644 --- a/core/intrabar_reference.py +++ b/core/intrabar_reference.py @@ -109,6 +109,59 @@ def from_arrays( level_mode=level_mode, ) + @classmethod + def from_frame( + cls, + frame: pd.DataFrame, + *, + entry_side_col: str = "entry_side", + signal_col: Optional[str] = None, + entry_size_col: str = "entry_size", + stop_col: str = "stop_value", + take_profit_col: str = "take_profit_value", + trailing_col: str = "trailing_value", + technical_exit_col: str = "technical_exit", + exit_long_col: str = "exit_long", + exit_short_col: str = "exit_short", + level_mode: IntrabarLevelMode = IntrabarLevelMode.PERCENT_DISTANCE, + ) -> "IntrabarIntentTape": + """Build intrabar intents from an alpha output frame. + + This is an adapter convenience only. Strategy code still owns signal + causality; the intrabar kernel still owns fills, SL/TP/trailing, fee, + funding, margin, and liquidation semantics. + """ + + if not isinstance(frame, pd.DataFrame): + raise TypeError("frame must be a pandas DataFrame") + if entry_side_col in frame: + side = np.sign(frame[entry_side_col].fillna(0.0).to_numpy(dtype=float)).astype(np.int8) + else: + raw_col = signal_col or ("signal" if "signal" in frame else "entry") + if raw_col not in frame: + raise ValueError(f"frame must contain {entry_side_col!r}, {raw_col!r}, or provide signal_col") + raw = frame[raw_col].fillna(0.0).to_numpy(dtype=float) + side = np.sign(raw).astype(np.int8) + if entry_size_col in frame: + size = np.abs(frame[entry_size_col].fillna(0.0).to_numpy(dtype=float)) + else: + size = np.abs(side.astype(np.float64)) + + def optional(name: str): + return frame[name].to_numpy() if name in frame else None + + return cls.from_arrays( + entry_side=side, + entry_size=size, + stop_value=optional(stop_col), + take_profit_value=optional(take_profit_col), + trailing_value=optional(trailing_col), + technical_exit=optional(technical_exit_col), + exit_long=optional(exit_long_col), + exit_short=optional(exit_short_col), + level_mode=level_mode, + ) + @dataclass(frozen=True) class IntrabarFill: diff --git a/docs/README.md b/docs/README.md index dff1653..cff475e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -18,6 +18,7 @@ Use this page as the first stop when deciding which QuantBT document to read. | Understand Portfolio Engine V3 roadmap | [Portfolio Engine V3](portfolio_engine_v3.md) | | Use Nautilus as third-party execution validation, reports, and depth preflight | [Nautilus backend](nautilus_backend.md) | | Understand WFO parameter selection methodology | [Walk-forward methodology](walkforward_methodology_vi.md) | +| Tune params across signal, intrabar, portfolio, and generic endpoints | [Domain-agnostic optimization](optimization.md) | ## Strategy Route Map @@ -33,6 +34,7 @@ Use this page as the first stop when deciding which QuantBT document to read. | Arbitrage | `QuantBTEndpoint.arbitrage(...)` | Domain specs for basis, stat-arb, funding, carry, and index-basket routes | | Walk-forward optimization | `QuantBTEndpoint.walk_forward(...)` | Folded OOS stitching, anti-leakage candidate selection, and full-sample robust calibration | | Single holdout train/test | `QuantBTEndpoint.train_test_split(...)` | One train period and one test period using the WFO scoring stack | +| Standalone Optuna optimization | `OptunaOptimizer` + evaluator adapters | Prepared signal/intrabar/portfolio tuning or generic endpoint fallback | | Third-party validation | `QuantBTEndpoint.nautilus_validation(...)` or `backend="nautilus"` | Independent event-driven accounting reports | ## Example Map diff --git a/docs/endpoint.md b/docs/endpoint.md index e631a80..a5bf4be 100644 --- a/docs/endpoint.md +++ b/docs/endpoint.md @@ -2143,6 +2143,68 @@ timestamp-to-timestamp position changes, not the notional size of those changes. This keeps the penalty focused on under-trading rather than allocation magnitude. +## Domain-Agnostic Optimization Adapters + +Use standalone optimization adapters when you want Optuna tuning without WFO +fold stitching: + +```python +from quantbt import ( + OptimizationConfig, + SamplerConfig, + OptunaOptimizer, + PreparedSignalEvaluator, + SharpeObjective, +) + +endpoint = QuantBTEndpoint.signal_notional( + backend="native_vectorized", + initial_capital=20_000, + leverage=5, + alloc_per_trade=10_000, + fee_rate=0.0002, + use_funding=False, +) + +prepared = endpoint.prepare_service_context( + data=df, + symbols=["BTCUSDT"], +) + +evaluator = PreparedSignalEvaluator( + prepared_context=prepared, + strategy_func=lambda params: build_signal(df, params), + objective_builder=SharpeObjective(), +) + +study = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig( + study_name="signal_search", + n_trials=200, + show_progress_bar=False, + ), + sampler_config=SamplerConfig(name="tpe"), +) + +result = study.optimize(param_ranges=param_ranges) +``` + +Routes: + +- `PreparedSignalEvaluator`: repeated single-symbol native-vectorized signal + replays. +- `PreparedIntrabarEvaluator`: compact entry/SL/TP/trailing frames through + `QuantBTEndpoint.intrabar_bracket(...).prepare_intrabar(...)`. +- `PreparedPortfolioEvaluator`: repeated native-portfolio position matrices. +- `GenericEndpointEvaluator`: arbitrage, grid/DCA, options, or any endpoint + where `build_run_inputs(params)` can call `run_func(**inputs)`. + +The optimizer core does not shift signals and does not own look-ahead +prevention. Strategy/research code owns feature causality; QuantBT endpoints +own execution simulation, fills, PnL, fee, funding, margin, and liquidation. +See [Domain-agnostic optimization](optimization.md) for full examples. + ## Service Integration Pattern Recommended shape for alpha services: diff --git a/docs/optimization.md b/docs/optimization.md new file mode 100644 index 0000000..013c29f --- /dev/null +++ b/docs/optimization.md @@ -0,0 +1,396 @@ +# Domain-Agnostic Optimization + +QuantBT exposes a domain-agnostic Optuna layer so notebooks and services can +tune parameters without rewriting optimization boilerplate for every strategy +family. + +The key design rule is simple: + +```text +optimizer core knows params, objectives, constraints, sampler state +domain evaluator knows signal, intrabar intent, portfolio matrix, order package +``` + +This prevents a single `pos_weight`-style schema from being forced onto +strategies that have different execution meaning. + +## Public Objects + +```python +from quantbt import ( + OptimizationConfig, + SamplerConfig, + OptunaOptimizer, + ObjectiveResult, + ReportMetricObjective, + SharpeObjective, + CandidateSelector, +) +``` + +Evaluator adapters: + +```python +from quantbt import ( + GenericEndpointEvaluator, + PreparedSignalEvaluator, + PreparedIntrabarEvaluator, + PreparedPortfolioEvaluator, + ArbitrageGenericEvaluator, + GridDCAGenericEvaluator, + OptionPackageGenericEvaluator, +) +``` + +## Objective Result + +Every evaluator returns: + +```python +ObjectiveResult( + values=(sharpe,), + metrics={ + "sharpe": 1.2, + "max_drawdown_pct": 12.5, + "num_trades": 100, + }, + constraints=(), + metadata={}, +) +``` + +For multi-objective studies: + +```python +ObjectiveResult( + values=(sharpe, max_drawdown_pct, turnover), +) +``` + +with: + +```python +OptimizationConfig( + directions=("maximize", "minimize", "minimize"), +) +``` + +## Formal Constraints + +QuantBT follows Optuna convention: + +```text +constraint <= 0: feasible +constraint > 0 : violated +``` + +Example: + +```python +from quantbt import ( + ReportMetricObjective, + min_trades_constraint, + max_drawdown_constraint, +) + +objective = ReportMetricObjective( + value_metrics=("sharpe",), + constraints=( + min_trades_constraint(100), + max_drawdown_constraint(25.0), + ), +) +``` + +This is preferred over arbitrary penalties when the domain rule can be expressed +as a formal constraint. + +Metrics used by objective values or formal constraints are strict. If a metric +is missing, QuantBT raises: + +```python +MissingOptimizationMetricError +``` + +There is no silent objective fallback such as: + +```text +missing sharpe -> 0.0 +missing turnover -> num_trades +``` + +Display metrics may be omitted from `ObjectiveResult.metrics`, but objective +and constraint metrics must exist explicitly or be derivable from certified +result fields. + +Samplers without native constrained sampling support require explicit +post-filter mode: + +```python +SamplerConfig( + name="grid", + constraint_mode="post_filter", +) +``` + +This is required for `random`, `grid`, and `cmaes` studies returning formal +constraints. `tpe` and `nsgaii` can pass constraints into Optuna when supported +by the installed Optuna version. + +## Search Space + +The optimizer keeps the same parameter style used by existing alpha notebooks: + +```python +param_ranges = { + "window": (10, 80, 2), + "threshold": (0.1, 1.0, 0.05), + "use_filter": [True, False], + "mode": ["fast", "slow"], +} +``` + +Fixed parameters are passed separately: + +```python +result = optimizer.optimize( + param_ranges=param_ranges, + fixed_params={"issl": True}, +) +``` + +Fixed params are preserved in `best_params`, `selected_params`, and trial +records. + +## Generic Endpoint Evaluator + +Use this when a domain does not yet have a prepared fast evaluator. + +```python +evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: { + "data": data, + "signal": build_signal(data, params), + "symbols": ["BTCUSDT"], + }, + run_func=endpoint.backtest, + objective_builder=SharpeObjective(), +) + +optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig( + study_name="generic_signal", + n_trials=200, + show_progress_bar=False, + ), + sampler_config=SamplerConfig(name="tpe"), +) + +result = optimizer.optimize(param_ranges=param_ranges) +``` + +This fallback is intentionally used for early arbitrage, grid/DCA, and option +package workflows until a specialized prepared evaluator is worth adding. + +## Prepared Signal Evaluator + +Use this for repeated single-symbol signal-notional replays on one fixed market +tape. + +```python +endpoint = QuantBTEndpoint.signal_notional( + backend="native_vectorized", + initial_capital=20_000, + leverage=5, + alloc_per_trade=10_000, + fee_rate=0.0002, + use_funding=False, +) + +prepared = endpoint.prepare_service_context( + data=df, + symbols=["BTCUSDT"], +) + +evaluator = PreparedSignalEvaluator( + prepared_context=prepared, + strategy_func=lambda params: build_signal(df, params), + objective_builder=SharpeObjective(), +) +``` + +Prepared contexts are run-local. They are not global caches and should not be +mutated by the strategy. + +## Prepared Intrabar Evaluator + +Use this for SL/TP/trailing strategies that return compact intrabar intent +columns. + +```python +endpoint = QuantBTEndpoint.intrabar_bracket( + initial_capital=20_000, + leverage=5, + fee_rate=0.0002, + slippage_bps=2.0, + report_level="minimal", +) + +runner = endpoint.prepare_intrabar( + data=df, + symbols=["BTCUSDT"], +) + +def strategy(params): + return pd.DataFrame( + { + "entry": signal, + "stop_value": stop_distance, + "take_profit_value": take_profit_distance, + "trailing_value": trailing_distance, + }, + index=df.index, + ) + +evaluator = PreparedIntrabarEvaluator( + runner=runner, + strategy_func=strategy, + objective_builder=SharpeObjective(), + report_level="minimal", +) +``` + +`IntrabarIntentTape.from_frame(...)` converts the DataFrame into the certified +intrabar kernel input. It does not shift signals and does not manage strategy +look-ahead. + +## Prepared Portfolio Evaluator + +Use this when many position matrices are replayed against the same multi-symbol +market tape. + +```python +endpoint = QuantBTEndpoint.portfolio( + portfolio_mode="longshort", + backend="native_portfolio", + initial_capital=100_000, + leverage=5, + alloc_per_trade=1_000, + hedge_type="signal_notional", + fee=0.0004, + use_funding=False, + report_level="minimal", +) + +prepared = endpoint.prepare_service_context( + data=data_dict, + symbols=["BTC", "ETH"], +) + +evaluator = PreparedPortfolioEvaluator( + prepared_context=prepared, + strategy_func=lambda params: build_positions(data_dict, params), + objective_builder=ReportMetricObjective( + value_metrics=("sharpe", "max_drawdown_pct"), + ), +) +``` + +Core accounting parity is tested against the normal endpoint path. + +## Candidate Selection + +Optuna's best trial is not always the production parameter set. + +For single-objective constrained studies: + +```python +selector = CandidateSelector(mode="feasible_best") +result = optimizer.optimize( + param_ranges=param_ranges, + candidate_selector=selector, +) +``` + +For multi-objective studies, QuantBT returns the Pareto front unless an explicit +selector is supplied. No hidden scalarization is applied. + +When constraints exist and no candidate selector is supplied: + +```text +result.best_params -> raw Optuna best, useful for diagnostics +result.selected_params -> None +``` + +This prevents an infeasible high-score trial from being treated as production +params. Use `CandidateSelector(mode="feasible_best")` or a domain-specific +selector when production params are required. + +`CandidateSelector(mode="pareto_first")` filters infeasible Pareto trials before +selection. + +## Reproducibility Safety + +Phase 32 final merge rules are conservative: + +```text +n_jobs must be 1 +``` + +Parallel optimization is rejected until evaluator mutable state and duplicate +detection are certified thread-safe. + +For persistent Optuna storage with `load_if_exists=True`, previous QuantBT +parameter keys are preloaded so duplicate detection still works after resume. +JSONL logs write `quantbt_full_params`, including fixed params. + +## Walk-Forward Relation + +Walk-forward still owns: + +```text +fold generation +IS/OOS isolation +decay/SBB/flat-minima/is-only/full-sample robust selection +OOS stitching +``` + +Phase 32C only consolidates safe shared primitives: + +```text +search-space suggestion +duplicate parameter keys +single-objective early stopping +``` + +Anti-leakage behavior remains locked by WFO regression tests. + +## Current Scope + +Supported prepared evaluators: + +```text +single-symbol signal_notional native_vectorized +single-symbol intrabar bracket runner +native_portfolio prepared context +``` + +Generic fallback contracts: + +```text +arbitrage +grid/DCA +options +any endpoint with build_run_inputs + run_func +``` + +Not claimed yet: + +```text +specialized prepared arbitrage evaluator +specialized prepared option package evaluator +specialized prepared dynamic grid/DCA evaluator +distributed duplicate detection across independent workers +multi-objective production selector without explicit policy +``` diff --git a/examples/README.md b/examples/README.md index 0a2a4c6..5d192b7 100644 --- a/examples/README.md +++ b/examples/README.md @@ -19,6 +19,7 @@ PYTHONPATH=/root/bobby/pool_alpha python3 quantbt/examples/single_order_event.py | `pair_basket_event.py` | `BacktestEngineV2(backend="native_event", basket=...)` | Frozen hedge-ratio pair/basket package | | `arbitrage_basis.py` | `QuantBTEndpoint.arbitrage(...)` | Basis arbitrage spec and package execution | | `walk_forward_train_test.py` | `QuantBTEndpoint.train_test_split(...)` | Single holdout train/test using the walk-forward adapter | +| `optimization_workflow.py` | `OptunaOptimizer` + prepared/generic evaluators | Domain-agnostic optimization smoke template | | `nautilus_validation.py` | `QuantBTEndpoint.nautilus_validation(...)` | Signal validation through NautilusTrader | | `nautilus_explicit_orders.py` | `BacktestEngineV2(backend="nautilus", orders=...)` | Explicit order replay and native-vs-Nautilus parity | | `phase6_public_api.py` | multiple | Compact API snippets for service authors | diff --git a/examples/optimization_workflow.py b/examples/optimization_workflow.py new file mode 100644 index 0000000..2fe95b3 --- /dev/null +++ b/examples/optimization_workflow.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Small domain-agnostic optimization examples.""" + +from __future__ import annotations + +from pathlib import Path +import sys + +import numpy as np +import pandas as pd + +PACKAGE_DIR = Path(__file__).resolve().parents[1] +PROJECT_DIR = PACKAGE_DIR.parent +if str(PROJECT_DIR) not in sys.path: + sys.path.insert(0, str(PROJECT_DIR)) + +from quantbt import ( # noqa: E402 + GenericEndpointEvaluator, + ObjectiveResult, + OptimizationConfig, + OptunaOptimizer, + PreparedSignalEvaluator, + QuantBTEndpoint, + SamplerConfig, + SharpeObjective, +) + + +def main() -> None: + df = _frame() + endpoint = QuantBTEndpoint.signal_notional( + backend="native_vectorized", + initial_capital=20_000.0, + leverage=5.0, + alloc_per_trade=1_000.0, + fee_rate=0.0, + use_funding=False, + ) + + prepared = endpoint.prepare_service_context(data=df, symbols=["BTC"]) + prepared_evaluator = PreparedSignalEvaluator( + prepared_context=prepared, + strategy_func=lambda params: _signal(df, float(params["threshold"])), + objective_builder=SharpeObjective(), + ) + prepared_result = OptunaOptimizer( + evaluator=prepared_evaluator, + config=OptimizationConfig(study_name="example_prepared_signal", n_trials=6, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ).optimize(param_ranges={"threshold": (0.0, 1.0, 0.25)}) + + generic_evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: { + "data": df, + "signal": _signal(df, float(params["threshold"])), + "symbols": ["BTC"], + }, + run_func=endpoint.backtest, + objective_builder=lambda result, params: ObjectiveResult.scalar( + result.full_report()["sharpe"], + metrics={"sharpe": result.full_report()["sharpe"]}, + ), + ) + generic_result = OptunaOptimizer( + evaluator=generic_evaluator, + config=OptimizationConfig(study_name="example_generic_endpoint", n_trials=6, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ).optimize(param_ranges={"threshold": (0.0, 1.0, 0.25)}) + + print("prepared selected:", prepared_result.selected_params) + print("generic selected:", generic_result.selected_params) + + +def _frame() -> pd.DataFrame: + idx = pd.date_range("2024-01-01", periods=120, freq="1h", tz="UTC") + x = np.linspace(0.0, 10.0, len(idx)) + close = 100.0 + np.sin(x) * 3.0 + np.arange(len(idx)) * 0.02 + return pd.DataFrame( + { + "open": close, + "high": close * 1.01, + "low": close * 0.99, + "close": close, + "volume": 1_000.0, + }, + index=idx, + ) + + +def _signal(df: pd.DataFrame, threshold: float) -> pd.Series: + returns = df["close"].pct_change().fillna(0.0) + return pd.Series(np.where(returns > threshold / 100.0, 1.0, 0.0), index=df.index) + + +if __name__ == "__main__": + main() + diff --git a/optimization/__init__.py b/optimization/__init__.py new file mode 100644 index 0000000..c0dab0b --- /dev/null +++ b/optimization/__init__.py @@ -0,0 +1,89 @@ +"""Domain-agnostic optimization API for QuantBT.""" + +from .callbacks import JsonlOptimizationLogger, SingleObjectiveEarlyStopping +from .candidate_selection import CandidateSelector, SelectedCandidate, constraints_feasible +from .config import OptimizationConfig, SamplerConfig +from .constraints import CONSTRAINTS_USER_ATTR, constraints_from_trial, set_trial_constraints +from .evaluator import TrialEvaluator +from .evaluators import ( + ArbitrageGenericEvaluator, + ArbitrageTrialOutput, + GenericEndpointEvaluator, + GridDCAGenericEvaluator, + GridDCATrialOutput, + OptionPackageGenericEvaluator, + OptionTrialOutput, + PreparedIntrabarEvaluator, + PreparedPortfolioEvaluator, + PreparedSignalEvaluator, +) +from .objectives import ( + MissingOptimizationMetricError, + ReportMetricObjective, + SharpeObjective, + max_drawdown_constraint, + max_margin_utilization_constraint, + max_rejection_rate_constraint, + max_turnover_constraint, + metric_from_result, + metrics_from_result, + min_trades_constraint, + result_full_report, +) +from .optimizer import OptunaOptimizer +from .result import ObjectiveResult, OptimizationResult, OptimizationTrialRecord +from .samplers import build_sampler +from .space import ( + SearchSpaceInfo, + build_grid_search_space, + search_space_info, + stable_params_key, + suggest_parameter, + suggest_params, +) + +__all__ = [ + "CONSTRAINTS_USER_ATTR", + "ArbitrageGenericEvaluator", + "ArbitrageTrialOutput", + "CandidateSelector", + "GenericEndpointEvaluator", + "GridDCAGenericEvaluator", + "GridDCATrialOutput", + "JsonlOptimizationLogger", + "MissingOptimizationMetricError", + "ObjectiveResult", + "OptionPackageGenericEvaluator", + "OptionTrialOutput", + "OptimizationConfig", + "OptimizationResult", + "OptimizationTrialRecord", + "OptunaOptimizer", + "PreparedIntrabarEvaluator", + "PreparedPortfolioEvaluator", + "PreparedSignalEvaluator", + "ReportMetricObjective", + "SamplerConfig", + "SearchSpaceInfo", + "SelectedCandidate", + "SharpeObjective", + "SingleObjectiveEarlyStopping", + "TrialEvaluator", + "build_grid_search_space", + "build_sampler", + "constraints_feasible", + "constraints_from_trial", + "max_drawdown_constraint", + "max_margin_utilization_constraint", + "max_rejection_rate_constraint", + "max_turnover_constraint", + "metric_from_result", + "metrics_from_result", + "min_trades_constraint", + "search_space_info", + "set_trial_constraints", + "stable_params_key", + "suggest_parameter", + "suggest_params", + "result_full_report", +] diff --git a/optimization/callbacks.py b/optimization/callbacks.py new file mode 100644 index 0000000..2c6997b --- /dev/null +++ b/optimization/callbacks.py @@ -0,0 +1,111 @@ +"""Callbacks shared by QuantBT optimization workflows.""" + +from __future__ import annotations + +import json +from pathlib import Path +import time +from typing import Optional + + +class SingleObjectiveEarlyStopping: + """Stop a single-objective Optuna study after best-value stagnation.""" + + def __init__(self, patience: int, direction: str, min_delta: float = 1e-4): + if patience <= 0: + raise ValueError("patience must be positive") + direction = str(direction).lower().strip() + if direction not in {"maximize", "minimize"}: + raise ValueError("direction must be maximize or minimize") + if min_delta < 0.0: + raise ValueError("min_delta must be >= 0") + self.patience = int(patience) + self.direction = direction + self.min_delta = float(min_delta) + self._best: Optional[float] = None + self._stale = 0 + + def __call__(self, study, trial) -> None: + try: + import optuna + except Exception: # pragma: no cover - optuna import guard + optuna = None + if optuna is not None and trial.state is not optuna.trial.TrialState.COMPLETE: + return + try: + current = float(study.best_value) + except Exception: + return + if self._is_improved(current): + self._best = current + self._stale = 0 + else: + self._stale += 1 + if self._stale >= self.patience: + study.stop() + + def _is_improved(self, current: float) -> bool: + if self._best is None: + return True + if self.direction == "maximize": + return current > self._best + self.min_delta + return current < self._best - self.min_delta + + +class JsonlOptimizationLogger: + """Append parseable JSONL trial records. + + Single-objective studies log when the best trial changes. Multi-objective + studies log every completed trial because there is no scalar best value. + """ + + def __init__(self, path, *, objective_count: int): + self.path = Path(path) + self.objective_count = int(objective_count) + self._previous_best_number: Optional[int] = None + self.path.parent.mkdir(parents=True, exist_ok=True) + + def __call__(self, study, frozen_trial) -> None: + try: + import optuna + except Exception: # pragma: no cover - optuna import guard + optuna = None + if optuna is not None and frozen_trial.state is not optuna.trial.TrialState.COMPLETE: + return + if self.objective_count == 1: + try: + best_number = int(study.best_trial.number) + except Exception: + return + if best_number == self._previous_best_number: + return + self._previous_best_number = best_number + row = { + "trial": int(frozen_trial.number), + "state": str(frozen_trial.state.name), + "values": _trial_values(frozen_trial), + "params": dict(frozen_trial.user_attrs.get("quantbt_full_params", frozen_trial.params)), + "metrics": dict(frozen_trial.user_attrs.get("quantbt_metrics", {})), + "constraints": list(frozen_trial.user_attrs.get("quantbt_constraints", ())), + "metadata": dict(frozen_trial.user_attrs.get("quantbt_metadata", {})), + "duration_seconds": _duration_seconds(frozen_trial), + "logged_at_unix": time.time(), + } + with self.path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(row, sort_keys=True, default=str) + "\n") + + +def _trial_values(frozen_trial) -> list[float]: + if getattr(frozen_trial, "values", None) is not None: + return [float(value) for value in frozen_trial.values] + if getattr(frozen_trial, "value", None) is not None: + return [float(frozen_trial.value)] + return [] + + +def _duration_seconds(frozen_trial) -> Optional[float]: + start = getattr(frozen_trial, "datetime_start", None) + complete = getattr(frozen_trial, "datetime_complete", None) + if start is None or complete is None: + return None + return float((complete - start).total_seconds()) diff --git a/optimization/candidate_selection.py b/optimization/candidate_selection.py new file mode 100644 index 0000000..f7c07a8 --- /dev/null +++ b/optimization/candidate_selection.py @@ -0,0 +1,108 @@ +"""Candidate selection helpers for optimization results.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Optional + +from .result import OptimizationResult, OptimizationTrialRecord + + +def constraints_feasible(constraints: tuple[float, ...]) -> bool: + """Return True when all Optuna formal constraints are feasible.""" + + return all(float(value) <= 0.0 for value in constraints) + + +@dataclass(frozen=True) +class SelectedCandidate: + """Selected production candidate after feasibility/robustness filtering.""" + + params: dict[str, Any] + values: tuple[float, ...] = () + metrics: dict[str, float] = field(default_factory=dict) + constraints: tuple[float, ...] = () + metadata: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class CandidateSelector: + """Small public selector interface. + + This is intentionally conservative. Robust WFO plateau selectors can plug + into this interface later; Phase 32B provides best/feasible/Pareto policies + so Optuna's best trial is not silently treated as production params. + """ + + mode: str = "feasible_best" + objective_index: int = 0 + + def select(self, result: OptimizationResult) -> SelectedCandidate: + mode = str(self.mode).lower().strip() + if mode in {"best", "single_best"}: + return self._single_best(result, require_feasible=False) + if mode in {"feasible_best", "best_feasible"}: + return self._single_best(result, require_feasible=True) + if mode in {"pareto_first", "first_pareto"}: + return self._pareto_first(result) + raise ValueError(f"unsupported candidate selector mode={self.mode!r}") + + def _single_best(self, result: OptimizationResult, *, require_feasible: bool) -> SelectedCandidate: + direction = _direction(result, int(self.objective_index)) + completed = [record for record in result.trials if record.state == "COMPLETE" and len(record.values) > int(self.objective_index)] + if require_feasible: + completed = [record for record in completed if constraints_feasible(record.constraints)] + if not completed: + raise ValueError("no completed feasible optimization trials") + reverse = direction == "maximize" + best = sorted(completed, key=lambda record: record.values[int(self.objective_index)], reverse=reverse)[0] + return _selected_from_record( + best, + metadata={ + "selector": self.mode, + "objective_index": int(self.objective_index), + "feasibility_filter": bool(require_feasible), + }, + ) + + def _pareto_first(self, result: OptimizationResult) -> SelectedCandidate: + pareto = [trial for trial in result.pareto_trials if constraints_feasible(tuple(float(value) for value in trial.user_attrs.get("quantbt_constraints", ())))] + if not pareto: + raise ValueError("optimization result has no Pareto trials") + trial = pareto[0] + params = dict(trial.user_attrs.get("quantbt_full_params", trial.params)) + return SelectedCandidate( + params=params, + values=tuple(float(value) for value in (trial.values or ())), + metrics=dict(trial.user_attrs.get("quantbt_metrics", {})), + constraints=tuple(float(value) for value in trial.user_attrs.get("quantbt_constraints", ())), + metadata={ + "selector": self.mode, + "trial_number": int(trial.number), + "pareto_count": int(len(result.pareto_trials)), + "feasible_pareto_count": int(len(pareto)), + }, + ) + + +def _selected_from_record(record: OptimizationTrialRecord, *, metadata: Optional[dict[str, Any]] = None) -> SelectedCandidate: + merged_metadata = dict(record.metadata) + merged_metadata.update(metadata or {}) + merged_metadata["trial_number"] = int(record.number) + return SelectedCandidate( + params=dict(record.params), + values=tuple(record.values), + metrics=dict(record.metrics), + constraints=tuple(record.constraints), + metadata=merged_metadata, + ) + + +def _direction(result: OptimizationResult, objective_index: int) -> str: + try: + directions = tuple(str(direction.name).lower() for direction in result.study.directions) + except Exception: + directions = ("maximize",) + if objective_index < 0 or objective_index >= len(directions): + raise ValueError("objective_index out of range for optimization directions") + return directions[objective_index] diff --git a/optimization/config.py b/optimization/config.py new file mode 100644 index 0000000..6897ca8 --- /dev/null +++ b/optimization/config.py @@ -0,0 +1,81 @@ +"""Configuration objects for QuantBT domain-agnostic optimization.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional, Tuple, Union + + +Direction = str + + +@dataclass(frozen=True) +class OptimizationConfig: + """Runtime configuration for :class:`OptunaOptimizer`. + + The config intentionally avoids strategy/domain fields. Domain-specific + data, endpoints, prepared runners, and metric extraction belong in + evaluator adapters. + """ + + study_name: str + n_trials: int = 300 + directions: Tuple[Direction, ...] = ("maximize",) + seed: int = 42 + n_jobs: int = 1 + early_stopping_rounds: Optional[int] = None + early_stopping_min_delta: float = 1e-4 + show_progress_bar: bool = True + storage: Optional[str] = None + load_if_exists: bool = True + log_path: Optional[Union[str, Path]] = None + duplicate_policy: str = "prune" + exception_policy: str = "raise" + + def __post_init__(self) -> None: + if not str(self.study_name).strip(): + raise ValueError("study_name must be non-empty") + if self.n_trials <= 0: + raise ValueError("n_trials must be positive") + if not self.directions: + raise ValueError("at least one direction is required") + directions = tuple(str(direction).lower().strip() for direction in self.directions) + invalid = set(directions) - {"maximize", "minimize"} + if invalid: + raise ValueError(f"invalid directions: {invalid}") + object.__setattr__(self, "directions", directions) + if self.n_jobs <= 0: + 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_delta < 0.0: + raise ValueError("early_stopping_min_delta must be >= 0") + duplicate_policy = str(self.duplicate_policy).lower().strip() + if duplicate_policy not in {"allow", "prune", "raise"}: + raise ValueError("duplicate_policy must be allow, prune, or raise") + object.__setattr__(self, "duplicate_policy", duplicate_policy) + exception_policy = str(self.exception_policy).lower().strip() + if exception_policy not in {"raise", "fail_trial", "prune"}: + raise ValueError("exception_policy must be raise, fail_trial, or prune") + object.__setattr__(self, "exception_policy", exception_policy) + + +@dataclass(frozen=True) +class SamplerConfig: + """Optuna sampler selection and sampler-specific kwargs.""" + + name: str = "tpe" + kwargs: dict[str, Any] = field(default_factory=dict) + constraint_mode: str = "sampler" + + def __post_init__(self) -> None: + name = str(self.name).lower().strip() + if not name: + raise ValueError("sampler name must be non-empty") + object.__setattr__(self, "name", name) + object.__setattr__(self, "kwargs", dict(self.kwargs or {})) + constraint_mode = str(self.constraint_mode).lower().strip() + if constraint_mode not in {"sampler", "post_filter"}: + raise ValueError("constraint_mode must be sampler or post_filter") + object.__setattr__(self, "constraint_mode", constraint_mode) diff --git a/optimization/constraints.py b/optimization/constraints.py new file mode 100644 index 0000000..b0ee19a --- /dev/null +++ b/optimization/constraints.py @@ -0,0 +1,22 @@ +"""Formal constraint helpers for Optuna-backed optimization.""" + +from __future__ import annotations + +from typing import Sequence + + +CONSTRAINTS_USER_ATTR = "quantbt_constraints" + + +def set_trial_constraints(trial, constraints: Sequence[float]) -> tuple[float, ...]: + """Store constraints on an Optuna trial using QuantBT's canonical key.""" + + values = tuple(float(value) for value in constraints) + trial.set_user_attr(CONSTRAINTS_USER_ATTR, values) + return values + + +def constraints_from_trial(frozen_trial) -> tuple[float, ...]: + """Optuna sampler callback returning trial constraints.""" + + return tuple(float(value) for value in frozen_trial.user_attrs.get(CONSTRAINTS_USER_ATTR, ())) diff --git a/optimization/evaluator.py b/optimization/evaluator.py new file mode 100644 index 0000000..8d1b776 --- /dev/null +++ b/optimization/evaluator.py @@ -0,0 +1,21 @@ +"""Evaluator protocol for domain-specific optimization adapters.""" + +from __future__ import annotations + +from typing import Any, Mapping, Protocol + +from .result import ObjectiveResult + + +class TrialEvaluator(Protocol): + """Protocol implemented by domain adapters. + + The optimizer only sees parameters and an ObjectiveResult. Signal, + intrabar, portfolio, arbitrage, grid/DCA, and options details must remain + inside evaluator implementations. + """ + + def evaluate(self, params: Mapping[str, Any]) -> ObjectiveResult: + """Evaluate one parameter set and return objective values.""" + + ... diff --git a/optimization/evaluators/__init__.py b/optimization/evaluators/__init__.py new file mode 100644 index 0000000..d69a266 --- /dev/null +++ b/optimization/evaluators/__init__.py @@ -0,0 +1,30 @@ +"""Domain-specific optimization evaluators. + +Phase 32A intentionally keeps this namespace empty except for package +discovery. Prepared signal/intrabar/portfolio and generic endpoint evaluators +are implemented in Phase 32B. +""" + +__all__: list[str] = [] +"""Domain evaluator adapters for QuantBT optimization.""" + +from .arbitrage import ArbitrageGenericEvaluator, ArbitrageTrialOutput +from .generic import GenericEndpointEvaluator +from .grid_dca import GridDCAGenericEvaluator, GridDCATrialOutput +from .intrabar import PreparedIntrabarEvaluator +from .options import OptionPackageGenericEvaluator, OptionTrialOutput +from .portfolio import PreparedPortfolioEvaluator +from .signal import PreparedSignalEvaluator + +__all__ = [ + "ArbitrageGenericEvaluator", + "ArbitrageTrialOutput", + "GenericEndpointEvaluator", + "GridDCAGenericEvaluator", + "GridDCATrialOutput", + "OptionPackageGenericEvaluator", + "OptionTrialOutput", + "PreparedIntrabarEvaluator", + "PreparedPortfolioEvaluator", + "PreparedSignalEvaluator", +] diff --git a/optimization/evaluators/arbitrage.py b/optimization/evaluators/arbitrage.py new file mode 100644 index 0000000..4a5e99b --- /dev/null +++ b/optimization/evaluators/arbitrage.py @@ -0,0 +1,22 @@ +"""Generic arbitrage optimization adapter contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from .generic import GenericEndpointEvaluator + + +@dataclass(frozen=True) +class ArbitrageTrialOutput: + """Domain output contract for arbitrage trial builders.""" + + signal: Any + hedge_ratios: Any = None + run_overrides: dict[str, Any] = field(default_factory=dict) + + +class ArbitrageGenericEvaluator(GenericEndpointEvaluator): + """Generic fallback for arbitrage endpoints until specialized evaluators exist.""" + diff --git a/optimization/evaluators/generic.py b/optimization/evaluators/generic.py new file mode 100644 index 0000000..4bc2a5a --- /dev/null +++ b/optimization/evaluators/generic.py @@ -0,0 +1,34 @@ +"""Generic QuantBT endpoint evaluator fallback.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Mapping + +from ..result import ObjectiveResult + + +ObjectiveBuilder = Callable[[Any, Mapping[str, Any]], ObjectiveResult] + + +@dataclass +class GenericEndpointEvaluator: + """Evaluate params by building endpoint inputs and calling a run function.""" + + build_run_inputs: Callable[[Mapping[str, Any]], Mapping[str, Any]] + run_func: Callable[..., Any] + objective_builder: ObjectiveBuilder + metadata: dict[str, Any] = field(default_factory=dict) + + last_result: Any = field(default=None, init=False) + last_run_inputs: dict[str, Any] = field(default_factory=dict, init=False) + + def evaluate(self, params: Mapping[str, Any]) -> ObjectiveResult: + run_inputs = dict(self.build_run_inputs(params)) + result = self.run_func(**run_inputs) + objective = self.objective_builder(result, params) + if not isinstance(objective, ObjectiveResult): + raise TypeError("objective_builder must return ObjectiveResult") + self.last_run_inputs = run_inputs + self.last_result = result + return objective diff --git a/optimization/evaluators/grid_dca.py b/optimization/evaluators/grid_dca.py new file mode 100644 index 0000000..a725333 --- /dev/null +++ b/optimization/evaluators/grid_dca.py @@ -0,0 +1,22 @@ +"""Generic grid/DCA optimization adapter contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from .generic import GenericEndpointEvaluator + + +@dataclass(frozen=True) +class GridDCATrialOutput: + """Domain output contract for structural grid/DCA trial builders.""" + + levels: Any = None + order_plan: Any = None + run_overrides: dict[str, Any] = field(default_factory=dict) + + +class GridDCAGenericEvaluator(GenericEndpointEvaluator): + """Generic fallback for grid/DCA endpoints until prepared adapters exist.""" + diff --git a/optimization/evaluators/intrabar.py b/optimization/evaluators/intrabar.py new file mode 100644 index 0000000..1f8e484 --- /dev/null +++ b/optimization/evaluators/intrabar.py @@ -0,0 +1,54 @@ +"""Prepared intrabar evaluator.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Mapping, Optional + +import pandas as pd + +from ..result import ObjectiveResult +from .generic import ObjectiveBuilder + + +@dataclass +class PreparedIntrabarEvaluator: + """Replay intrabar strategy intents through a prepared intrabar runner.""" + + runner: Any + strategy_func: Callable[..., Any] + objective_builder: ObjectiveBuilder + intent_builder: Optional[Callable[[Any, Mapping[str, Any]], Any]] = None + report_level: str = "minimal" + pass_runner: bool = False + pass_market: bool = False + + last_result: Any = field(default=None, init=False) + last_intent: Any = field(default=None, init=False) + + def evaluate(self, params: Mapping[str, Any]) -> ObjectiveResult: + if self.pass_runner: + output = self.strategy_func(self.runner, params) + elif self.pass_market: + output = self.strategy_func(self.runner.market, params) + else: + output = self.strategy_func(params) + intent = self._to_intent(output, params) + result = self.runner.run(intent, report_level=self.report_level) + objective = self.objective_builder(result, params) + if not isinstance(objective, ObjectiveResult): + raise TypeError("objective_builder must return ObjectiveResult") + self.last_intent = intent + self.last_result = result + return objective + + def _to_intent(self, output: Any, params: Mapping[str, Any]) -> Any: + from ...core.intrabar_reference import IntrabarIntentTape + + if self.intent_builder is not None: + return self.intent_builder(output, params) + if isinstance(output, IntrabarIntentTape): + return output + if isinstance(output, pd.DataFrame): + return IntrabarIntentTape.from_frame(output) + raise TypeError("intrabar strategy must return IntrabarIntentTape or DataFrame, or provide intent_builder") diff --git a/optimization/evaluators/options.py b/optimization/evaluators/options.py new file mode 100644 index 0000000..ad84a7c --- /dev/null +++ b/optimization/evaluators/options.py @@ -0,0 +1,22 @@ +"""Generic option-package optimization adapter contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from .generic import GenericEndpointEvaluator + + +@dataclass(frozen=True) +class OptionTrialOutput: + """Domain output contract for option package trial builders.""" + + package: Any = None + hedge_plan: Any = None + run_overrides: dict[str, Any] = field(default_factory=dict) + + +class OptionPackageGenericEvaluator(GenericEndpointEvaluator): + """Generic fallback for option package endpoints until prepared adapters exist.""" + diff --git a/optimization/evaluators/portfolio.py b/optimization/evaluators/portfolio.py new file mode 100644 index 0000000..75d864b --- /dev/null +++ b/optimization/evaluators/portfolio.py @@ -0,0 +1,42 @@ +"""Prepared native portfolio evaluator.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Mapping, Optional + +from ..result import ObjectiveResult +from .generic import ObjectiveBuilder + + +@dataclass +class PreparedPortfolioEvaluator: + """Replay strategy position matrices through a prepared portfolio context.""" + + prepared_context: Any + strategy_func: Callable[..., Any] + objective_builder: ObjectiveBuilder + pass_context: bool = False + positions_key: Optional[str] = None + + last_result: Any = field(default=None, init=False) + last_positions: Any = field(default=None, init=False) + + def evaluate(self, params: Mapping[str, Any]) -> ObjectiveResult: + output = self.strategy_func(self.prepared_context, params) if self.pass_context else self.strategy_func(params) + positions = _extract_positions(output, positions_key=self.positions_key) + result = self.prepared_context.backtest(positions=positions) + objective = self.objective_builder(result, params) + if not isinstance(objective, ObjectiveResult): + raise TypeError("objective_builder must return ObjectiveResult") + self.last_positions = positions + self.last_result = result + return objective + + +def _extract_positions(output: Any, *, positions_key: Optional[str]) -> Any: + if positions_key is None: + return output + if isinstance(output, Mapping): + return output[positions_key] + return getattr(output, positions_key) diff --git a/optimization/evaluators/signal.py b/optimization/evaluators/signal.py new file mode 100644 index 0000000..7a5cd0e --- /dev/null +++ b/optimization/evaluators/signal.py @@ -0,0 +1,43 @@ +"""Prepared single-symbol signal evaluator.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Mapping, Optional + +from ..result import ObjectiveResult +from .generic import ObjectiveBuilder + + +@dataclass +class PreparedSignalEvaluator: + """Replay strategy signals through a prepared single-symbol context.""" + + prepared_context: Any + strategy_func: Callable[..., Any] + objective_builder: ObjectiveBuilder + pass_context: bool = False + signal_key: Optional[str] = None + signal_col: Optional[str] = None + + last_result: Any = field(default=None, init=False) + last_signal: Any = field(default=None, init=False) + + def evaluate(self, params: Mapping[str, Any]) -> ObjectiveResult: + output = self.strategy_func(self.prepared_context, params) if self.pass_context else self.strategy_func(params) + signal = _extract_signal(output, signal_key=self.signal_key) + result = self.prepared_context.backtest(signal=signal, signal_col=self.signal_col) + objective = self.objective_builder(result, params) + if not isinstance(objective, ObjectiveResult): + raise TypeError("objective_builder must return ObjectiveResult") + self.last_signal = signal + self.last_result = result + return objective + + +def _extract_signal(output: Any, *, signal_key: Optional[str]) -> Any: + if signal_key is None: + return output + if isinstance(output, Mapping): + return output[signal_key] + return getattr(output, signal_key) diff --git a/optimization/objectives.py b/optimization/objectives.py new file mode 100644 index 0000000..80aebac --- /dev/null +++ b/optimization/objectives.py @@ -0,0 +1,221 @@ +"""Common objective builders for domain-agnostic optimization.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Mapping, Optional, Sequence + +from .result import ObjectiveResult + + +MetricMap = Mapping[str, float] +ConstraintBuilder = Callable[[MetricMap, Mapping[str, Any], Any], float] + + +class MissingOptimizationMetricError(KeyError): + """Raised when an objective/constraint metric is required but unavailable.""" + + +_METRIC_ALIASES = { + "trades": "num_trades", + "trade_count": "num_trades", + "max_drawdown": "max_drawdown_pct", + "mdd": "max_drawdown_pct", + "margin_util": "margin_utilization", + "rejections": "rejection_rate", +} + + +def normalize_metric_name(name: str) -> str: + """Return the canonical QuantBT objective metric name.""" + + key = str(name).strip() + return _METRIC_ALIASES.get(key, key) + + +def result_full_report(result: Any, *, trading_days: int = 365, scope: str = "auto") -> dict[str, Any]: + """Extract the standard metrics report from a QuantBT result-like object.""" + + if hasattr(result, "full_report") and callable(result.full_report): + return dict(result.full_report(trading_days=trading_days, scope=scope)) + metadata = dict(getattr(result, "metadata", {}) or {}) + for key in ("report", "full_report", "metrics"): + value = metadata.get(key) + if isinstance(value, Mapping): + return dict(value) + raise TypeError("result must expose full_report(...) or metadata report/metrics") + + +def metric_from_result( + result: Any, + name: str, + *, + trading_days: int = 365, + scope: str = "auto", + required: bool = True, + default: Optional[float] = None, +) -> float: + """Read a common objective metric from report, diagnostics, or metadata.""" + + canonical = normalize_metric_name(name) + report = result_full_report(result, trading_days=trading_days, scope=scope) + if canonical in report: + return float(report[canonical]) + metadata = dict(getattr(result, "metadata", {}) or {}) + if canonical in metadata: + return float(metadata[canonical]) + if canonical == "margin_utilization": + value = _margin_utilization(result) + if value is not None: + return value + if canonical == "rejection_rate": + value = _rejection_rate(result) + if value is not None: + return value + if required: + raise MissingOptimizationMetricError(f"missing required optimization metric: {canonical}") + return float(0.0 if default is None else default) + + +def metrics_from_result( + result: Any, + *, + names: Sequence[str] = ("sharpe", "max_drawdown_pct", "num_trades", "profit_factor"), + trading_days: int = 365, + scope: str = "auto", +) -> dict[str, float]: + """Extract optional display metrics from a QuantBT result. + + Missing display metrics are omitted. Metrics used as objective values or + formal constraints must be requested through `metric_from_result(..., + required=True)` or the constraint helper functions below. + """ + + metrics: dict[str, float] = {} + report = result_full_report(result, trading_days=trading_days, scope=scope) + for name in names: + canonical = normalize_metric_name(name) + if canonical in report: + metrics[canonical] = float(report[canonical]) + else: + try: + metrics[canonical] = metric_from_result(result, canonical, trading_days=trading_days, scope=scope, required=True) + except MissingOptimizationMetricError: + pass + return metrics + + +def max_drawdown_constraint(max_drawdown_pct: float) -> ConstraintBuilder: + """Constraint: realized max drawdown must be <= `max_drawdown_pct`.""" + + limit = float(max_drawdown_pct) + return lambda metrics, params, result: _required_metric(metrics, "max_drawdown_pct") - limit + + +def min_trades_constraint(min_trades: float) -> ConstraintBuilder: + """Constraint: realized number of trades must be >= `min_trades`.""" + + required = float(min_trades) + return lambda metrics, params, result: required - _required_metric(metrics, "num_trades") + + +def max_turnover_constraint(max_turnover: float) -> ConstraintBuilder: + """Constraint: realized turnover must be <= `max_turnover`.""" + + limit = float(max_turnover) + return lambda metrics, params, result: _required_metric(metrics, "turnover") - limit + + +def max_margin_utilization_constraint(max_margin_utilization: float) -> ConstraintBuilder: + """Constraint: maximum margin utilization must be <= limit.""" + + limit = float(max_margin_utilization) + return lambda metrics, params, result: _required_metric(metrics, "margin_utilization") - limit + + +def max_rejection_rate_constraint(max_rejection_rate: float) -> ConstraintBuilder: + """Constraint: package/order rejection rate must be <= limit.""" + + limit = float(max_rejection_rate) + return lambda metrics, params, result: _required_metric(metrics, "rejection_rate") - limit + + +@dataclass(frozen=True) +class ReportMetricObjective: + """Build an ObjectiveResult from QuantBT full-report metrics. + + Formal constraints keep Optuna's convention: values `<= 0` are feasible. + The score itself is not polluted by arbitrary penalties when a constraint + can express the domain rule explicitly. + """ + + value_metrics: Sequence[str] = ("sharpe",) + metric_names: Sequence[str] = ( + "sharpe", + "max_drawdown_pct", + "num_trades", + "turnover", + "profit_factor", + "margin_utilization", + "rejection_rate", + ) + trading_days: int = 365 + scope: str = "auto" + constraints: Sequence[ConstraintBuilder] = field(default_factory=tuple) + metadata_builder: Optional[Callable[[Any, Mapping[str, Any], MetricMap], Mapping[str, Any]]] = None + + def __call__(self, result: Any, params: Mapping[str, Any]) -> ObjectiveResult: + metrics = metrics_from_result(result, names=self.metric_names, trading_days=self.trading_days, scope=self.scope) + values = tuple(metric_from_result(result, name, trading_days=self.trading_days, scope=self.scope, required=True) for name in self.value_metrics) + constraints = tuple(float(builder(metrics, params, result)) for builder in self.constraints) + metadata = {} if self.metadata_builder is None else dict(self.metadata_builder(result, params, metrics)) + return ObjectiveResult(values=values, metrics=metrics, constraints=constraints, metadata=metadata) + + +@dataclass(frozen=True) +class SharpeObjective(ReportMetricObjective): + """Single-objective Sharpe score with optional formal constraints.""" + + value_metrics: Sequence[str] = ("sharpe",) + + +def _required_metric(metrics: MetricMap, name: str) -> float: + canonical = normalize_metric_name(name) + if canonical not in metrics: + raise MissingOptimizationMetricError(f"missing required optimization metric: {canonical}") + return float(metrics[canonical]) + + +def _margin_utilization(result: Any) -> Optional[float]: + margin = getattr(result, "margin", None) + equity = getattr(result, "equity", None) + try: + if margin is not None and equity is not None and len(margin) and len(equity): + initial = margin["initial_margin"] if "initial_margin" in margin else margin.iloc[:, 0] + util = (initial.astype(float) / equity.astype(float).replace(0.0, float("nan"))).max() + return float(0.0 if util != util else util) + except Exception: + pass + return None + + +def _rejection_rate(result: Any) -> Optional[float]: + metadata = dict(getattr(result, "metadata", {}) or {}) + for key in ("rejection_rate", "package_rejection_rate"): + if key in metadata: + return float(metadata[key]) + rejected = metadata.get("rejected_count", metadata.get("rejections")) + fills = metadata.get("fill_count", metadata.get("fills_count")) + if rejected is not None and fills is not None: + denom = float(rejected) + float(fills) + return 0.0 if denom <= 0.0 else float(rejected) / denom + fills_obj = getattr(result, "fills", ()) + try: + fill_count = len(fills_obj) + if "rejected_count" not in metadata: + return None + rejected_count = int(metadata["rejected_count"]) + denom = fill_count + rejected_count + return 0.0 if denom <= 0 else float(rejected_count) / float(denom) + except Exception: + return None diff --git a/optimization/optimizer.py b/optimization/optimizer.py new file mode 100644 index 0000000..70a6ae5 --- /dev/null +++ b/optimization/optimizer.py @@ -0,0 +1,208 @@ +"""Domain-agnostic Optuna optimizer core.""" + +from __future__ import annotations + +import math +from typing import Any, Mapping, Optional + +from .callbacks import JsonlOptimizationLogger, SingleObjectiveEarlyStopping +from .candidate_selection import CandidateSelector +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 + + +class OptunaOptimizer: + """Generic Optuna orchestration over a domain-specific evaluator.""" + + def __init__( + self, + *, + evaluator: TrialEvaluator, + config: OptimizationConfig, + sampler_config: Optional[SamplerConfig] = None, + ): + self.evaluator = evaluator + self.config = config + self.sampler_config = sampler_config or SamplerConfig() + self._seen_params: set[str] = set() + + def optimize( + self, + *, + param_ranges: Mapping[str, Any], + fixed_params: Optional[Mapping[str, Any]] = None, + candidate_selector=None, + ) -> OptimizationResult: + """Run an Optuna study and return a QuantBT result schema.""" + + try: + import optuna + except Exception as exc: # pragma: no cover - dependency guard + raise ImportError("QuantBT optimization requires optuna") from exc + if int(self.config.n_jobs) != 1: + raise NotImplementedError("parallel optimization is not certified") + + objective_count = len(self.config.directions) + self._seen_params = set() + constraints_callback = ( + constraints_from_trial + if self.sampler_config.name in {"tpe", "nsgaii"} and self.sampler_config.constraint_mode == "sampler" + else None + ) + sampler = build_sampler( + self.sampler_config, + seed=int(self.config.seed), + search_space=param_ranges, + objective_count=objective_count, + constraints_func=constraints_callback, + ) + study = optuna.create_study( + study_name=self.config.study_name, + directions=list(self.config.directions), + sampler=sampler, + storage=self.config.storage, + load_if_exists=bool(self.config.load_if_exists), + pruner=optuna.pruners.NopPruner(), + ) + self._preload_seen_params(study) + callbacks = [] + if self.config.early_stopping_rounds is not None: + if objective_count != 1: + raise ValueError("early stopping is supported for single-objective optimization only") + callbacks.append( + SingleObjectiveEarlyStopping( + self.config.early_stopping_rounds, + self.config.directions[0], + min_delta=float(self.config.early_stopping_min_delta), + ) + ) + if self.config.log_path is not None: + callbacks.append(JsonlOptimizationLogger(self.config.log_path, objective_count=objective_count)) + + catch = (Exception,) if self.config.exception_policy == "fail_trial" else () + study.optimize( + lambda trial: self._objective(trial, param_ranges, fixed_params, objective_count), + n_trials=int(self.config.n_trials), + n_jobs=int(self.config.n_jobs), + callbacks=callbacks, + show_progress_bar=bool(self.config.show_progress_bar), + catch=catch, + ) + result = _build_result(study, objective_count) + if candidate_selector is not None: + selected = candidate_selector.select(result) + result.selected_params = dict(getattr(selected, "params", selected)) + result.selection_metadata = dict(getattr(selected, "metadata", {})) + elif objective_count == 1: + if _result_has_constraints(result): + result.selected_params = None + result.selection_metadata = {"selected_by": None, "reason": "constraints_require_explicit_candidate_selector"} + else: + result.selected_params = dict(result.best_params or {}) + return result + + def _preload_seen_params(self, study) -> None: + if not self.config.load_if_exists: + return + for trial in getattr(study, "trials", ()): + key = trial.user_attrs.get("quantbt_params_key") + if key is None: + params = trial.user_attrs.get("quantbt_full_params", trial.params) + if params: + key = stable_params_key(params) + if key: + self._seen_params.add(str(key)) + + def _objective(self, trial, param_ranges, fixed_params, objective_count: int): + 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) + trial.set_user_attr("quantbt_full_params", dict(params)) + trial.set_user_attr("quantbt_params_key", params_key) + if params_key in self._seen_params: + if self.config.duplicate_policy == "prune": + raise optuna.TrialPruned("duplicate parameter set") + if self.config.duplicate_policy == "raise": + raise ValueError(f"duplicate parameter set: {params_key}") + self._seen_params.add(params_key) + + try: + objective = self.evaluator.evaluate(params) + except optuna.TrialPruned: + raise + except Exception as exc: + if self.config.exception_policy == "prune": + raise optuna.TrialPruned(str(exc)) from exc + raise + if not isinstance(objective, ObjectiveResult): + raise TypeError("TrialEvaluator.evaluate must return ObjectiveResult") + if objective.constraints and self.sampler_config.name not in {"tpe", "nsgaii"} and self.sampler_config.constraint_mode != "post_filter": + raise ValueError( + f"sampler {self.sampler_config.name!r} does not support formal constraints; " + "set SamplerConfig(..., constraint_mode='post_filter') to filter candidates after optimization" + ) + if len(objective.values) != objective_count: + raise ValueError(f"objective returned {len(objective.values)} values but config has {objective_count} directions") + if not all(math.isfinite(float(value)) for value in objective.values): + 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)) + set_trial_constraints(trial, objective.constraints) + + if objective_count == 1: + return float(objective.values[0]) + return tuple(float(value) for value in objective.values) + + +def _build_result(study, objective_count: int) -> OptimizationResult: + trials = [_trial_record(trial) for trial in study.trials] + trials_frame = None + try: + trials_frame = study.trials_dataframe() + except Exception: + trials_frame = None + if objective_count == 1: + try: + best_params = dict(study.best_trial.user_attrs.get("quantbt_full_params", study.best_params)) + best_values = (float(study.best_value),) + except Exception: + best_params = None + best_values = None + pareto_trials = [] + else: + best_params = None + best_values = None + pareto_trials = list(study.best_trials) + return OptimizationResult( + study=study, + best_params=best_params, + best_values=best_values, + pareto_trials=pareto_trials, + trials=trials, + trials_frame=trials_frame, + ) + + +def _result_has_constraints(result: OptimizationResult) -> bool: + return any(len(record.constraints) > 0 for record in result.trials if record.state == "COMPLETE") + + +def _trial_record(trial) -> OptimizationTrialRecord: + values = tuple(float(value) for value in (trial.values or ())) + return OptimizationTrialRecord( + number=int(trial.number), + state=str(trial.state.name), + params=dict(trial.user_attrs.get("quantbt_full_params", trial.params)), + 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", {})), + ) diff --git a/optimization/result.py b/optimization/result.py new file mode 100644 index 0000000..4a38a53 --- /dev/null +++ b/optimization/result.py @@ -0,0 +1,73 @@ +"""Result schemas for QuantBT optimization.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Optional, Sequence, Tuple + + +@dataclass(frozen=True) +class ObjectiveResult: + """Evaluator output consumed by the domain-agnostic optimizer. + + `values` follows Optuna conventions: one value for single-objective + optimization and one value per configured direction for multi-objective + optimization. Formal constraints use Optuna's sign convention: + `<= 0` means feasible and `> 0` means violated. + """ + + values: Tuple[float, ...] + metrics: dict[str, float] = field(default_factory=dict) + constraints: Tuple[float, ...] = () + metadata: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + values = tuple(float(value) for value in self.values) + if not values: + raise ValueError("ObjectiveResult.values must be non-empty") + constraints = tuple(float(value) for value in self.constraints) + metrics = {str(key): float(value) for key, value in dict(self.metrics or {}).items()} + object.__setattr__(self, "values", values) + object.__setattr__(self, "constraints", constraints) + object.__setattr__(self, "metrics", metrics) + object.__setattr__(self, "metadata", dict(self.metadata or {})) + + @classmethod + def scalar( + cls, + value: float, + *, + metrics: Optional[dict[str, float]] = None, + constraints: Sequence[float] = (), + metadata: Optional[dict[str, Any]] = None, + ) -> "ObjectiveResult": + """Build a single-objective result.""" + + return cls(values=(float(value),), metrics=dict(metrics or {}), constraints=tuple(constraints), metadata=dict(metadata or {})) + + +@dataclass(frozen=True) +class OptimizationTrialRecord: + """Compact, serializable record of one completed/pruned/failed trial.""" + + number: int + state: str + params: dict[str, Any] + values: Tuple[float, ...] = () + metrics: dict[str, float] = field(default_factory=dict) + constraints: Tuple[float, ...] = () + metadata: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class OptimizationResult: + """Public result returned by :class:`OptunaOptimizer`.""" + + study: Any + best_params: Optional[dict[str, Any]] + best_values: Optional[Tuple[float, ...]] + pareto_trials: list[Any] + trials: list[OptimizationTrialRecord] + trials_frame: Any + selected_params: Optional[dict[str, Any]] = None + selection_metadata: dict[str, Any] = field(default_factory=dict) diff --git a/optimization/samplers.py b/optimization/samplers.py new file mode 100644 index 0000000..6f97a71 --- /dev/null +++ b/optimization/samplers.py @@ -0,0 +1,71 @@ +"""Optuna sampler factory with QuantBT compatibility checks.""" + +from __future__ import annotations + +import inspect +from typing import Any, Callable, Mapping, Optional + +from .config import SamplerConfig +from .space import build_grid_search_space, search_space_info + + +def build_sampler( + sampler_config: SamplerConfig, + *, + seed: int, + search_space: Mapping[str, Any], + objective_count: int, + constraints_func: Optional[Callable] = None, +): + """Build an Optuna sampler and validate domain-agnostic compatibility.""" + + try: + import optuna + except Exception as exc: # pragma: no cover - dependency guard + raise ImportError("QuantBT optimization requires optuna") from exc + + cfg = sampler_config if isinstance(sampler_config, SamplerConfig) else SamplerConfig(**dict(sampler_config)) + name = cfg.name + kwargs = dict(cfg.kwargs) + info = search_space_info(search_space) + + if name == "tpe": + payload = {"seed": int(seed), **kwargs} + if constraints_func is not None and _accepts(optuna.samplers.TPESampler, "constraints_func"): + payload.setdefault("constraints_func", constraints_func) + return optuna.samplers.TPESampler(**payload) + + if name == "random": + if constraints_func is not None: + raise ValueError("RandomSampler does not support formal constraints") + return optuna.samplers.RandomSampler(seed=int(seed), **kwargs) + + if name == "grid": + if constraints_func is not None: + raise ValueError("GridSampler does not support formal constraints") + max_grid_size = int(kwargs.pop("max_grid_size", 100_000)) + grid = build_grid_search_space(search_space, max_grid_size=max_grid_size) + return optuna.samplers.GridSampler(grid, seed=int(seed), **kwargs) + + if name == "cmaes": + if constraints_func is not None: + raise ValueError("CmaEsSampler does not support formal constraints") + if info.has_categorical: + raise ValueError("CMA-ES requires a numeric continuous/int search space; categorical params are not supported") + if info.has_dynamic_float is False and not info.variable_names: + raise ValueError("CMA-ES requires at least one variable numeric parameter") + return optuna.samplers.CmaEsSampler(seed=int(seed), **kwargs) + + if name == "nsgaii": + payload = {"seed": int(seed), **kwargs} + if constraints_func is not None and _accepts(optuna.samplers.NSGAIISampler, "constraints_func"): + payload.setdefault("constraints_func", constraints_func) + if objective_count < 1: + raise ValueError("objective_count must be positive") + return optuna.samplers.NSGAIISampler(**payload) + + raise ValueError("sampler name must be one of: tpe, random, grid, cmaes, nsgaii") + + +def _accepts(callable_obj, parameter: str) -> bool: + return parameter in inspect.signature(callable_obj).parameters diff --git a/optimization/space.py b/optimization/space.py new file mode 100644 index 0000000..33ba5c6 --- /dev/null +++ b/optimization/space.py @@ -0,0 +1,223 @@ +"""Search-space parsing shared by QuantBT optimization surfaces.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import math +from typing import Any, Mapping, Optional + +import numpy as np + + +@dataclass(frozen=True) +class SearchSpaceInfo: + """Static facts used by sampler compatibility checks.""" + + has_categorical: bool + has_continuous: bool + has_dynamic_float: bool + variable_names: tuple[str, ...] + grid_size: Optional[int] + + +def suggest_parameter(trial, name: str, spec: Any) -> Any: + """Suggest one parameter from a QuantBT param range spec. + + Supported specs are intentionally compatible with existing alpha notebooks: + numeric tuples, categorical lists/tuples, ranges, bool choices, and scalar + constants. + """ + + if _is_bool_choice(spec): + return trial.suggest_categorical(name, [True, False]) + if isinstance(spec, tuple) and len(spec) in (2, 3) and all(_is_number(value) for value in spec): + low, high = spec[0], spec[1] + step = spec[2] if len(spec) == 3 else None + if _looks_int(low) and _looks_int(high) and (step is None or _looks_int(step)): + return trial.suggest_int(name, int(low), int(high), step=1 if step is None else int(step)) + if step is None: + return trial.suggest_float(name, float(low), float(high)) + return trial.suggest_float(name, float(low), float(high), step=float(step)) + if isinstance(spec, range): + values = list(spec) + if not values: + raise ValueError(f"param_ranges[{name!r}] is empty") + return trial.suggest_categorical(name, values) + if isinstance(spec, (list, tuple)): + if not spec: + raise ValueError(f"param_ranges[{name!r}] is empty") + return trial.suggest_categorical(name, list(spec)) + return spec + + +def suggest_params(trial, param_ranges: Mapping[str, Any], fixed_params: Optional[Mapping[str, Any]] = None) -> dict[str, Any]: + """Suggest params and merge fixed params. + + Fixed params override `param_ranges` entries by name. Additional fixed + params are appended to the final parameter dict. + """ + + fixed = dict(fixed_params or {}) + params: dict[str, Any] = {} + for name, spec in dict(param_ranges or {}).items(): + if name in fixed: + params[name] = fixed[name] + else: + params[name] = suggest_parameter(trial, name, spec) + for name, value in fixed.items(): + params.setdefault(name, value) + return params + + +def stable_params_key(params: Mapping[str, Any]) -> str: + """Return a deterministic key for duplicate-trial detection.""" + + return json.dumps(_jsonable(params), sort_keys=True, separators=(",", ":")) + + +def search_space_info(param_ranges: Mapping[str, Any], fixed_params: Optional[Mapping[str, Any]] = None) -> SearchSpaceInfo: + """Inspect a QuantBT search space for sampler compatibility.""" + + fixed = set(dict(fixed_params or {})) + has_categorical = False + has_continuous = False + has_dynamic_float = False + variable_names: list[str] = [] + grid_size = 1 + finite_grid = True + for name, spec in dict(param_ranges or {}).items(): + if name in fixed: + continue + kind = _spec_kind(spec) + if kind == "constant": + continue + variable_names.append(name) + if kind == "categorical": + has_categorical = True + if kind in {"float", "int"}: + has_continuous = has_continuous or kind == "float" + values = _grid_values(name, spec, allow_dynamic=True) + if values is None: + finite_grid = False + has_dynamic_float = True + else: + grid_size *= len(values) + return SearchSpaceInfo( + has_categorical=has_categorical, + has_continuous=has_continuous, + has_dynamic_float=has_dynamic_float, + variable_names=tuple(variable_names), + grid_size=grid_size if finite_grid else None, + ) + + +def build_grid_search_space( + param_ranges: Mapping[str, Any], + fixed_params: Optional[Mapping[str, Any]] = None, + *, + max_grid_size: int = 100_000, +) -> dict[str, list[Any]]: + """Build an Optuna GridSampler search space from finite specs.""" + + fixed = set(dict(fixed_params or {})) + grid: dict[str, list[Any]] = {} + size = 1 + for name, spec in dict(param_ranges or {}).items(): + if name in fixed: + continue + values = _grid_values(name, spec, allow_dynamic=False) + if values is None: + raise ValueError(f"grid sampler requires finite values for {name!r}") + if len(values) == 1 and _spec_kind(spec) == "constant": + continue + grid[name] = values + size *= len(values) + if size > int(max_grid_size): + raise ValueError(f"grid search space has {size:,} combinations, above max_grid_size={int(max_grid_size):,}") + if not grid: + raise ValueError("grid sampler requires at least one non-fixed finite parameter") + return grid + + +def _grid_values(name: str, spec: Any, *, allow_dynamic: bool) -> Optional[list[Any]]: + if _is_bool_choice(spec): + return [True, False] + if isinstance(spec, tuple) and len(spec) in (2, 3) and all(_is_number(value) for value in spec): + low, high = spec[0], spec[1] + step = spec[2] if len(spec) == 3 else None + if _looks_int(low) and _looks_int(high) and (step is None or _looks_int(step)): + step_i = 1 if step is None else int(step) + if step_i <= 0: + raise ValueError(f"integer step for {name!r} must be positive") + return list(range(int(low), int(high) + 1, step_i)) + if step is None: + if allow_dynamic: + return None + raise ValueError(f"grid sampler requires a float step for {name!r}") + return _float_grid(float(low), float(high), float(step), name) + if isinstance(spec, range): + values = list(spec) + if not values: + raise ValueError(f"param_ranges[{name!r}] is empty") + return values + if isinstance(spec, (list, tuple)): + if not spec: + raise ValueError(f"param_ranges[{name!r}] is empty") + return list(spec) + return [spec] + + +def _float_grid(low: float, high: float, step: float, name: str) -> list[float]: + if step <= 0.0: + raise ValueError(f"float step for {name!r} must be positive") + if high < low: + raise ValueError(f"high must be >= low for {name!r}") + count = int(math.floor((high - low) / step + 1e-12)) + 1 + values = [float(low + i * step) for i in range(count)] + if values and values[-1] < high and math.isclose(values[-1] + step, high, rel_tol=1e-9, abs_tol=1e-12): + values.append(float(high)) + return values + + +def _spec_kind(spec: Any) -> str: + if _is_bool_choice(spec): + return "categorical" + if isinstance(spec, range): + return "categorical" + if isinstance(spec, tuple) and len(spec) in (2, 3) and all(_is_number(value) for value in spec): + if _looks_int(spec[0]) and _looks_int(spec[1]) and (len(spec) == 2 or _looks_int(spec[2])): + return "int" + return "float" + if isinstance(spec, (list, tuple)): + return "categorical" + return "constant" + + +def _is_bool_choice(spec: Any) -> bool: + return ( + isinstance(spec, (list, tuple)) + and len(spec) == 2 + and all(isinstance(value, bool) for value in spec) + and set(spec) == {True, False} + ) + + +def _looks_int(value: Any) -> bool: + return isinstance(value, (int, np.integer)) and not isinstance(value, bool) + + +def _is_number(value: Any) -> bool: + return isinstance(value, (int, float, np.integer, np.floating)) and not isinstance(value, bool) + + +def _jsonable(value: Any) -> Any: + if isinstance(value, Mapping): + return {str(key): _jsonable(val) for key, val in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(item) for item in value] + if isinstance(value, np.generic): + return value.item() + if isinstance(value, np.ndarray): + return [_jsonable(item) for item in value.tolist()] + return value diff --git a/tests/test_optimization_core.py b/tests/test_optimization_core.py new file mode 100644 index 0000000..71f7049 --- /dev/null +++ b/tests/test_optimization_core.py @@ -0,0 +1,203 @@ +import json + +import optuna +import pytest + +from quantbt.optimization import ( + ObjectiveResult, + OptimizationConfig, + OptunaOptimizer, + SamplerConfig, + SingleObjectiveEarlyStopping, + build_grid_search_space, + stable_params_key, + suggest_params, +) + + +class QuadraticEvaluator: + def __init__(self): + self.calls = [] + + def evaluate(self, params): + self.calls.append(dict(params)) + x = float(params["x"]) + score = -((x - 3.0) ** 2) + return ObjectiveResult.scalar(score, metrics={"score": score, "x": x}, metadata={"family": "mock"}) + + +class ConstantEvaluator: + def __init__(self, value=1.0): + self.value = float(value) + + def evaluate(self, params): + return ObjectiveResult.scalar(self.value, metrics={"constant": self.value}) + + +def test_single_objective_result_and_config_validation(): + result = ObjectiveResult.scalar(1.25, metrics={"sharpe": 1}, constraints=[-0.1], metadata={"a": "b"}) + + assert result.values == (1.25,) + assert result.metrics["sharpe"] == 1.0 + assert result.constraints == (-0.1,) + assert result.metadata == {"a": "b"} + + with pytest.raises(ValueError, match="n_trials"): + OptimizationConfig(study_name="bad", n_trials=0) + with pytest.raises(ValueError, match="invalid directions"): + OptimizationConfig(study_name="bad", directions=("max",)) + + +def test_fixed_params_override_and_search_space_specs(): + trial = optuna.trial.FixedTrial({"window": 20, "kind": "fast", "flag": True, "threshold": 0.3}) + params = suggest_params( + trial, + { + "window": (5, 50, 5), + "kind": ["fast", "slow"], + "flag": [True, False], + "threshold": (0.1, 1.0, 0.1), + "constant": "keep", + }, + fixed_params={"window": 34, "extra": 7}, + ) + + assert params == { + "window": 34, + "kind": "fast", + "flag": True, + "threshold": 0.3, + "constant": "keep", + "extra": 7, + } + assert stable_params_key({"b": 2, "a": 1}) == stable_params_key({"a": 1, "b": 2}) + + +def test_grid_search_space_and_size_guard(): + grid = build_grid_search_space( + { + "window": (10, 14, 2), + "kind": ["a", "b"], + "flag": [True, False], + "fixed": 1, + }, + fixed_params={"kind": "a"}, + ) + + assert grid == {"window": [10, 12, 14], "flag": [True, False]} + with pytest.raises(ValueError, match="float step"): + build_grid_search_space({"x": (0.0, 1.0)}) + with pytest.raises(ValueError, match="above max_grid_size"): + build_grid_search_space({"x": range(200), "y": range(200)}, max_grid_size=100) + + +def test_optuna_optimizer_single_objective_and_trial_records(): + evaluator = QuadraticEvaluator() + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="single_core", n_trials=12, seed=7, show_progress_bar=False), + sampler_config=SamplerConfig(name="tpe", kwargs={"n_startup_trials": 3}), + ) + + result = optimizer.optimize(param_ranges={"x": (0, 6, 1)}) + + assert result.best_params is not None + assert result.best_values is not None + assert result.selected_params == result.best_params + assert len(result.trials) == 12 + assert all(record.state in {"COMPLETE", "PRUNED", "FAIL"} for record in result.trials) + assert any(record.metrics.get("x") == result.best_params["x"] for record in result.trials if record.metrics) + + +def test_constraint_storage(): + class ConstraintEvaluator: + def evaluate(self, params): + x = float(params["x"]) + return ObjectiveResult.scalar(x, metrics={"x": x}, constraints=(x - 0.5,)) + + optimizer = OptunaOptimizer( + evaluator=ConstraintEvaluator(), + config=OptimizationConfig(study_name="constraints_core", n_trials=4, seed=4, show_progress_bar=False), + sampler_config=SamplerConfig(name="tpe", kwargs={"n_startup_trials": 1}), + ) + + result = optimizer.optimize(param_ranges={"x": (0.0, 1.0, 0.5)}) + + completed = [trial for trial in result.trials if trial.state == "COMPLETE"] + assert completed + assert all(len(trial.constraints) == 1 for trial in completed) + assert all("quantbt_constraints" in trial.user_attrs for trial in result.study.trials if trial.state.name == "COMPLETE") + + +def test_duplicate_pruning_and_nonfinite_objective_pruned(): + duplicate = OptunaOptimizer( + evaluator=ConstantEvaluator(1.0), + config=OptimizationConfig(study_name="duplicate_core", n_trials=3, seed=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + duplicate_result = duplicate.optimize(param_ranges={"x": [1]}) + + states = [record.state for record in duplicate_result.trials] + assert states.count("COMPLETE") == 1 + assert states.count("PRUNED") == 2 + + class InfiniteEvaluator: + def evaluate(self, params): + return ObjectiveResult.scalar(float("inf")) + + nonfinite = OptunaOptimizer( + evaluator=InfiniteEvaluator(), + config=OptimizationConfig(study_name="nonfinite_core", n_trials=2, seed=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + result = nonfinite.optimize(param_ranges={"x": [1, 2]}) + + assert all(record.state == "PRUNED" for record in result.trials) + assert result.best_params is None + + +def test_exception_policy_raise(): + class BrokenEvaluator: + def evaluate(self, params): + raise RuntimeError("boom") + + optimizer = OptunaOptimizer( + evaluator=BrokenEvaluator(), + config=OptimizationConfig(study_name="raise_core", n_trials=2, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + with pytest.raises(RuntimeError, match="boom"): + optimizer.optimize(param_ranges={"x": [1, 2]}) + + +def test_single_objective_early_stopping_and_jsonl_logger(tmp_path): + log_path = tmp_path / "study.jsonl" + optimizer = OptunaOptimizer( + evaluator=ConstantEvaluator(1.0), + config=OptimizationConfig( + study_name="early_stop_core", + n_trials=10, + seed=1, + early_stopping_rounds=2, + early_stopping_min_delta=0.0, + show_progress_bar=False, + log_path=log_path, + ), + sampler_config=SamplerConfig(name="random"), + ) + + result = optimizer.optimize(param_ranges={"x": (1, 10, 1)}) + + assert len(result.trials) < 10 + rows = [json.loads(line) for line in log_path.read_text().splitlines()] + assert rows + assert rows[0]["values"] == [1.0] + + +def test_pruned_trials_do_not_consume_patience(): + callback = SingleObjectiveEarlyStopping(patience=1, direction="maximize") + study = optuna.create_study(direction="maximize") + study.optimize(lambda trial: (_ for _ in ()).throw(optuna.TrialPruned()), n_trials=2, callbacks=[callback]) + + assert callback._stale == 0 diff --git a/tests/test_optimization_evaluators.py b/tests/test_optimization_evaluators.py new file mode 100644 index 0000000..db8d19f --- /dev/null +++ b/tests/test_optimization_evaluators.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from quantbt import ( + ArbitrageGenericEvaluator, + ArbitrageTrialOutput, + GenericEndpointEvaluator, + GridDCAGenericEvaluator, + GridDCATrialOutput, + IntrabarIntentTape, + ObjectiveResult, + OptionPackageGenericEvaluator, + OptionTrialOutput, + PreparedIntrabarEvaluator, + PreparedPortfolioEvaluator, + PreparedSignalEvaluator, + QuantBTEndpoint, + ReportMetricObjective, + SharpeObjective, + max_drawdown_constraint, + max_rejection_rate_constraint, + min_trades_constraint, +) + + +def _single_frame(n: int = 8) -> pd.DataFrame: + idx = pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC") + close = np.linspace(100.0, 107.0, n) + return pd.DataFrame( + { + "open": close, + "high": close + 2.0, + "low": close - 2.0, + "close": close, + "volume": np.full(n, 1000.0), + }, + index=idx, + ) + + +def _portfolio_data(): + idx = pd.date_range("2024-01-01", periods=6, freq="1D", tz="UTC") + btc = pd.DataFrame( + { + "open": [100, 100, 104, 106, 105, 107], + "high": [101, 105, 107, 108, 108, 109], + "low": [99, 99, 103, 104, 103, 106], + "close": [100, 104, 106, 105, 107, 108], + "volume": 1000.0, + }, + index=idx, + ) + eth = pd.DataFrame( + { + "open": [50, 50, 49, 51, 52, 51], + "high": [51, 51, 52, 53, 53, 52], + "low": [49, 48, 48, 50, 50, 50], + "close": [50, 49, 51, 52, 51, 50], + "volume": 1000.0, + }, + index=idx, + ) + return {"BTC": btc, "ETH": eth} + + +def test_generic_endpoint_evaluator_custom_objective_override(): + calls = [] + + class Result: + def __init__(self, value): + self.value = value + + def full_report(self, trading_days=365, scope="auto"): + return {"sharpe": self.value, "max_drawdown_pct": 1.0, "num_trades": 3} + + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"]) * 2.0}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value + 1.0, metrics={"custom": result.value}), + ) + + objective = evaluator.evaluate({"x": 4}) + calls.append(evaluator.last_run_inputs) + + assert objective.values == (9.0,) + assert objective.metrics["custom"] == 8.0 + assert calls == [{"value": 8.0}] + + +def test_prepared_signal_evaluator_matches_normal_endpoint(): + df = _single_frame() + signal = pd.Series([0, 1, 1, 0, -1, -1, 0, 0], index=df.index, dtype=float) + endpoint = QuantBTEndpoint.signal_notional( + backend="native_vectorized", + initial_capital=10_000.0, + leverage=5.0, + alloc_per_trade=1_000.0, + fee_rate=0.0, + use_funding=False, + ) + + normal = endpoint.backtest(data=df, signal=signal, symbols=["BTC"]) + prepared = endpoint.prepare_service_context(data=df, symbols=["BTC"]) + evaluator = PreparedSignalEvaluator( + prepared_context=prepared, + strategy_func=lambda params: signal * float(params["scale"]), + objective_builder=ReportMetricObjective(value_metrics=("sharpe",)), + ) + objective = evaluator.evaluate({"scale": 1.0}) + + np.testing.assert_allclose(evaluator.last_result.equity.to_numpy(), normal.equity.to_numpy(), rtol=0.0, atol=1e-9) + assert prepared.metadata["runs"] == 1 + assert "sharpe" in objective.metrics + + +def test_prepared_intrabar_evaluator_from_frame_and_minimal_audit_accounting_match(): + df = _single_frame(5) + endpoint = QuantBTEndpoint.intrabar_bracket( + initial_capital=10_000.0, + leverage=5.0, + fee_rate=0.0, + slippage_bps=0.0, + use_funding=False, + report_level="minimal", + close_on_last_bar=True, + ) + runner = endpoint.prepare_intrabar(data=df, symbols=["BTC"]) + alpha = pd.DataFrame( + { + "entry": [1.0, 0.0, 0.0, 0.0, 0.0], + "stop_value": [0.03, np.nan, np.nan, np.nan, np.nan], + "take_profit_value": [0.03, np.nan, np.nan, np.nan, np.nan], + }, + index=df.index, + ) + audit_intent = IntrabarIntentTape.from_frame(alpha) + audit = runner.run(audit_intent, report_level="audit") + + evaluator = PreparedIntrabarEvaluator( + runner=runner, + strategy_func=lambda params: alpha, + objective_builder=SharpeObjective(), + report_level="minimal", + ) + objective = evaluator.evaluate({}) + + np.testing.assert_allclose(evaluator.last_result.equity.to_numpy(), audit.equity.to_numpy(), rtol=0.0, atol=1e-9) + assert evaluator.last_result.metadata["report_level"] == "minimal" + assert audit.metadata["report_level"] == "audit" + assert objective.values == (objective.metrics["sharpe"],) + + +def test_prepared_intrabar_evaluator_requires_intent_contract(): + df = _single_frame(3) + endpoint = QuantBTEndpoint.intrabar_bracket(initial_capital=10_000.0, use_funding=False) + runner = endpoint.prepare_intrabar(data=df, symbols=["BTC"]) + evaluator = PreparedIntrabarEvaluator(runner=runner, strategy_func=lambda params: object(), objective_builder=SharpeObjective()) + + with pytest.raises(TypeError, match="IntrabarIntentTape"): + evaluator.evaluate({}) + + +def test_prepared_portfolio_evaluator_matches_normal_endpoint(): + data = _portfolio_data() + idx = next(iter(data.values())).index + positions = pd.DataFrame( + { + "BTC": [0.0, 1.0, 1.0, 0.0, -1.0, -1.0], + "ETH": [0.0, -1.0, -1.0, 0.0, 1.0, 1.0], + }, + index=idx, + ) + endpoint = QuantBTEndpoint.portfolio( + portfolio_mode="longshort", + backend="native_portfolio", + initial_capital=100_000.0, + leverage=5.0, + alloc_per_trade={"BTC": 1_000.0, "ETH": 500.0}, + hedge_type="signal_notional", + fee=0.0, + use_funding=False, + ) + + normal = endpoint.backtest(data=data, positions=positions, symbols=["BTC", "ETH"]) + prepared = endpoint.prepare_service_context(data=data, symbols=["BTC", "ETH"]) + evaluator = PreparedPortfolioEvaluator( + prepared_context=prepared, + strategy_func=lambda params: positions, + objective_builder=ReportMetricObjective(value_metrics=("sharpe", "max_drawdown_pct")), + ) + objective = evaluator.evaluate({}) + + np.testing.assert_allclose(evaluator.last_result.equity.to_numpy(), normal.equity.to_numpy(), rtol=0.0, atol=1e-9) + assert objective.values[1] == objective.metrics["max_drawdown_pct"] + + +def test_common_objective_helpers_use_formal_constraints(): + df = _single_frame() + signal = pd.Series([0, 1, 1, 0, 0, 0, 0, 0], index=df.index, dtype=float) + endpoint = QuantBTEndpoint.signal_notional( + backend="native_vectorized", + initial_capital=10_000.0, + leverage=5.0, + alloc_per_trade=1_000.0, + fee_rate=0.0, + use_funding=False, + ) + result = endpoint.backtest(data=df, signal=signal, symbols=["BTC"]) + result.metadata["rejected_count"] = 0 + result.metadata["fill_count"] = 1 + objective = ReportMetricObjective( + value_metrics=("sharpe",), + constraints=(min_trades_constraint(10), max_drawdown_constraint(99), max_rejection_rate_constraint(0.01)), + )(result, {}) + + assert objective.constraints[0] > 0.0 + assert objective.constraints[1] <= 0.0 + assert objective.constraints[2] <= 0.0 + + +def test_arbitrage_grid_dca_and_option_generic_adapters(): + class Result: + def __init__(self, value, metadata=None): + self.metadata = dict(metadata or {}) + self.value = float(value) + + def full_report(self, trading_days=365, scope="auto"): + return {"sharpe": self.value, "max_drawdown_pct": 0.0, "num_trades": 1, "profit_factor": 1.0} + + objective = SharpeObjective() + + arb = ArbitrageGenericEvaluator( + build_run_inputs=lambda params: {"output": ArbitrageTrialOutput(signal=params["x"], hedge_ratios=1.0)}, + run_func=lambda output: Result(float(output.signal), {"kind": "arb"}), + objective_builder=objective, + ) + grid = GridDCAGenericEvaluator( + build_run_inputs=lambda params: {"output": GridDCATrialOutput(levels=params["x"])}, + run_func=lambda output: Result(float(output.levels), {"kind": "grid"}), + objective_builder=objective, + ) + option = OptionPackageGenericEvaluator( + build_run_inputs=lambda params: {"output": OptionTrialOutput(package=params["x"])}, + run_func=lambda output: Result(float(output.package), {"kind": "option"}), + objective_builder=objective, + ) + + assert arb.evaluate({"x": 1.0}).values == (1.0,) + assert grid.evaluate({"x": 2.0}).values == (2.0,) + assert option.evaluate({"x": 3.0}).values == (3.0,) diff --git a/tests/test_optimization_integration.py b/tests/test_optimization_integration.py new file mode 100644 index 0000000..c394ccd --- /dev/null +++ b/tests/test_optimization_integration.py @@ -0,0 +1,370 @@ +from __future__ import annotations + +import json +import optuna +import pytest + +import quantbt.walkforward as walkforward_module +from quantbt import ( + CandidateSelector, + GenericEndpointEvaluator, + MissingOptimizationMetricError, + ObjectiveResult, + OptimizationConfig, + OptunaOptimizer, + ReportMetricObjective, + SamplerConfig, + SharpeObjective, + constraints_feasible, + max_turnover_constraint, +) +from quantbt.optimization.space import suggest_params + + +class Result: + def __init__(self, value, report=None, metadata=None): + self.value = float(value) + self._report = report + self.metadata = dict(metadata or {}) + + def full_report(self, trading_days=365, scope="auto"): + if self._report is not None: + return dict(self._report) + return {"sharpe": self.value, "max_drawdown_pct": abs(self.value), "num_trades": 1} + + +def test_optimizer_preserves_fixed_params_in_best_and_trial_records(): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": params["x"]}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value, metrics={"x": result.value}), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="fixed_params_integration", n_trials=4, seed=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + result = optimizer.optimize(param_ranges={"x": [1, 2, 3]}, fixed_params={"issl": True}, candidate_selector=CandidateSelector()) + + assert result.best_params["issl"] is True + assert result.selected_params["issl"] is True + assert all(record.params.get("issl") is True for record in result.trials if record.state == "COMPLETE") + + +def test_walkforward_sampling_reuses_optimization_core_and_preserves_float_int_ranges(): + trial = optuna.trial.FixedTrial( + { + "window": 3, + "threshold": 0.2, + "flag": True, + "mode": "fast", + } + ) + ranges = { + "window": (1.0, 5.0, 1.0), + "threshold": (0.1, 0.5, 0.1), + "flag": [True, False], + "mode": ["fast", "slow"], + "constant": 7, + } + + assert walkforward_module._sample_params(trial, ranges) == suggest_params(trial, ranges) + + +def test_constrained_optimization_and_feasible_candidate_selector(): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar( + result.value, + metrics={"score": result.value}, + constraints=(result.value - 1.0,), + ), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="constraints_integration", n_trials=4, seed=2, show_progress_bar=False), + sampler_config=SamplerConfig(name="grid", constraint_mode="post_filter"), + ) + + result = optimizer.optimize(param_ranges={"x": [0.0, 1.0, 2.0]}, candidate_selector=CandidateSelector("feasible_best")) + + assert result.selected_params["x"] == 1.0 + assert all(constraints_feasible(record.constraints) for record in result.trials if record.params.get("x") <= 1.0) + assert any(not constraints_feasible(record.constraints) for record in result.trials if record.params.get("x") > 1.0) + + +def test_missing_objective_metric_raises(): + result = Result(1.0, report={"max_drawdown_pct": 1.0, "num_trades": 10}) + + with pytest.raises(MissingOptimizationMetricError, match="sharpe"): + SharpeObjective()(result, {}) + + +def test_missing_constraint_metric_raises(): + result = Result(1.0, report={"sharpe": 1.0, "max_drawdown_pct": 1.0, "num_trades": 10}) + + with pytest.raises(MissingOptimizationMetricError, match="turnover"): + ReportMetricObjective(constraints=(max_turnover_constraint(1.0),))(result, {}) + + +def test_turnover_does_not_fallback_to_trade_count(): + result = Result(1.0, report={"sharpe": 1.0, "max_drawdown_pct": 1.0, "num_trades": 99}) + + with pytest.raises(MissingOptimizationMetricError, match="turnover"): + ReportMetricObjective(value_metrics=("turnover",))(result, {}) + + +def test_infeasible_highest_score_not_selected(): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar( + result.value, + metrics={"score": result.value}, + constraints=(result.value - 1.0,), + ), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="infeasible_best", n_trials=3, seed=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="grid", constraint_mode="post_filter"), + ) + + raw = optimizer.optimize(param_ranges={"x": [0.0, 1.0, 2.0]}) + filtered = optimizer.optimize(param_ranges={"x": [0.0, 1.0, 2.0]}, candidate_selector=CandidateSelector("feasible_best")) + + assert raw.best_params["x"] == 2.0 + assert raw.selected_params is None + assert filtered.selected_params["x"] == 1.0 + + +def test_no_feasible_trial_returns_no_selected_params(): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value, constraints=(1.0,)), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="no_feasible", n_trials=2, seed=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="grid", constraint_mode="post_filter"), + ) + + result = optimizer.optimize(param_ranges={"x": [1.0, 2.0]}) + + assert result.best_params["x"] == 2.0 + assert result.selected_params is None + + +def test_multi_objective_pareto_smoke_and_selector_policy(): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"x": float(params["x"])}, + run_func=lambda x: x, + objective_builder=lambda result, params: ObjectiveResult( + values=(float(result), abs(float(result) - 1.0)), + metrics={"score": float(result), "risk": abs(float(result) - 1.0)}, + ), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig( + study_name="pareto_integration", + n_trials=4, + directions=("maximize", "minimize"), + seed=3, + show_progress_bar=False, + duplicate_policy="allow", + ), + sampler_config=SamplerConfig(name="nsgaii"), + ) + + result = optimizer.optimize(param_ranges={"x": [0.0, 1.0, 2.0]}) + + assert result.best_params is None + assert result.selected_params is None + assert result.pareto_trials + selected = CandidateSelector("pareto_first").select(result) + assert "x" in selected.params + + +def test_pareto_selector_filters_infeasible_trials(): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"x": float(params["x"])}, + run_func=lambda x: x, + objective_builder=lambda result, params: ObjectiveResult( + values=(float(result), abs(float(result) - 1.0)), + constraints=(float(result) - 1.0,), + metrics={"score": float(result)}, + ), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig( + study_name="pareto_feasible_filter", + n_trials=3, + directions=("maximize", "minimize"), + show_progress_bar=False, + duplicate_policy="allow", + ), + sampler_config=SamplerConfig(name="grid", constraint_mode="post_filter"), + ) + + result = optimizer.optimize(param_ranges={"x": [0.0, 1.0, 2.0]}) + selected = CandidateSelector("pareto_first").select(result) + + assert selected.params["x"] <= 1.0 + assert constraints_feasible(selected.constraints) + + +def test_unsupported_constraint_sampler_requires_post_filter(): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value, constraints=(0.0,)), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig(study_name="unsupported_constraints", n_trials=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + with pytest.raises(ValueError, match="constraint_mode='post_filter'"): + optimizer.optimize(param_ranges={"x": [1.0]}) + + +def test_parallel_mode_rejected_until_thread_safe(): + optimizer = OptunaOptimizer( + evaluator=GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": params["x"]}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value), + ), + config=OptimizationConfig(study_name="parallel_reject", n_trials=1, n_jobs=2, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + with pytest.raises(NotImplementedError, match="parallel optimization is not certified"): + optimizer.optimize(param_ranges={"x": [1.0]}) + + +def test_duplicate_detection_after_sqlite_resume(tmp_path): + storage = f"sqlite:///{tmp_path / 'dup_resume.db'}" + + def make_optimizer(): + return OptunaOptimizer( + evaluator=GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value), + ), + config=OptimizationConfig( + study_name="dup_resume", + n_trials=1, + storage=storage, + load_if_exists=True, + show_progress_bar=False, + ), + sampler_config=SamplerConfig(name="random"), + ) + + first = make_optimizer().optimize(param_ranges={"x": [1.0]}) + second = make_optimizer().optimize(param_ranges={"x": [1.0]}) + + assert [record.state for record in first.trials] == ["COMPLETE"] + assert [record.state for record in second.trials][-1] == "PRUNED" + + +def test_repeated_optimize_does_not_reuse_stale_seen_set(): + optimizer = OptunaOptimizer( + evaluator=GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value), + ), + config=OptimizationConfig(study_name="stale_seen", n_trials=1, show_progress_bar=False), + sampler_config=SamplerConfig(name="random"), + ) + + first = optimizer.optimize(param_ranges={"x": [1.0]}) + second = optimizer.optimize(param_ranges={"x": [2.0]}) + + assert first.trials[-1].state == "COMPLETE" + assert second.trials[-1].state == "COMPLETE" + + +def test_jsonl_contains_fixed_and_search_params(tmp_path): + log_path = tmp_path / "study.jsonl" + optimizer = OptunaOptimizer( + evaluator=GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value), + ), + config=OptimizationConfig(study_name="jsonl_full_params", n_trials=1, show_progress_bar=False, log_path=log_path), + sampler_config=SamplerConfig(name="random"), + ) + + optimizer.optimize(param_ranges={"x": [1.0]}, fixed_params={"issl": True}) + row = json.loads(log_path.read_text().splitlines()[0]) + + assert row["params"] == {"issl": True, "x": 1.0} + + +def test_custom_objective_can_raise_and_exception_policy_prunes(): + class BrokenObjective: + def __call__(self, result, params): + raise ValueError("bad score") + + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": params["x"]}, + run_func=lambda value: Result(value), + objective_builder=BrokenObjective(), + ) + optimizer = OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig( + study_name="custom_objective_prune", + n_trials=2, + seed=1, + show_progress_bar=False, + exception_policy="prune", + ), + sampler_config=SamplerConfig(name="random"), + ) + + result = optimizer.optimize(param_ranges={"x": [1, 2]}) + + assert all(record.state == "PRUNED" for record in result.trials) + + +def test_persistent_sqlite_resume_smoke(tmp_path): + storage = f"sqlite:///{tmp_path / 'resume.db'}" + + def make_optimizer(n_trials): + evaluator = GenericEndpointEvaluator( + build_run_inputs=lambda params: {"value": float(params["x"])}, + run_func=lambda value: Result(value), + objective_builder=lambda result, params: ObjectiveResult.scalar(result.value, metrics={"score": result.value}), + ) + return OptunaOptimizer( + evaluator=evaluator, + config=OptimizationConfig( + study_name="sqlite_resume", + n_trials=n_trials, + seed=11, + show_progress_bar=False, + storage=storage, + load_if_exists=True, + duplicate_policy="allow", + ), + sampler_config=SamplerConfig(name="random"), + ) + + first = make_optimizer(2).optimize(param_ranges={"x": [0.0, 1.0, 2.0]}) + second = make_optimizer(3).optimize(param_ranges={"x": [0.0, 1.0, 2.0]}) + + assert len(first.trials) == 2 + assert len(second.trials) == 5 + assert second.best_params is not None diff --git a/tests/test_optimization_samplers.py b/tests/test_optimization_samplers.py new file mode 100644 index 0000000..25d58af --- /dev/null +++ b/tests/test_optimization_samplers.py @@ -0,0 +1,115 @@ +import optuna +import pytest + +from quantbt.optimization import ObjectiveResult, OptimizationConfig, OptunaOptimizer, SamplerConfig, build_sampler + + +class MultiObjectiveEvaluator: + def evaluate(self, params): + x = float(params["x"]) + return ObjectiveResult(values=(x, abs(x - 0.5)), metrics={"x": x}) + + +def test_tpe_factory(): + sampler = build_sampler(SamplerConfig(name="tpe"), seed=42, search_space={"x": (0.0, 1.0, 0.1)}, objective_count=1) + + assert isinstance(sampler, optuna.samplers.TPESampler) + + +def test_random_factory(): + sampler = build_sampler(SamplerConfig(name="random"), seed=42, search_space={"x": (0, 5, 1)}, objective_count=1) + + assert isinstance(sampler, optuna.samplers.RandomSampler) + + +def test_grid_factory(): + sampler = build_sampler(SamplerConfig(name="grid"), seed=42, search_space={"x": (1, 3, 1), "kind": ["a", "b"]}, objective_count=1) + + assert isinstance(sampler, optuna.samplers.GridSampler) + + +def test_cmaes_rejects_categorical(): + with pytest.raises(ValueError, match="categorical"): + build_sampler(SamplerConfig(name="cmaes"), seed=42, search_space={"x": (0.0, 1.0, 0.1), "kind": ["a", "b"]}, objective_count=1) + + +def test_nsgaii_multiobjective(): + optimizer = OptunaOptimizer( + evaluator=MultiObjectiveEvaluator(), + config=OptimizationConfig( + study_name="nsgaii_sampler", + n_trials=8, + directions=("maximize", "minimize"), + seed=42, + show_progress_bar=False, + ), + sampler_config=SamplerConfig(name="nsgaii", kwargs={"population_size": 4}), + ) + + result = optimizer.optimize(param_ranges={"x": (0.0, 1.0, 0.25)}) + + assert result.best_params is None + assert result.pareto_trials + assert all(len(trial.values) == 2 for trial in result.pareto_trials) + + +def test_constraints_func_propagation(): + def constraints_func(trial): + return (0.0,) + + tpe = build_sampler( + SamplerConfig(name="tpe"), + seed=42, + search_space={"x": (0.0, 1.0, 0.1)}, + objective_count=1, + constraints_func=constraints_func, + ) + nsgaii = build_sampler( + SamplerConfig(name="nsgaii"), + seed=42, + search_space={"x": (0.0, 1.0, 0.1)}, + objective_count=2, + constraints_func=constraints_func, + ) + + assert getattr(tpe, "_constraints_func") is constraints_func + assert getattr(nsgaii, "_constraints_func") is constraints_func + with pytest.raises(ValueError, match="does not support formal constraints"): + build_sampler(SamplerConfig(name="random"), seed=42, search_space={"x": [1, 2]}, objective_count=1, constraints_func=constraints_func) + + +def test_sampler_seed_reproducibility(): + def run_once(): + seen = [] + + class Recorder: + def evaluate(self, params): + seen.append(dict(params)) + return ObjectiveResult.scalar(float(params["x"])) + + optimizer = OptunaOptimizer( + evaluator=Recorder(), + config=OptimizationConfig(study_name="seed_repro", n_trials=5, seed=123, show_progress_bar=False, duplicate_policy="allow"), + sampler_config=SamplerConfig(name="random"), + ) + optimizer.optimize(param_ranges={"x": (0, 100, 1)}) + return seen + + assert run_once() == run_once() + + +def test_multiobjective_rejects_single_best_callback(): + optimizer = OptunaOptimizer( + evaluator=MultiObjectiveEvaluator(), + config=OptimizationConfig( + study_name="bad_multi_early", + n_trials=2, + directions=("maximize", "minimize"), + early_stopping_rounds=1, + show_progress_bar=False, + ), + sampler_config=SamplerConfig(name="nsgaii", kwargs={"population_size": 4}), + ) + + with pytest.raises(ValueError, match="single-objective"): + optimizer.optimize(param_ranges={"x": (0.0, 1.0, 0.5)}) diff --git a/upgrade/implement.md b/upgrade/implement.md index d9d16d4..c16f0fd 100644 --- a/upgrade/implement.md +++ b/upgrade/implement.md @@ -5789,3 +5789,531 @@ Chỉ merge vào `dev` khi: Kiến trúc nên chốt theo nguyên tắc: > **Một façade và một profile dùng lại cho nhiều alpha, nhưng mỗi họ chiến lược phải có output contract và backend phù hợp riêng. Không dùng một schema duy nhất để ép target-position, intrabar, portfolio, grid, DCA và arbitrage vào cùng semantics.** + +--- + +# Phase 32 - Domain-Agnostic Optimization Framework + +Status: planned, pending approval. + +Primary design guide: + +- [`upgrade/quantbt_domain_agnostic_optimization_upgrade.md`](./quantbt_domain_agnostic_optimization_upgrade.md) + +This section is only the implementation tracking layer. The detailed domain +rules, module layout, evaluator contracts, sampler compatibility, constraints, +tests, and merge gates must follow the primary design guide above. + +## Why This Phase Exists + +Current QuantBT optimization is strongest inside `walkforward.py`, but the +Optuna plumbing is too tightly coupled to walk-forward semantics: + +- search-space parsing lives inside WFO; +- sampler creation is mostly WFO-specific; +- duplicate pruning and callbacks are WFO-specific; +- robust candidate selection is useful beyond WFO but not exposed as a generic + optimizer layer; +- prepared market contexts already exist for native vectorized, native + portfolio, and intrabar, but there is no domain-agnostic evaluator contract + that lets Optuna reuse those contexts across trials. + +The upgrade should create a reusable optimization core while preserving the +important domain separation already built into QuantBT: + +```text +optimizer core knows params/objectives/constraints only +domain evaluator knows signal/intrabar/portfolio/arbitrage/grid/options output +backtest backend keeps its own execution and accounting semantics +``` + +Do not build an `IntrabarOptimizer`. Build: + +```text +optimization/ + config.py + result.py + space.py + callbacks.py + samplers.py + constraints.py + evaluator.py + evaluators/ + candidate_selection.py + optimizer.py +``` + +Public API should eventually expose: + +```python +OptimizationConfig +SamplerConfig +ObjectiveResult +OptimizationResult +TrialEvaluator +OptunaOptimizer +GenericEndpointEvaluator +PreparedSignalEvaluator +PreparedIntrabarEvaluator +PreparedPortfolioEvaluator +``` + +## Branch Plan + +Create a new branch from current `dev` after this plan is approved: + +```bash +git switch dev +git pull --ff-only origin dev +git switch -c feat/domain-agnostic-optimization +``` + +All implementation commits for this phase should stay on that feature branch +until tests and benchmarks pass. Do not merge into `dev` until the merge gates +below are satisfied. + +## Condensed Phase Plan + +The source guide lists Phase A through Phase G. To keep the work practical, we +will implement it as three larger phases without dropping any required checks. + +### Phase 32A - Optimization Core Extraction And Compatibility Lock + +Status: implemented on `feat/domain-agnostic-optimization`. + +Goal: create the generic optimization package and move shared Optuna utilities +out of WFO without changing current WFO behavior. + +Implementation scope: + +- Created `optimization/` package with: + - `OptimizationConfig`; + - `SamplerConfig`; + - `ObjectiveResult`; + - `OptimizationResult`; + - `TrialEvaluator` protocol; + - search-space helpers compatible with existing `param_ranges`; + - fixed-param override semantics; + - process-local duplicate detection; + - JSONL logger; + - single-objective early stopping callback; + - constraint user-attr helper. +- Implemented sampler factory for Phase 1 samplers: + - `tpe`; + - `random`; + - `grid`; + - `cmaes`; + - `nsgaii`. +- Validated sampler compatibility: + - CMA-ES rejects categorical/mixed spaces; + - Grid rejects dynamic/infinite spaces and warns/rejects huge Cartesian grids; + - multi-objective does not use single-objective `study.best_value`; + - constraints are passed through Optuna user attrs when supported. +- Kept `walkforward.py` behavior unchanged: + - add compatibility imports first; + - do not remove existing WFO utilities until parity tests are written; + - no scoring/objective behavior drift. + +Implemented files: + +```text +optimization/__init__.py +optimization/config.py +optimization/result.py +optimization/space.py +optimization/callbacks.py +optimization/samplers.py +optimization/constraints.py +optimization/evaluator.py +optimization/optimizer.py +optimization/evaluators/__init__.py +tests/test_optimization_core.py +tests/test_optimization_samplers.py +``` + +Important correctness note: + +- Bool choice detection requires actual `bool` values. Numeric specs such as + `(0.0, 1.0)` must not be misclassified as `[False, True]`, because Python + equality makes `0.0 == False` and `1.0 == True`. +- Unsupported formal-constraint samplers reject `constraints_func` in the + factory, while `OptunaOptimizer` only passes the constraint callback to + samplers that support it in Phase 32A (`tpe`, `nsgaii`). +- `cmaes` factory compatibility exists, but the environment currently does not + include the optional external `cmaes` package; Phase 32A tests therefore + validate construction/rejection semantics rather than running a CMA-ES study. + +Tests: + +- `test_single_objective_result`; +- `test_multi_objective_result`; +- `test_constraint_storage`; +- `test_fixed_params_override`; +- `test_search_space_specs`; +- `test_duplicate_pruning`; +- `test_nonfinite_objective_pruned`; +- `test_exception_policy_raise`; +- `test_tpe_factory`; +- `test_random_factory`; +- `test_grid_factory`; +- `test_cmaes_rejects_categorical`; +- `test_nsgaii_multiobjective`; +- `test_constraints_func_propagation`; +- `test_sampler_seed_reproducibility`; +- `test_single_objective_early_stopping`; +- `test_pruned_trials_do_not_consume_patience`; +- `test_multiobjective_rejects_single_best_callback`; +- `test_jsonl_logger`. + +Validation gate: + +```bash +pytest -q tests/test_optimization_core.py tests/test_optimization_samplers.py +pytest -q tests/test_walkforward_phase1.py +``` + +Validation after implementation: + +```text +tests/test_optimization_core.py tests/test_optimization_samplers.py: 17 passed +tests/test_walkforward_phase1.py: 51 passed +tests/test_endpoint.py: 22 passed +pytest -q: 489 passed, 1 skipped +``` + +### Phase 32B - Domain Evaluators, Constraints, And Prepared Context Parity + +Goal: make the optimizer useful across QuantBT domains without forcing every +domain into one output schema. + +Implementation scope: + +- Add `GenericEndpointEvaluator` as mandatory fallback. +- Add prepared evaluators: + - `PreparedSignalEvaluator` for single-symbol close-target/vectorized routes; + - `PreparedIntrabarEvaluator` using `QuantBTEndpoint.prepare_intrabar(...)`; + - `PreparedPortfolioEvaluator` using native portfolio prepared market arrays. +- Add initial adapter contracts for: + - arbitrage generic fallback; + - grid/DCA generic fallback; + - options generic fallback. +- Keep domain-specific imports inside evaluator adapters only. +- Add objective builder helpers for common metrics: + - Sharpe; + - max drawdown; + - trade count; + - turnover; + - margin utilization; + - rejection rate. +- Add official constraint semantics: + - feasible when value `<= 0`; + - infeasible when value `> 0`; + - do not convert constraints into arbitrary penalty scores when formal + constraints are possible. +- Add candidate selector interface: + - Optuna best trial is not automatically production params; + - feasibility filter precedes robust selection; + - single-objective returns best params; + - multi-objective returns Pareto trials unless a selector policy is passed. + +Tests: + +- `test_prepared_signal_evaluator`; +- `test_prepared_intrabar_evaluator`; +- `test_prepared_portfolio_evaluator`; +- `test_generic_endpoint_evaluator`; +- `test_arbitrage_adapter`; +- `test_grid_dca_adapter`; +- `test_option_adapter_contract`; +- `normal endpoint == prepared evaluator`; +- `minimal == audit core accounting` where the backend supports audit; +- constrained optimization smoke; +- multi-objective Pareto smoke; +- custom objective override smoke; +- persistent SQLite resume smoke. + +Validation gate: + +```bash +pytest -q tests/test_optimization_evaluators.py +pytest -q tests/test_optimization_integration.py +pytest -q tests/test_phase31*.py +pytest -q tests/test_phase11_native_portfolio_backend.py +``` + +Status: completed in Phase 32B. + +Implemented: + +- Added public evaluator adapters: + - `GenericEndpointEvaluator`; + - `PreparedSignalEvaluator`; + - `PreparedIntrabarEvaluator`; + - `PreparedPortfolioEvaluator`; + - `ArbitrageGenericEvaluator`; + - `GridDCAGenericEvaluator`; + - `OptionPackageGenericEvaluator`. +- Added initial domain output contracts: + - `ArbitrageTrialOutput`; + - `GridDCATrialOutput`; + - `OptionTrialOutput`. +- Added common objective helpers: + - `ReportMetricObjective`; + - `SharpeObjective`; + - `metric_from_result(...)`; + - `metrics_from_result(...)`; + - formal constraint helpers for minimum trades, max drawdown, turnover, + margin utilization, and rejection rate. +- Added candidate selector layer: + - `CandidateSelector`; + - `SelectedCandidate`; + - `constraints_feasible(...)`. +- Added `IntrabarIntentTape.from_frame(...)` as an adapter helper for compact + alpha DataFrames. This does not change the intrabar execution kernel. +- Fixed optimizer result bookkeeping so `fixed_params` are preserved in + `best_params`, `selected_params`, and trial records via `quantbt_full_params`. + +Tests added: + +- `tests/test_optimization_evaluators.py`; +- `tests/test_optimization_integration.py`. + +Validation: + +```bash +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q tests/test_optimization_evaluators.py tests/test_optimization_integration.py +# 12 passed + +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q tests/test_phase31*.py tests/test_phase11_native_portfolio_backend.py tests/test_optimization_core.py tests/test_optimization_samplers.py +# 82 passed + +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q +# 501 passed, 1 skipped +``` + +Scope note: + +- Arbitrage, grid/DCA, and options are intentionally available through generic + endpoint fallback contracts in Phase 32B. Specialized prepared evaluators for + these domains are future extensions and should not be claimed as done. +- Candidate selection is conservative: single-objective can select best or + feasible-best; multi-objective keeps Pareto unless an explicit selector is + supplied. + +### Phase 32C - Walk-Forward Consolidation, Docs, And Performance Benchmark + +Goal: reuse the generic optimizer in WFO without breaking anti-leakage logic or +the five existing WFO optimization modes. + +Implementation scope: + +- Replace duplicated WFO utilities with imports from `optimization/`: + - search-space suggestion; + - fixed-param merging; + - sampler factory; + - duplicate handling; + - JSONL logging where applicable; + - early stopping where applicable. +- Keep WFO-only logic in `walkforward.py`: + - fold generation; + - anti-leakage train/test isolation; + - mode 1/2/3/4/5 scoring semantics; + - temporal/plateau/full-sample robust selection metadata; + - OOS stitching. +- Add backward compatibility tests: + - old WFO sampling equals new search-space sampling; + - existing robust candidate selection metadata preserved; + - train-test split remains OOS-isolated; + - `mode_4_is_only_robust` still does not use OOS for selection; + - `mode_5_full_robust` remains explicitly full-sample, not WFO anti-leakage. +- Add docs: + - `docs/optimization.md`; + - update `docs/endpoint.md`; + - README pointer to optimization docs; + - example snippets for signal, intrabar, portfolio, and generic endpoint. +- Add benchmark: + - optimizer overhead separate from backtest runtime; + - prepared evaluator vs normal endpoint in repeated trials; + - cold vs warm Numba where applicable; + - JSON artifact under `benchmarks/results/`. + +Validation gate: + +```bash +pytest -q tests/test_walkforward_phase1.py +pytest -q tests/test_optimization*.py +pytest -q tests/test_endpoint.py +pytest -q tests/test_phase31*.py +pytest -q +python benchmarks/run_optimization_overhead.py +``` + +Status: completed in Phase 32C. + +Implemented: + +- Consolidated safe WFO utilities onto the domain-agnostic optimization layer: + - `_sample_params(...)` now delegates to `optimization.suggest_params(...)`; + - WFO duplicate keys use `optimization.stable_params_key(...)`; + - public `EarlyStoppingCallback` now reuses + `optimization.SingleObjectiveEarlyStopping`. +- Kept WFO-only anti-leakage logic in `walkforward.py`: + - fold generation; + - IS/OOS isolation; + - mode 1/2/3/4/5 objective semantics; + - robust candidate selection metadata; + - OOS stitching. +- Added documentation: + - `docs/optimization.md`; + - updated `docs/endpoint.md`; + - updated `docs/README.md`; + - updated `examples/README.md`; + - updated README performance/feature pointers. +- Added runnable example: + - `examples/optimization_workflow.py`. +- Added benchmark: + - `benchmarks/run_optimization_overhead.py`; + - `benchmarks/results/optimization_overhead.json`; + - `benchmarks/results/optimization_overhead.md`. + +Benchmark result on the committed smoke workload: + +```text +status: pass +optimizer overhead: 0.017357s for 24 trials +optimizer overhead per trial: 0.000723s +normal signal replays: 0.165146s +prepared signal replays: 0.081492s +prepared signal speedup: 2.027x +intrabar first run: 0.017772s +intrabar warm run: 0.004809s +intrabar first/warm ratio: 3.695x +signal final equity diff: 0.0 +intrabar final equity diff: 0.0 +``` + +Validation: + +```bash +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q tests/test_optimization_integration.py tests/test_optimization_evaluators.py tests/test_optimization_core.py tests/test_optimization_samplers.py +# 30 passed + +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q tests/test_walkforward_phase1.py +# 51 passed + +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q tests/test_endpoint.py tests/test_phase31*.py +# 66 passed + +MPLCONFIGDIR=/tmp PYTHONPATH=/root/bobby/pool_alpha poetry run python benchmarks/run_optimization_overhead.py --rows 360 --trials 24 --loops 24 +# status: pass + +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q +# 502 passed, 1 skipped +``` + +Scope note: + +- Phase 32C intentionally did not rewrite WFO around `OptunaOptimizer`; WFO has + anti-leakage/fold semantics that remain domain-specific and are locked by + regression tests. +- Specialized prepared evaluators for arbitrage, grid/DCA, and options remain + future work. They can be added without changing optimizer core. + +### Phase 32 Final Merge Blockers - Sol Feedback + +Status: completed after Phase 32C. + +Assessment: + +- Feedback was correct. The optimizer should fail fast when an objective or + formal-constraint metric is missing, should not use raw infeasible Optuna + best params as selected production params, and should not claim parallel + optimization safety while evaluator adapters keep mutable `last_result` / + `last_intent` state. + +Implemented: + +- Added `MissingOptimizationMetricError`. +- Objective/constraint metrics are strict: + - missing Sharpe / MaxDD / turnover / margin / rejection rate now raises when + used by objective values or formal constraints; + - `turnover` no longer falls back to `num_trades`. +- Candidate selection is constraint-safe: + - unconstrained single-objective studies still auto-populate + `selected_params`; + - constrained studies without an explicit selector now keep + `selected_params=None`; + - `CandidateSelector("feasible_best")` selects the best feasible trial; + - `CandidateSelector("pareto_first")` filters infeasible Pareto trials. +- Added `SamplerConfig.constraint_mode`: + - default: `"sampler"`; + - unsupported constrained samplers such as random/grid/CMA-ES require + `constraint_mode="post_filter"`; + - otherwise they raise instead of silently ignoring constraints. +- Reproducibility safety: + - `n_jobs != 1` raises `NotImplementedError`; + - `_seen_params` is reset at the start of every study; + - persistent studies preload previous `quantbt_params_key` / + `quantbt_full_params` so resume duplicate detection works; + - JSONL logs now write full params including fixed params. + +Tests added: + +- `test_missing_objective_metric_raises`; +- `test_missing_constraint_metric_raises`; +- `test_turnover_does_not_fallback_to_trade_count`; +- `test_infeasible_highest_score_not_selected`; +- `test_pareto_selector_filters_infeasible_trials`; +- `test_unsupported_constraint_sampler_requires_post_filter`; +- `test_no_feasible_trial_returns_no_selected_params`; +- `test_parallel_mode_rejected_until_thread_safe`; +- `test_duplicate_detection_after_sqlite_resume`; +- `test_repeated_optimize_does_not_reuse_stale_seen_set`; +- `test_jsonl_contains_fixed_and_search_params`. + +Validation: + +```bash +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q tests/test_optimization_core.py tests/test_optimization_samplers.py tests/test_optimization_evaluators.py tests/test_optimization_integration.py +# 41 passed + +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q tests/test_walkforward_phase1.py tests/test_endpoint.py tests/test_phase31*.py +# 117 passed + +PYTHONPATH=/root/bobby/pool_alpha poetry run pytest -q +# 513 passed, 1 skipped +``` + +## Merge Gates + +Do not merge unless all are true: + +- Existing walk-forward tests pass. +- Existing endpoint tests pass. +- Single-objective and multi-objective studies pass. +- Constraint semantics pass. +- TPE, Random, Grid, CMA-ES, and NSGA-II factory tests pass. +- CMA-ES rejects incompatible mixed spaces. +- Prepared signal/intrabar/portfolio parity passes. +- Generic evaluator can run arbitrage/options/grid-DCA fallback without adding + optimizer-core imports from those domains. +- No generic exception is silently converted to score `0`. +- Multi-objective code never calls `study.best_value`. +- JSONL logs are deterministic and parseable. +- SQLite resume test passes. +- Optimizer overhead benchmark is recorded. +- Documentation and examples are updated. + +## Scope Certification Target + +Target after Phase 32C: + +> Domain-agnostic Optuna orchestration with prepared evaluators for signal, +> intrabar, and portfolio; generic fallback for arbitrage, grid/DCA, and +> options; single/multi-objective studies, formal constraints, robust candidate +> selection hooks, and WFO utility consolidation without anti-leakage regression. + +Do not claim every strategy family has the same prepared performance path. +Arbitrage, grid/DCA, and options can begin through `GenericEndpointEvaluator` +and receive specialized prepared evaluators later without changing optimizer +core. diff --git a/walkforward.py b/walkforward.py index 4c8d3b2..4be680c 100644 --- a/walkforward.py +++ b/walkforward.py @@ -13,7 +13,6 @@ from dataclasses import dataclass, field import hashlib import json -import operator import time import warnings from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union @@ -22,6 +21,8 @@ import pandas as pd from .core.preprocessor import validate_datetime +from .optimization.callbacks import SingleObjectiveEarlyStopping as _OptimizationEarlyStopping +from .optimization.space import stable_params_key, suggest_params as _optimization_suggest_params try: # optional acceleration; Python/NumPy baseline remains available from numba import njit @@ -370,29 +371,12 @@ class WalkForwardTrialRecord: selection_metadata: Dict[str, Any] = field(default_factory=dict) -class EarlyStoppingCallback: +class EarlyStoppingCallback(_OptimizationEarlyStopping): """Stop Optuna if best value does not improve after N trials.""" def __init__(self, early_stopping_rounds: int, direction: str = "maximize"): + super().__init__(patience=int(early_stopping_rounds), direction=direction, min_delta=0.0) self.early_stopping_rounds = int(early_stopping_rounds) - self._iter = 0 - if direction == "minimize": - self._operator = operator.lt - self._score = np.inf - elif direction == "maximize": - self._operator = operator.gt - self._score = -np.inf - else: - raise ValueError("direction must be maximize or minimize") - - def __call__(self, study, trial) -> None: - if self._operator(study.best_value, self._score): - self._iter = 0 - self._score = study.best_value - else: - self._iter += 1 - if self._iter >= self.early_stopping_rounds: - study.stop() class DuplicatePruner(_optuna.pruners.BasePruner if _optuna is not None else object): @@ -404,7 +388,7 @@ def __init__(self): self.trial_params = set() def prune(self, study, trial) -> bool: - params_key = tuple(sorted(trial.params.items())) + params_key = stable_params_key(trial.params) if params_key in self.trial_params: return True self.trial_params.add(params_key) @@ -661,7 +645,7 @@ def optimize_params( def objective(trial): params = _sample_params(trial, param_ranges) - params_key = tuple(sorted(params.items())) + params_key = stable_params_key(params) if params_key in seen_params: record = WalkForwardTrialRecord( trial_id=int(trial.number), @@ -3009,30 +2993,7 @@ def _fold_table(folds: Sequence[WalkForwardFold]) -> pd.DataFrame: def _sample_params(trial, param_ranges: Dict[str, Any]) -> Dict[str, Any]: - params: Dict[str, Any] = {} - for name, spec in param_ranges.items(): - if isinstance(spec, tuple) and len(spec) in (2, 3) and all(_is_number(x) for x in spec): - low = spec[0] - high = spec[1] - step = spec[2] if len(spec) == 3 else None - if _looks_int(low) and _looks_int(high) and (step is None or _looks_int(step)): - params[name] = trial.suggest_int(name, int(low), int(high), step=1 if step is None else int(step)) - else: - if step is None: - params[name] = trial.suggest_float(name, float(low), float(high)) - else: - params[name] = trial.suggest_float(name, float(low), float(high), step=float(step)) - elif isinstance(spec, list): - if not spec: - raise ValueError(f"param_ranges[{name!r}] is empty") - params[name] = trial.suggest_categorical(name, spec) - elif isinstance(spec, tuple): - if not spec: - raise ValueError(f"param_ranges[{name!r}] is empty") - params[name] = trial.suggest_categorical(name, list(spec)) - else: - params[name] = spec - return params + return _optimization_suggest_params(trial, param_ranges) def _looks_int(value: Any) -> bool: