第 10 章 · 02 自定义优化器 BaseOptimizer

第 10 章 · 02 自定义优化器 BaseOptimizer

本节摘要:PyPortfolioOpt 内置的优化器(EfficientFrontier、HRPOpt、CLA、BlackLittermanModel)都继承自 BaseOptimizer 或 BaseConvexOptimizer 这两个抽象基类。它们提供统一的「后处理」接口(clean_weights、save_weights_to_file、portfolio_performance),让你的自定义优化器无缝接入四层架构(收益/风险/约束/目标)。本节讲清两个基类的职责划分、关键方法、以及如何继承 BaseConvexOptimizer 用 cvxpy 写一个自定义优化器(以跟踪误差最小化为例)。读完本节,你能扩展 PyPortfolioOpt 实现任意自定义目标。

内容来源:源码 pypfopt/base/_base_optimizer.py、文档 docs/OtherOptimizers.rst、docs/GeneralEfficientFrontier.rst,汉化并套用体系化模板。

学习目标

阅读完本节,你应当能够:

1. 区分 BaseOptimizer 与 BaseConvexOptimizer 的职责。
2. 用 clean_weights / save_weights_to_file 做后处理。
3. 用 convex_objective 写自定义凸目标。
4. 用 nonconvex_objective 走 scipy 后端处理非凸问题。
5. 继承 BaseOptimizer 写一个完整自定义优化器。

一、两层抽象:BaseOptimizer vs BaseConvexOptimizer

源码定义两个基类,职责严格分层:

BaseOptimizer 极简——只管 n_assets、tickers、weights,以及三个工具方法。BaseConvexOptimizer 在此之上加了 cvxpy 的 _w 变量、约束列表、目标函数。

💡 选哪个继承:你的优化器若能用 cvxpy 表达,继承 BaseConvexOptimizer;若是完全自定义算法(遗传、模拟退火、聚类等),继承 BaseOptimizer。

二、BaseOptimizer 的核心 API

源码 __init__:

class BaseOptimizer:
    def __init__(self, n_assets, tickers=None):
        self.n_assets = n_assets
        if tickers is None:
            self.tickers = list(range(n_assets))
        else:
            self.tickers = tickers
        self._risk_free_rate = None
        self.weights = None     # 优化结果,由子类填
​

三个核心方法:

set_weights:从 dict 填 self.weights

def set_weights(self, input_weights):
    self.weights = np.array([input_weights[ticker] for ticker in self.tickers])
​

clean_weights:清理 + 舍入

def clean_weights(self, cutoff=1e-4, rounding=5):
    if self.weights is None:
        raise AttributeError("Weights not yet computed")
    clean_weights = self.weights.copy()
    clean_weights[np.abs(clean_weights) < cutoff] = 0   # 小于 cutoff 置 0
    if rounding is not None:
        clean_weights = np.round(clean_weights, rounding)
    return self._make_output_weights(clean_weights)
​

• cutoff=1e-4:绝对值小于 0.0001 的权重置 0(避免 1e-17 这种数值垃圾)。
• rounding=5:保留 5 位小数。

save_weights_to_file:持久化

def save_weights_to_file(self, filename="weights.csv"):
    clean_weights = self.clean_weights()
    ext = filename.split(".")[-1].lower()
    if ext == "csv":
        pd.Series(clean_weights).to_csv(filename, header=False)
    elif ext == "json":
        with open(filename, "w") as fp:
            json.dump(clean_weights, fp)
    elif ext == "txt":
        with open(filename, "w") as f:
            f.write(str(dict(clean_weights)))
    else:
        raise NotImplementedError("Only supports .txt .json .csv")
​

支持 csv / json / txt 三种格式。

💡 API 统一的价值:因为所有内置优化器都继承自这两个基类,无论你用 EfficientFrontier 还是 HRPOpt,clean_weights / save_weights_to_file / portfolio_performance 的签名完全一致。这让上层代码可以无缝替换优化器。

三、BaseConvexOptimizer 的 cvxpy 后端

class BaseConvexOptimizer(BaseOptimizer):
    def __init__(self, n_assets, tickers=None, weight_bounds=(0, 1),
                 solver=None, verbose=False, solver_options=None):
        super().__init__(n_assets, tickers)
        self._w = cp.Variable(n_assets)                # 权重变量
        self._objective = None
        self._additional_objectives = []
        self._constraints = []
        # ...
        self._map_bounds_to_constraints(weight_bounds)
​

关键私有变量:

• self._w:cvxpy 的权重变量,所有目标/约束都引用它。
• self._objective:主目标。
• self._additional_objectives:额外目标(L2 正则等)。
• self._constraints:约束列表。

add_constraint 与 add_objective

ef.add_constraint(lambda w: w[0] >= 0.2)              # 单资产下界
ef.add_constraint(lambda w: w[3] + w[4] <= 0.10)      # 组合上界
ef.add_objective(objective_functions.L2_reg)          # L2 正则
​

源码:

def add_objective(self, new_objective, **kwargs):
    if self._opt is not None:
        raise InstantiationError(
            "Adding objectives to an already solved problem might have unintended consequences. "
            "A new instance should be created for the new set of objectives."
        )
    self._additional_objectives.append(new_objective(self._w, **kwargs))

def add_constraint(self, new_constraint):
    if not callable(new_constraint):
        raise TypeError("New constraint must be provided as a callable (e.g lambda function)")
    if self._opt is not None:
        raise InstantiationError(...)
    self._constraints.append(new_constraint(self._w))
