MCM-ICM_Toolkit/tests/test_stats.py

"""stats 包测试。

原则（见 docs/02-architecture.md §5）：**测已知解析解，不测实现细节**。

未实现的函数**逐条** `skip`——**不要**用模块级 `pytestmark`：那会把同文件里
已经写好的函数一起跳过，绿灯就是假的（决策 26 / 个人易错点 9）。
"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

_NOT_IMPLEMENTED = "未实现：见 docs/03-roadmap.md L1"


@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
def test_bootstrap_ci_matches_normal_theory_on_large_n(rng):
    """大样本下 bootstrap 百分位区间应逼近正态解析区间（±10% 容差）。"""
    from mcm_toolkit.stats import bootstrap_ci

    x = rng.normal(10, 2, size=5000)
    r = bootstrap_ci(x, statistic=np.mean, confidence=0.95, n_resamples=5000, random_state=0)
    se = x.std(ddof=1) / np.sqrt(x.size)
    assert abs((r.params["high"] - r.params["low"]) - 2 * 1.96 * se) < 0.1 * 2 * 1.96 * se


@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
def test_bootstrap_ci_works_for_arbitrary_statistic(rng):
    """bootstrap 的核心价值：任意统计量都能给区间（这里用中位数）。"""
    from mcm_toolkit.stats import bootstrap_ci

    x = rng.lognormal(0, 1, size=2000)
    r = bootstrap_ci(x, statistic=np.median, random_state=0)
    assert r.params["low"] < r.params["high"]
    assert r.params["low"] < np.median(x) < r.params["high"]


@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
def test_bootstrap_ci_is_reproducible(rng):
    from mcm_toolkit.stats import bootstrap_ci

    x = rng.normal(size=500)
    a = bootstrap_ci(x, statistic=np.mean, random_state=123)
    b = bootstrap_ci(x, statistic=np.mean, random_state=123)
    assert a.params["low"] == b.params["low"]


@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
def test_confidence_interval_matches_scipy(rng):
    from scipy import stats

    from mcm_toolkit.stats import confidence_interval

    x = rng.normal(5, 1, size=50)
    lo, hi = confidence_interval(x, confidence=0.95, method="t")
    ref = stats.t.interval(0.95, len(x) - 1, loc=x.mean(), scale=stats.sem(x))
    assert np.allclose([lo, hi], ref, rtol=1e-6)


@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
def test_correlation_matrix_diagonal_is_one(rng):
    from mcm_toolkit.stats import correlation_matrix

    x = rng.normal(size=(200, 4))
    c = correlation_matrix(x)
    assert np.allclose(np.diag(np.asarray(c)), 1.0)


# ------------------------------------------------------ stats: describe


def test_describe_reports_expected_keys(rng):
    from mcm_toolkit.stats import describe

    table = describe(rng.normal(size=(100, 3)))
    for key in ["mean", "std", "min", "max", "median"]:
        assert key in table.index


def test_describe_rows_are_metrics_and_columns_are_variables():
    """方向约定：行 = 指标，列 = 变量。多转置一次就反了。"""
    from mcm_toolkit.stats import describe

    df = pd.DataFrame({"a": [1.0, 2.0, 3.0], "b": [10.0, 20.0, 30.0]})
    table = describe(df)

    assert list(table.columns) == ["a", "b"]
    assert table.loc["mean", "a"] == pytest.approx(2.0)
    assert table.loc["max", "b"] == pytest.approx(30.0)
    assert table.loc["median", "b"] == pytest.approx(20.0)


def test_describe_missing_and_shape_rows_are_not_nan():
    """曾经的静默 bug：转置后 `.loc[指标名] = <按变量索引的 Series>` 对不上列，
    整行会被填成 NaN —— 不报错，但数据全丢。"""
    from mcm_toolkit.stats import describe

    df = pd.DataFrame({"a": [1.0, np.nan, 3.0, 4.0], "b": [1.0, 2.0, 3.0, 4.0]})
    table = describe(df)

    assert table.loc["missing", "a"] == pytest.approx(1.0)
    assert table.loc["missing", "b"] == pytest.approx(0.0)
    assert table.loc["missing_ratio", "a"] == pytest.approx(0.25)
    assert table.loc["missing_ratio", "b"] == pytest.approx(0.0)
    assert table.loc["skewness", "b"] == pytest.approx(0.0, abs=1e-12)
    extra = table.loc[["missing", "missing_ratio", "skewness", "kurtosis"]]
    assert not bool(extra.isna().all(axis=None))


def test_describe_skips_non_numeric_columns():
    from mcm_toolkit.stats import describe

    df = pd.DataFrame({"a": [1.0, 2.0], "city": ["X", "Y"]})
    table = describe(df)

    assert list(table.columns) == ["a"]


def test_describe_accepts_ndarray_list_and_series(rng):
    """兄弟函数都收 ndarray / list，describe 没理由只收 DataFrame。"""
    from mcm_toolkit.stats import describe

    assert "mean" in describe(rng.normal(size=(50, 2))).index
    assert "mean" in describe([[1.0, 2.0], [3.0, 4.0]]).index
    assert "mean" in describe(pd.Series([1.0, 2.0, 3.0])).index


def test_describe_weighted_rows_match_manual_formula():
    from mcm_toolkit.stats import describe

    df = pd.DataFrame({"a": [1.0, 2.0, 3.0]})
    w = np.array([1.0, 1.0, 2.0])
    table = describe(df, weights=w)

    expect_mean = np.average(df["a"], weights=w)
    expect_std = np.sqrt(np.average((df["a"] - expect_mean) ** 2, weights=w))
    assert table.loc["weighted_mean", "a"] == pytest.approx(expect_mean)
    assert table.loc["weighted_std", "a"] == pytest.approx(expect_std)


@pytest.mark.parametrize(
    ("weights", "exc"),
    [
        (np.ones(2), ValueError),
        (np.zeros(3), ValueError),
        (np.array([1.0, -1.0, 1.0]), ValueError),
        (np.array([1.0, np.nan, 1.0]), ValueError),
        ("abc", TypeError),
    ],
)
def test_describe_rejects_bad_weights(weights, exc):
    from mcm_toolkit.stats import describe

    with pytest.raises(exc):
        describe(pd.DataFrame({"a": [1.0, 2.0, 3.0]}), weights=weights)


def test_describe_rejects_unsupported_type_and_all_text_frame():
    from mcm_toolkit.stats import describe

    with pytest.raises(TypeError):
        describe({"a": [1.0]})
    with pytest.raises(ValueError):
        describe(pd.DataFrame({"city": ["X", "Y"]}))


# ------------------------------------------------------ stats: zscore


def test_zscore_matches_manual_formula():
    from mcm_toolkit.stats import zscore

    x = np.array([1.0, 2.0, 3.0, 4.0])
    assert np.allclose(np.asarray(zscore(x)), (x - x.mean()) / x.std())


def test_zscore_is_per_column_and_keeps_frame(rng):
    from mcm_toolkit.stats import zscore

    df = pd.DataFrame({"a": rng.normal(size=200), "b": rng.normal(5, 3, size=200)})
    out = zscore(df)

    assert isinstance(out, pd.DataFrame)
    assert list(out.columns) == ["a", "b"]
    assert np.allclose(np.asarray(out).mean(axis=0), 0.0, atol=1e-12)
    assert np.allclose(np.asarray(out).std(axis=0), 1.0, atol=1e-12)


def test_zscore_constant_column_stays_zero():
    from mcm_toolkit.stats import zscore

    assert np.allclose(np.asarray(zscore([2.0, 2.0, 2.0])), 0.0)


def test_zscore_rejects_bad_type_and_all_text_frame():
    from mcm_toolkit.stats import zscore

    with pytest.raises(TypeError):
        zscore("abc")
    with pytest.raises(ValueError):
        zscore(pd.DataFrame({"city": ["X"]}))


# ------------------------------------------- stats: coefficient_of_variation


def test_coefficient_of_variation_matches_manual_formula():
    from mcm_toolkit.stats import coefficient_of_variation

    x = np.array([1.0, 2.0, 3.0, 4.0])
    assert coefficient_of_variation(x) == pytest.approx(x.std() / x.mean())


def test_coefficient_of_variation_accepts_single_column_frame():
    from mcm_toolkit.stats import coefficient_of_variation

    df = pd.DataFrame({"a": [1.0, 2.0, 3.0]})
    single = coefficient_of_variation([1.0, 2.0, 3.0])
    assert coefficient_of_variation(df) == pytest.approx(single)


def test_coefficient_of_variation_uses_absolute_mean():
    """均值为负时 CV 仍是**正数**：符号不该被读成"波动更小"。"""
    from mcm_toolkit.stats import coefficient_of_variation

    x = np.array([-1.0, -2.0, -3.0, -4.0])
    assert coefficient_of_variation(x) == pytest.approx(x.std() / abs(x.mean()))
    assert coefficient_of_variation(x) > 0


def test_coefficient_of_variation_zero_mean_is_value_error():
    """均值为 0 是**取值**问题不是类型问题：必须是 ValueError 而不是 TypeError。"""
    from mcm_toolkit.stats import coefficient_of_variation

    with pytest.raises(ValueError):
        coefficient_of_variation([-1.0, 1.0])


def test_coefficient_of_variation_rejects_empty_2d_and_bad_type():
    from mcm_toolkit.stats import coefficient_of_variation

    with pytest.raises(ValueError):
        coefficient_of_variation([])
    with pytest.raises(ValueError):
        coefficient_of_variation([[1.0, 2.0], [3.0, 4.0]])
    with pytest.raises(ValueError):
        coefficient_of_variation(pd.DataFrame({"a": [1.0], "b": [2.0]}))
    with pytest.raises(TypeError):
        coefficient_of_variation(5.0)