跳转至

优化器 API

MeanVarianceOptimizer

Bases: Optimizer


              flowchart TD
              fund_cli.core.optimizers.mean_variance.MeanVarianceOptimizer[MeanVarianceOptimizer]
              fund_cli.core.optimizer.Optimizer[Optimizer]

                              fund_cli.core.optimizer.Optimizer --> fund_cli.core.optimizers.mean_variance.MeanVarianceOptimizer
                


              click fund_cli.core.optimizers.mean_variance.MeanVarianceOptimizer href "" "fund_cli.core.optimizers.mean_variance.MeanVarianceOptimizer"
              click fund_cli.core.optimizer.Optimizer href "" "fund_cli.core.optimizer.Optimizer"

均值-方差优化器 - PORTFOLIO-OPT-001

源代码位于： src/fund_cli/core/optimizers/mean_variance.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class MeanVarianceOptimizer(Optimizer):
    """均值-方差优化器 - PORTFOLIO-OPT-001"""

    def __init__(self, risk_free_rate: float = 0.03):
        self.risk_free_rate = risk_free_rate

    def optimize(
        self,
        returns: pd.DataFrame,
        constraints: OptimizationConstraint | None = None,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """
        执行均值-方差优化

        Args:
            returns: 收益率 DataFrame，每列一只基金
            constraints: 优化约束条件

        Returns:
            优化结果字典，包含 weights, expected_return, volatility, sharpe_ratio
        """
        try:
            from pypfopt import EfficientFrontier, expected_returns, risk_models
        except ImportError:
            return self._fallback_optimize(returns, constraints)

        try:
            mu = expected_returns.mean_historical_return(returns)
            S = risk_models.sample_cov(returns)

            min_w = constraints.min_weight if constraints else 0.0
            max_w = constraints.max_weight if constraints else 1.0

            ef = EfficientFrontier(mu, S, weight_bounds=(min_w, max_w))

            if constraints and constraints.target_return is not None:
                ef.efficient_return(constraints.target_return)
            else:
                ef.max_sharpe(risk_free_rate=self.risk_free_rate)

            weights = ef.clean_weights()
            perf = ef.portfolio_performance(verbose=False, risk_free_rate=self.risk_free_rate)

            return {
                "weights": {k: round(v, 6) for k, v in weights.items()},
                "expected_return": round(perf[0], 6),
                "volatility": round(perf[1], 6),
                "sharpe_ratio": round(perf[2], 4),
                "method": "mean_variance",
            }
        except Exception:
            return self._fallback_optimize(returns, constraints)

    def get_methods(self) -> list[str]:
        return ["mean_variance"]

    @staticmethod
    def _fallback_optimize(
        returns: pd.DataFrame, constraints: OptimizationConstraint | None = None
    ) -> dict[str, Any]:
        """PyPortfolioOpt不可用时的回退实现"""
        n = returns.shape[1]
        equal_weights = {col: round(1.0 / n, 6) for col in returns.columns}
        port_return = (returns.mean() * 252).mean()
        port_vol = returns.std().mean() * np.sqrt(252)
        return {
            "weights": equal_weights,
            "expected_return": round(port_return, 6),
            "volatility": round(port_vol, 6),
            "sharpe_ratio": round((port_return - 0.03) / port_vol, 4) if port_vol > 0 else 0.0,
            "method": "equal_weight_fallback",
        }

optimize 

optimize(
    returns: DataFrame,
    constraints: OptimizationConstraint | None = None,
    **kwargs: Any,
) -> dict[str, Any]

执行均值-方差优化

参数：

名称 类型 描述 默认
returns DataFrame

收益率 DataFrame，每列一只基金

 必需
constraints OptimizationConstraint | None

优化约束条件

 None

返回：

类型 描述
dict[str, Any]

优化结果字典，包含 weights, expected_return, volatility, sharpe_ratio
源代码位于： src/fund_cli/core/optimizers/mean_variance.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def optimize(
    self,
    returns: pd.DataFrame,
    constraints: OptimizationConstraint | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """
    执行均值-方差优化

    Args:
        returns: 收益率 DataFrame，每列一只基金
        constraints: 优化约束条件

    Returns:
        优化结果字典，包含 weights, expected_return, volatility, sharpe_ratio
    """
    try:
        from pypfopt import EfficientFrontier, expected_returns, risk_models
    except ImportError:
        return self._fallback_optimize(returns, constraints)

    try:
        mu = expected_returns.mean_historical_return(returns)
        S = risk_models.sample_cov(returns)

        min_w = constraints.min_weight if constraints else 0.0
        max_w = constraints.max_weight if constraints else 1.0

        ef = EfficientFrontier(mu, S, weight_bounds=(min_w, max_w))

        if constraints and constraints.target_return is not None:
            ef.efficient_return(constraints.target_return)
        else:
            ef.max_sharpe(risk_free_rate=self.risk_free_rate)

        weights = ef.clean_weights()
        perf = ef.portfolio_performance(verbose=False, risk_free_rate=self.risk_free_rate)

        return {
            "weights": {k: round(v, 6) for k, v in weights.items()},
            "expected_return": round(perf[0], 6),
            "volatility": round(perf[1], 6),
            "sharpe_ratio": round(perf[2], 4),
            "method": "mean_variance",
        }
    except Exception:
        return self._fallback_optimize(returns, constraints)

MaxSharpeOptimizer

Bases: Optimizer


              flowchart TD
              fund_cli.core.optimizers.max_sharpe.MaxSharpeOptimizer[MaxSharpeOptimizer]
              fund_cli.core.optimizer.Optimizer[Optimizer]

                              fund_cli.core.optimizer.Optimizer --> fund_cli.core.optimizers.max_sharpe.MaxSharpeOptimizer
                


              click fund_cli.core.optimizers.max_sharpe.MaxSharpeOptimizer href "" "fund_cli.core.optimizers.max_sharpe.MaxSharpeOptimizer"
              click fund_cli.core.optimizer.Optimizer href "" "fund_cli.core.optimizer.Optimizer"

最大夏普比率优化器 - PORTFOLIO-OPT-002

源代码位于： src/fund_cli/core/optimizers/max_sharpe.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class MaxSharpeOptimizer(Optimizer):
    """最大夏普比率优化器 - PORTFOLIO-OPT-002"""

    def __init__(self, risk_free_rate: float = 0.03):
        self.risk_free_rate = risk_free_rate

    def optimize(
        self,
        returns: pd.DataFrame,
        constraints: OptimizationConstraint | None = None,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """执行最大夏普比率优化"""
        try:
            from pypfopt import EfficientFrontier, expected_returns, risk_models
        except ImportError:
            return MeanVarianceOptimizer._fallback_optimize(returns, constraints)

        try:
            mu = expected_returns.mean_historical_return(returns)
            S = risk_models.sample_cov(returns)

            min_w = constraints.min_weight if constraints else 0.0
            max_w = constraints.max_weight if constraints else 1.0

            ef = EfficientFrontier(mu, S, weight_bounds=(min_w, max_w))
            ef.max_sharpe(risk_free_rate=self.risk_free_rate)

            weights = ef.clean_weights()
            perf = ef.portfolio_performance(verbose=False, risk_free_rate=self.risk_free_rate)

            return {
                "weights": {k: round(v, 6) for k, v in weights.items()},
                "expected_return": round(perf[0], 6),
                "volatility": round(perf[1], 6),
                "sharpe_ratio": round(perf[2], 4),
                "method": "max_sharpe",
            }
        except Exception:
            return MeanVarianceOptimizer._fallback_optimize(returns, constraints)

    def get_methods(self) -> list[str]:
        return ["max_sharpe"]

optimize 

optimize(
    returns: DataFrame,
    constraints: OptimizationConstraint | None = None,
    **kwargs: Any,
) -> dict[str, Any]

执行最大夏普比率优化

源代码位于： src/fund_cli/core/optimizers/max_sharpe.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def optimize(
    self,
    returns: pd.DataFrame,
    constraints: OptimizationConstraint | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """执行最大夏普比率优化"""
    try:
        from pypfopt import EfficientFrontier, expected_returns, risk_models
    except ImportError:
        return MeanVarianceOptimizer._fallback_optimize(returns, constraints)

    try:
        mu = expected_returns.mean_historical_return(returns)
        S = risk_models.sample_cov(returns)

        min_w = constraints.min_weight if constraints else 0.0
        max_w = constraints.max_weight if constraints else 1.0

        ef = EfficientFrontier(mu, S, weight_bounds=(min_w, max_w))
        ef.max_sharpe(risk_free_rate=self.risk_free_rate)

        weights = ef.clean_weights()
        perf = ef.portfolio_performance(verbose=False, risk_free_rate=self.risk_free_rate)

        return {
            "weights": {k: round(v, 6) for k, v in weights.items()},
            "expected_return": round(perf[0], 6),
            "volatility": round(perf[1], 6),
            "sharpe_ratio": round(perf[2], 4),
            "method": "max_sharpe",
        }
    except Exception:
        return MeanVarianceOptimizer._fallback_optimize(returns, constraints)

RiskParityOptimizer

Bases: Optimizer


              flowchart TD
              fund_cli.core.optimizers.risk_parity.RiskParityOptimizer[RiskParityOptimizer]
              fund_cli.core.optimizer.Optimizer[Optimizer]

                              fund_cli.core.optimizer.Optimizer --> fund_cli.core.optimizers.risk_parity.RiskParityOptimizer
                


              click fund_cli.core.optimizers.risk_parity.RiskParityOptimizer href "" "fund_cli.core.optimizers.risk_parity.RiskParityOptimizer"
              click fund_cli.core.optimizer.Optimizer href "" "fund_cli.core.optimizer.Optimizer"

风险平价优化器 - PORTFOLIO-OPT-003

源代码位于： src/fund_cli/core/optimizers/risk_parity.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class RiskParityOptimizer(Optimizer):
    """风险平价优化器 - PORTFOLIO-OPT-003"""

    def optimize(
        self,
        returns: pd.DataFrame,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """执行风险平价优化"""
        try:
            from pypfopt import HRPOpt
        except ImportError:
            return self._fallback_risk_parity(returns)

        hrp = HRPOpt(returns)
        weights = hrp.optimize()

        port_returns = (returns * pd.Series(weights)).sum(axis=1)
        expected_return = port_returns.mean() * 252
        volatility = port_returns.std() * np.sqrt(252)
        sharpe = (expected_return - 0.03) / volatility if volatility > 0 else 0.0

        return {
            "weights": {k: round(v, 6) for k, v in weights.items()},
            "expected_return": round(expected_return, 6),
            "volatility": round(volatility, 6),
            "sharpe_ratio": round(sharpe, 4),
            "method": "risk_parity",
        }

    def get_methods(self) -> list[str]:
        return ["risk_parity"]

    @staticmethod
    def _fallback_risk_parity(returns: pd.DataFrame) -> dict[str, Any]:
        """回退实现：基于波动率倒数加权"""
        vols = returns.std()
        inv_vols = 1.0 / vols
        weights = inv_vols / inv_vols.sum()
        port_returns = (returns * weights).sum(axis=1)
        expected_return = port_returns.mean() * 252
        volatility = port_returns.std() * np.sqrt(252)
        sharpe = (expected_return - 0.03) / volatility if volatility > 0 else 0.0
        return {
            "weights": {k: round(v, 6) for k, v in weights.items()},
            "expected_return": round(expected_return, 6),
            "volatility": round(volatility, 6),
            "sharpe_ratio": round(sharpe, 4),
            "method": "inverse_volatility_fallback",
        }

optimize 

optimize(
    returns: DataFrame, **kwargs: Any
) -> dict[str, Any]

执行风险平价优化

源代码位于： src/fund_cli/core/optimizers/risk_parity.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def optimize(
    self,
    returns: pd.DataFrame,
    **kwargs: Any,
) -> dict[str, Any]:
    """执行风险平价优化"""
    try:
        from pypfopt import HRPOpt
    except ImportError:
        return self._fallback_risk_parity(returns)

    hrp = HRPOpt(returns)
    weights = hrp.optimize()

    port_returns = (returns * pd.Series(weights)).sum(axis=1)
    expected_return = port_returns.mean() * 252
    volatility = port_returns.std() * np.sqrt(252)
    sharpe = (expected_return - 0.03) / volatility if volatility > 0 else 0.0

    return {
        "weights": {k: round(v, 6) for k, v in weights.items()},
        "expected_return": round(expected_return, 6),
        "volatility": round(volatility, 6),
        "sharpe_ratio": round(sharpe, 4),
        "method": "risk_parity",
    }

EfficientFrontierCalculator

有效前沿计算器 - PORTFOLIO-OPT-006

源代码位于： src/fund_cli/core/optimizers/efficient_frontier.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class EfficientFrontierCalculator:
    """有效前沿计算器 - PORTFOLIO-OPT-006"""

    def calculate(
        self,
        returns: pd.DataFrame,
        n_points: int = 50,
        risk_free_rate: float = 0.03,
    ) -> dict[str, Any]:
        """
        计算有效前沿上的点集

        Args:
            returns: 收益率 DataFrame
            n_points: 前沿点数
            risk_free_rate: 无风险利率

        Returns:
            包含 frontier_returns, frontier_volatilities, frontier_sharpes 的字典
        """
        try:
            from pypfopt import EfficientFrontier, expected_returns, risk_models

            mu = expected_returns.mean_historical_return(returns)
            S = risk_models.sample_cov(returns)

            min_ret = mu.min()
            max_ret = mu.max()
            target_returns = np.linspace(min_ret, max_ret, n_points)

            frontier_volatilities = []
            frontier_returns = []
            frontier_sharpes = []

            for target in target_returns:
                try:
                    ef = EfficientFrontier(mu, S)
                    ef.efficient_return(target)
                    ret, vol, _ = ef.portfolio_performance(
                        verbose=False, risk_free_rate=risk_free_rate
                    )
                    frontier_returns.append(ret)
                    frontier_volatilities.append(vol)
                    frontier_sharpes.append((ret - risk_free_rate) / vol if vol > 0 else 0)
                except Exception:
                    continue

            if len(frontier_returns) == 0:
                return self._fallback_calculate(returns, n_points, risk_free_rate)

            return {
                "frontier_returns": [round(r, 6) for r in frontier_returns],
                "frontier_volatilities": [round(v, 6) for v in frontier_volatilities],
                "frontier_sharpes": [round(s, 4) for s in frontier_sharpes],
                "n_points": len(frontier_returns),
            }
        except Exception:
            return self._fallback_calculate(returns, n_points, risk_free_rate)

    @staticmethod
    def _fallback_calculate(
        returns: pd.DataFrame, n_points: int, risk_free_rate: float
    ) -> dict[str, Any]:
        """回退实现"""
        n = returns.shape[1]
        vols = []
        rets = []
        for i in range(n_points):
            np.random.seed(i)
            w = np.random.dirichlet(np.ones(n))
            port_ret = (returns.mean() * 252 * w).sum()
            port_vol = np.sqrt(w @ returns.cov().values @ w * 252)
            vols.append(port_vol)
            rets.append(port_ret)
        return {
            "frontier_returns": [round(r, 6) for r in rets],
            "frontier_volatilities": [round(v, 6) for v in vols],
            "frontier_sharpes": [
                round((r - risk_free_rate) / v, 4) if v > 0 else 0
                for r, v in zip(rets, vols, strict=False)
            ],
            "n_points": n,
        }

calculate 

calculate(
    returns: DataFrame,
    n_points: int = 50,
    risk_free_rate: float = 0.03,
) -> dict[str, Any]

计算有效前沿上的点集

参数：

名称 类型 描述 默认
returns DataFrame

收益率 DataFrame

 必需
n_points int

前沿点数

 50
risk_free_rate float

无风险利率

 0.03

返回：

类型 描述
dict[str, Any]

包含 frontier_returns, frontier_volatilities, frontier_sharpes 的字典
源代码位于： src/fund_cli/core/optimizers/efficient_frontier.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def calculate(
    self,
    returns: pd.DataFrame,
    n_points: int = 50,
    risk_free_rate: float = 0.03,
) -> dict[str, Any]:
    """
    计算有效前沿上的点集

    Args:
        returns: 收益率 DataFrame
        n_points: 前沿点数
        risk_free_rate: 无风险利率

    Returns:
        包含 frontier_returns, frontier_volatilities, frontier_sharpes 的字典
    """
    try:
        from pypfopt import EfficientFrontier, expected_returns, risk_models

        mu = expected_returns.mean_historical_return(returns)
        S = risk_models.sample_cov(returns)

        min_ret = mu.min()
        max_ret = mu.max()
        target_returns = np.linspace(min_ret, max_ret, n_points)

        frontier_volatilities = []
        frontier_returns = []
        frontier_sharpes = []

        for target in target_returns:
            try:
                ef = EfficientFrontier(mu, S)
                ef.efficient_return(target)
                ret, vol, _ = ef.portfolio_performance(
                    verbose=False, risk_free_rate=risk_free_rate
                )
                frontier_returns.append(ret)
                frontier_volatilities.append(vol)
                frontier_sharpes.append((ret - risk_free_rate) / vol if vol > 0 else 0)
            except Exception:
                continue

        if len(frontier_returns) == 0:
            return self._fallback_calculate(returns, n_points, risk_free_rate)

        return {
            "frontier_returns": [round(r, 6) for r in frontier_returns],
            "frontier_volatilities": [round(v, 6) for v in frontier_volatilities],
            "frontier_sharpes": [round(s, 4) for s in frontier_sharpes],
            "n_points": len(frontier_returns),
        }
    except Exception:
        return self._fallback_calculate(returns, n_points, risk_free_rate)