quant_engine/tests/test_portfolio_decomp.py

"""src/shared/portfolio_decomp.py 单测（v0.18.1 加，提升覆盖率 40% → 80%+）。"""

from __future__ import annotations

import numpy as np
import pytest


# ── l1_rebalance_decompose ────────────────────────────


def test_l1_rebalance_basic():
    """l1_rebalance_decompose 单资产案例：y == X[0] → β ≈ [1, 0, ..., 0]。"""
    from quant_engine.portfolio_decomp import l1_rebalance_decompose

    np.random.seed(0)
    n = 50
    X = np.random.randn(n, 3)
    # y = X[:, 0]（纯资产 0）
    y = X[:, 0]
    beta = l1_rebalance_decompose(y, X)
    assert len(beta) == 3
    assert abs(beta.sum() - 1.0) < 1e-4
    assert beta[0] > 0.9  # 主要权重在资产 0


def test_l1_rebalance_y_x_shape_mismatch():
    """l1_rebalance_decompose y 与 X 行数不一致 → ValueError。"""
    from quant_engine.portfolio_decomp import l1_rebalance_decompose

    y = np.array([1.0, 2.0, 3.0])
    X = np.array([[1.0, 2.0], [3.0, 4.0]])  # 2 行
    with pytest.raises(ValueError, match="行数不一致"):
        l1_rebalance_decompose(y, X)


def test_l1_rebalance_bounds_length_mismatch():
    """l1_rebalance_decompose bounds 长度不匹配 → ValueError。"""
    from quant_engine.portfolio_decomp import l1_rebalance_decompose

    y = np.array([1.0, 2.0])
    X = np.array([[1.0, 2.0], [3.0, 4.0]])
    bounds = [(0.0, 1.0)]  # 长度 1，资产数 2
    with pytest.raises(ValueError, match="bounds 长度"):
        l1_rebalance_decompose(y, X, bounds=bounds)


def test_l1_rebalance_with_custom_bounds():
    """l1_rebalance_decompose 自定义 bounds → 权重在 bounds 内。"""
    from quant_engine.portfolio_decomp import l1_rebalance_decompose

    np.random.seed(0)
    X = np.random.randn(50, 3)
    y = X[:, 0]
    bounds = [(0.0, 0.5), (0.0, 1.0), (0.0, 1.0)]
    beta = l1_rebalance_decompose(y, X, bounds=bounds)
    assert all(b <= 0.5 + 1e-4 for b in beta)


def test_l1_rebalance_list_input():
    """l1_rebalance_decompose 支持 list 输入（自动转 ndarray）。"""
    from quant_engine.portfolio_decomp import l1_rebalance_decompose

    y = [1.0, 2.0, 3.0, 4.0, 5.0]
    X = [[1.0, 0.0], [2.0, 0.0], [3.0, 0.0], [4.0, 0.0], [5.0, 0.0]]
    beta = l1_rebalance_decompose(y, X)
    assert len(beta) == 2
    # y 完全等于 X[:, 0]，所以 β ≈ [1, 0]
    assert beta[0] > 0.9


def test_l1_rebalance_result_in_bounds():
    """l1_rebalance_decompose 输出 β 在 [0, 1] 内。"""
    from quant_engine.portfolio_decomp import l1_rebalance_decompose

    np.random.seed(42)
    X = np.random.randn(30, 4)
    y = X.mean(axis=1)
    beta = l1_rebalance_decompose(y, X)
    assert all(0.0 <= b <= 1.0 + 1e-9 for b in beta)
    assert abs(beta.sum() - 1.0) < 1e-4


# ── risk_parity_weights ────────────────────────────


