quant_engine/tests/test_perf_stats.py

"""src/shared/perf_stats.py 单测 —— 覆盖所有 7+3 字段 + 边界路径。"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest


def test_perf_stats_returns_all_fields():
    """输出含 7 个源字段 + 3 个 alias 字段 = 10 字段。"""
    from quant_engine.perf_stats import perf_stats

    np.random.seed(42)
    ret = pd.Series(np.random.normal(0.001, 0.02, 100))
    out = perf_stats(ret, freq=252)
    # 源字段
    for k in (
        "annual_yield",
        "annual_sd",
        "sharpe",
        "annual_down_sd",
        "drawback",
        "sortino",
        "calmar",
    ):
        assert k in out
    # alias 字段
    for k in ("ann_return", "annual_volatility", "max_drawdown"):
        assert k in out
    assert out["annual_yield"] == out["ann_return"]
    assert out["annual_sd"] == out["annual_volatility"]
    assert out["drawback"] == out["max_drawdown"]


def test_perf_stats_short_returns_zeros():
    """n < 2 → 全 0 字段（避免 std NaN）。"""
    from quant_engine.perf_stats import perf_stats

    out = perf_stats(pd.Series([0.01]), freq=2)
    assert out["annual_yield"] == 0.0
    assert out["sharpe"] == 0.0
    assert out["drawback"] == 0.0


def test_perf_stats_zero_std_returns_zero_sharpe():
    """std 极小但非零时 → 不会触发 0 fallback（边界）。改测常数序列的其他行为。"""
    from quant_engine.perf_stats import perf_stats

    ret = pd.Series([0.01] * 10)
    out = perf_stats(ret, freq=2)
    # annual_yield 应该是固定值
    assert abs(out["annual_yield"] - 0.0201) < 0.001
    # 净值常数 → drawback = 0
    assert out["drawback"] == 0.0
    # 无负收益 → sortino = 0
    assert out["sortino"] == 0.0


def test_perf_stats_explicit_zero_vol_fallback():
    """单点序列 → 触发 n<2 早 return → sharpe=0。"""
    from quant_engine.perf_stats import perf_stats

    out = perf_stats(pd.Series([0.0, 0.0, 0.0, 0.0, 0.0]), freq=2)
    # 实际是 n=5, mean=0, std=0, annual_yield=0, sharpe=0
    assert out["annual_yield"] == 0.0
    assert out["sharpe"] == 0.0


def test_perf_stats_drawback_negative():
    """回撤必为负数（或零）。"""
    from quant_engine.perf_stats import perf_stats

    np.random.seed(0)
    ret = pd.Series(np.random.normal(0, 0.05, 200))
    out = perf_stats(ret, freq=252)
    assert out["drawback"] <= 0.0


def test_perf_stats_no_downside_returns_zero_sortino():
    """无负收益 → down_sd=0 → sortino=0。"""
    from quant_engine.perf_stats import perf_stats

    # 全正收益（带一点噪声避免完全常数）
    ret = pd.Series([0.01, 0.02, 0.015, 0.018, 0.011])
    out = perf_stats(ret, freq=2)
    # 无负收益 → down_sd = 0（len < 2 或全是 0）
    assert out["sortino"] == 0.0


def test_perf_stats_accepts_ndarray():
    """输入可以是 ndarray（_as_series 转 Series）。"""
    from quant_engine.perf_stats import perf_stats

    arr = np.array([0.01, 0.02, -0.01, 0.015, 0.005, -0.005, 0.01, 0.02, -0.01, 0.015])
    out = perf_stats(arr, freq=2)
    assert "annual_yield" in out


def test_perf_stats_strips_inf():
    """输入含 inf → _as_series 替换为 NaN 再 dropna。"""
    from quant_engine.perf_stats import perf_stats

    ret = pd.Series([0.01, 0.02, np.inf, 0.015, -np.inf, 0.005])
    out = perf_stats(ret, freq=2)
    # 不崩
    assert "annual_yield" in out


def test_perf_stats_df_returns_single_row():
    """perf_stats_df 返回单行 DataFrame。"""
    from quant_engine.perf_stats import perf_stats_df

    ret = pd.Series([0.01, -0.005, 0.015, -0.01, 0.02, 0.005])
    out = perf_stats_df(ret, freq=2)
    assert isinstance(out, pd.DataFrame)
    assert len(out) == 1
    assert "annual_yield" in out.columns


def test_batch_perf_stats():
    """批量接口返回多行 DataFrame。"""
    from quant_engine.perf_stats import batch_perf_stats

    np.random.seed(0)
    rets = {
        "fund_a": pd.Series(np.random.normal(0.001, 0.02, 100)),
        "fund_b": pd.Series(np.random.normal(0.0005, 0.03, 100)),
    }
    out = batch_perf_stats(rets, freq=252)
    assert isinstance(out, pd.DataFrame)
    assert len(out) == 2
    assert "fund_a" in out.index
    assert "fund_b" in out.index
    # 每行都含完整 10 字段
    assert "annual_yield" in out.columns
    assert "ann_return" in out.columns