cyb50-quant/cyb50-pro/utils/bayesian_optimizer.py

"""
贝叶斯优化器

用于智能体参数的在线优化
"""

from typing import Callable, Dict, List, Tuple, Optional
import numpy as np

from skopt import gp_minimize
from skopt.space import Real, Integer, Categorical
from skopt.utils import use_named_args


class BayesianOptimizer:
    """
    贝叶斯优化器

    使用高斯过程代理模型进行样本高效的参数优化
    """

    def __init__(
        self,
        dimensions: List,
        n_initial_points: int = 5,
        acquisition_function: str = "EI",  # Expected Improvement
        noise: float = 1e-5
    ):
        self.dimensions = dimensions
        self.n_initial_points = n_initial_points
        self.acquisition_function = acquisition_function
        self.noise = noise

        self.X: List[List] = []  # 参数历史
        self.y: List[float] = []  # 结果历史

    def optimize(
        self,
        objective: Callable,
        n_calls: int = 10,
        x0: Optional[List] = None,
        y0: Optional[List] = None
    ) -> Dict:
        """
        执行优化

        Args:
            objective: 目标函数，返回要最小化的值（负收益）
            n_calls: 总采样次数
            x0: 初始参数点
            y0: 初始结果

        Returns:
            优化结果字典
        """
        result = gp_minimize(
            func=objective,
            dimensions=self.dimensions,
            n_calls=n_calls,
            n_initial_points=self.n_initial_points,
            acq_func=self.acquisition_function,
            noise=self.noise,
            x0=x0,
            y0=y0,
            random_state=42
        )

        return {
            "best_params": result.x,
            "best_value": result.fun,
            "all_params": result.x_iters,
            "all_values": result.func_vals
        }

    def suggest_next_point(self) -> List:
        """建议下一个采样点（用于在线增量优化）"""
        if len(self.X) < self.n_initial_points:
            # 随机采样初始点
            return [dim.rvs()[0] for dim in self.dimensions]

        # 使用采集函数选择下一点（简化实现）
        # 实际应使用训练好的GP模型
        return [dim.rvs()[0] for dim in self.dimensions]

    def update(self, params: List, value: float):
        """更新观测历史"""
        self.X.append(params)
        self.y.append(value)