def test_risk_parity_weights_basic():
    """risk_parity_weights 简单 2 资产协方差 → 权重合理。"""
    from quant_engine.portfolio_decomp import risk_parity_weights

    # 资产 1 风险 = 1，资产 2 风险 = 2（独立）→ 风险平价权重应偏向资产 1
    cov = np.array(
        [
            [1.0, 0.0],
            [0.0, 4.0],  # 高波动
        ]
    )
    w = risk_parity_weights(cov)
    assert len(w) == 2
    assert abs(w.sum() - 1.0) < 1e-4
    # 低波动资产应有更高权重
    assert w[0] > w[1]


def test_risk_parity_weights_correlated():
    """risk_parity_weights 相关资产。"""
    from quant_engine.portfolio_decomp import risk_parity_weights

    np.random.seed(0)
    A = np.random.randn(100, 3)
    cov = np.cov(A.T)
    w = risk_parity_weights(cov)
    assert abs(w.sum() - 1.0) < 1e-3
    assert all(w > 0)


def test_risk_parity_weights_zero_variance():
    """risk_parity_weights 全 0 方差 → 退化路径（返回等权 fallback）。"""
    from quant_engine.portfolio_decomp import risk_parity_weights

    cov = np.zeros((3, 3))
    w = risk_parity_weights(cov)
    assert len(w) == 3
    # 全 0 时退化，结果可能异常但函数不应崩
    assert abs(w.sum() - 1.0) < 0.5  # 容忍大误差


# ── mean_variance_weights ────────────────────────────


def test_mean_variance_weights_basic():
    """mean_variance_weights 单调收益 + 差异协方差 → 权重偏向高收益。

    v1.2.0 修复：原 risk_aversion=1.0 时 SLSQP 把资产 1/2 卡在下边界 1e-4，
    导致 w[1] > w[2] 因浮点噪声 flaky。改用 risk_aversion=100（解析解
    [0.951, 0.045, 0.004] 在内部，稳定收敛）。
    """
    from quant_engine.portfolio_decomp import mean_variance_weights

    # mu = [0.10, 0.05, 0.02]，cov 不等（资产 1 风险最高）
    mu = np.array([0.10, 0.05, 0.02])
    cov = np.diag([0.001, 0.01, 0.04])  # 资产 0 风险最低
    w = mean_variance_weights(mu, cov, risk_aversion=100.0)
    assert len(w) == 3
    assert abs(w.sum() - 1.0) < 1e-3
    # 资产 0：低风险高收益 → 应有最高权重（解析解 [0.951, 0.045, 0.004]）
    assert w[0] > w[1] > w[2]
    # 与解析解一致（允许优化误差）
    assert w[0] == pytest.approx(0.951, abs=0.01)


def test_mean_variance_weights_high_aversion():
    """mean_variance_weights 高风险厌恶 → 权重接近等权（偏好分散）。"""
    from quant_engine.portfolio_decomp import mean_variance_weights

    mu = np.array([0.1, 0.05, 0.02])
    cov = np.diag([0.01, 0.01, 0.01])
    w_low = mean_variance_weights(mu, cov, risk_aversion=0.5)
    w_high = mean_variance_weights(mu, cov, risk_aversion=10.0)
    # 高厌恶 → 权重差异小（更接近等权 1/3）
    diff_low = w_low.max() - w_low.min()
    diff_high = w_high.max() - w_high.min()
    assert diff_high < diff_low


def test_mean_variance_weights_list_input():
    """mean_variance_weights 支持 list 输入。"""
    from quant_engine.portfolio_decomp import mean_variance_weights

    w = mean_variance_weights([0.1, 0.05], [[0.01, 0.001], [0.001, 0.01]])
    assert len(w) == 2
    assert abs(w.sum() - 1.0) < 1e-3


def test_mean_variance_weights_zero_mu():
    """mean_variance_weights 全 0 收益 → 等权。"""
    from quant_engine.portfolio_decomp import mean_variance_weights

    mu = np.zeros(3)
    cov = np.eye(3) * 0.01
    w = mean_variance_weights(mu, cov)
    # 全 0 收益 → 等权
    for wi in w:
        assert abs(wi - 1.0 / 3) < 0.01