​

⚠️ 优化后不可再 add:一旦调用过 max_sharpe / min_volatility 等,self._opt 就非 None,继续 add 会抛 InstantiationError。要加新约束/目标,必须新建实例。

add_sector_constraints:行业约束

sector_mapper = {"GOOG": "tech", "FB": "tech", "XOM": "Oil/Gas", ...}
sector_lower = {"tech": 0.1}              # tech 至少 10%
sector_upper = {"tech": 0.4, "Oil/Gas": 0.1}

ef.add_sector_constraints(sector_mapper, sector_lower, sector_upper)
​

四、convex_objective:自定义凸目标

如果你的目标能用 cvxpy 原子函数表达,用 convex_objective:

def logarithmic_barrier(w, cov_matrix, k=0.1):
    # Kolm et al (2014)
    return cp.quad_form(w, cov_matrix) - k * cp.sum(cp.log(w))

w = ef.convex_objective(logarithmic_barrier, cov_matrix=ef.cov_matrix)
​

源码:

def convex_objective(self, custom_objective, weights_sum_to_one=True, **kwargs):
    self._objective = custom_objective(self._w, **kwargs)
    for obj in self._additional_objectives:
        self._objective += obj
    if weights_sum_to_one:
        self.add_constraint(lambda w: cp.sum(w) == 1)
    return self._solve_cvxpy_opt_problem()
​

文档里的跟踪误差最小化示例——直接用 BaseConvexOptimizer 而非 EfficientFrontier:

from pypfopt.base import BaseConvexOptimizer
from pypfopt.objective_functions import ex_post_tracking_error

historic_rets = ...      # DataFrame
benchmark_rets = ...     # Series

opt = BaseConvexOptimizer(
    n_assets=len(historic_returns.columns),
    tickers=historic_returns.columns,
    weight_bounds=(0, 1),
)
opt.convex_objective(
    ex_post_tracking_error,
    historic_returns=historic_rets,
    benchmark_returns=benchmark_rets,
)
weights = opt.clean_weights()
​

💡 何时直接用 BaseConvexOptimizer:目标与「收益/方差」无关(如跟踪误差、最大分散化、风险平价)时,直接用基类比继承 EfficientFrontier 更干净——避免父类无关方法干扰。

五、nonconvex_objective:scipy 后端

如果目标非凸(不能写成 cvxpy 原子),用 nonconvex_objective,走 scipy.optimize:

def nonconvex_objective(
    self, custom_objective, objective_args=None,
    weights_sum_to_one=True, constraints=None,
    solver="SLSQP", initial_guess=None,
):
    # ... 构造 scipy 约束
    result = sco.minimize(
        custom_objective, x0=initial_guess, args=objective_args,
        method=solver, bounds=bounds, constraints=final_constraints,
    )
    self.weights = result["x"]
    return self._make_output_weights()
​

示例——市场中性下的有效风险:

constraints = [
    {"type": "eq", "fun": lambda w: np.sum(w)},   # 权重和为 0(市场中性)
    {"type": "eq", "fun": lambda w: target_risk**2 - w.T @ ef.cov_matrix @ w},
]
ef.nonconvex_objective(
    lambda w, mu: -w.dot(mu),                       # 最小化负收益 = 最大化收益
    objective_args=(ef.expected_returns,),
    weights_sum_to_one=False,
    constraints=constraints,
)
​

⚠️ scipy 后端的局限:可能陷入局部最优、约束需手写 scipy 字典格式、速度比 cvxpy 慢。官方明确不推荐除非真非凸。

六、继承写一个完整的自定义优化器

完全自定义算法时,继承 BaseOptimizer:

import numpy as np
import pandas as pd
from pypfopt.base import BaseOptimizer

class MyEqualWeightOptimizer(BaseOptimizer):
    """最简单的示例:等权分配"""

    def __init__(self, tickers):
        super().__init__(n_assets=len(tickers), tickers=tickers)

    def optimize(self):
        w = np.ones(self.n_assets) / self.n_assets
        self.weights = w
        return self._make_output_weights()
​

由于继承自 BaseOptimizer,自动获得 clean_weights、save_weights_to_file、set_weights,可被 plotting.plot_weights 直接消费,与其他模块无缝拼接。

opt = MyEqualWeightOptimizer(["AAPL", "MSFT", "GOOG"])
weights = opt.optimize()
print(opt.clean_weights())
opt.save_weights_to_file("eq_weights.json")

from pypfopt import plotting
plotting.plot_weights(weights)
​

本节要点回顾

1. 两层基类:BaseOptimizer(纯后处理,无 cvxpy);BaseConvexOptimizer(+ cvxpy 变量/约束/目标,父类被 EfficientFrontier 系列继承)。
2. 统一 API:set_weights、clean_weights(cutoff, rounding)、save_weights_to_file 在所有内置优化器签名一致,可无缝替换。
3. add_constraint / add_objective 必须在优化前调,优化后再调会抛 InstantiationError,要改需新建实例。
4. convex_objective:目标用 cvxpy 原子表达,如跟踪误差、对数壁垒;目标与收益/方差无关时直接用 BaseConvexOptimizer 更干净。
5. nonconvex_objective:scipy 后端,处理非凸目标但可能陷局部最优,官方不推荐。
6. 继承 BaseOptimizer 写自定义优化器:实现 optimize 填 self.weights,即可获得全部后处理与生态对接能力。

下一节,我们看 PyPortfolioOpt 的测试规范与贡献指南——把这个库当成可参与的开放项目。