经常会遇到这样的需求,要把一组数据按分位数划分成4档打分。Java端用Apache Commons Math算出来的结果,和Python端用NumPy算出来的不一样。排查了一下,发现是两边默认的分位数算法不同,记录一下这个问题。

什么是分位数

分位数(Quantile)是把一组有序数据分成若干等份的分割点。最常用的是四分位数:P25、P50、P75,它们把数据分成4段,每段大约包含25%的数据。P50就是中位数。

举个例子,一组考试成绩的P75是82分,意思是大约75%的人在82分及以下(小于等于82)。现在需要根据P25、P50、P75把每个数据映射成1~4的分数,落在哪个区间就给几分。

Java Apache Commons Math 的实现

package com.example.demo.percentile;

import org.apache.commons.math3.stat.descriptive.rank.Percentile;
import java.util.Arrays;

public class PercentileExample {
    private static final String TEST_NUMS = """
            0, 0, 1, 1, 1, 2, 3, 4, 5, 7
            0, 0, 4, 4, 6, 6, 8, 10, 11, 14
            20, 32, 58, 69, 74, 81, 82, 84, 84, 88
            19, 31, 40, 58, 68, 72, 84, 84, 94, 142
            0, 0, 0, 0, 0, 0, 0, 0, 0, 1
            0, 0, 0, 0, 0, 0, 0, 0, 6, 129
            99, 99, 99, 99, 99, 99, 99, 99, 100, 101
            """;

    public static double[][] parseLinesToDoubleArrays(String multiLine) {
        if (multiLine == null || multiLine.isEmpty()) {
            return new double[0][0];
        }

        String[] lines = multiLine.split("\\R");
        int nonEmptyCount = 0;
        for (String line : lines) {
            if (!line.trim().isEmpty()) {
                nonEmptyCount++;
            }
        }

        double[][] result = new double[nonEmptyCount][];
        int index = 0;
        for (String line : lines) {
            String trimmed = line.trim();
            if (trimmed.isEmpty()) {
                continue;
            }
            String[] parts = trimmed.split("\\s*,\\s*");
            double[] row = new double[parts.length];
            for (int i = 0; i < parts.length; i++) {
                row[i] = Double.parseDouble(parts[i]);
            }
            result[index++] = row;
        }
        return result;
    }

    public static int[] percent(double[] list) {
        if (list == null || list.length == 0) {
            return new int[0];
        }
        Percentile percentile = new Percentile();
        percentile.setData(list);

        double p25 = percentile.evaluate(25.0);
        double p50 = percentile.evaluate(50.0);
        double p75 = percentile.evaluate(75.0);

        System.out.printf("P25=%.2f, P50=%.2f, P75=%.2f\n", p25, p50, p75);

        int[] scores = new int[list.length];
        for (int i = 0; i < list.length; i++) {
            double value = list[i];
            if (value <= p25) {
                scores[i] = 1;
            } else if (value <= p50) {
                scores[i] = 2;
            } else if (value <= p75) {
                scores[i] = 3;
            } else {
                scores[i] = 4;
            }
        }
        return scores;
    }

    public static void main(String[] args) {
        double[][] testNums = parseLinesToDoubleArrays(TEST_NUMS);
        for (double[] list : testNums) {
            System.out.println("原始数据:" + Arrays.toString(list));
            int[] scores = percent(list);
            System.out.println(Arrays.toString(scores));
            System.out.println("--------");
        }
    }
}

Python NumPy 的实现

import numpy as np

TEST_NUMS = """
0, 0, 1, 1, 1, 2, 3, 4, 5, 7
0, 0, 4, 4, 6, 6, 8, 10, 11, 14
20, 32, 58, 69, 74, 81, 82, 84, 84, 88
19, 31, 40, 58, 68, 72, 84, 84, 94, 142
0, 0, 0, 0, 0, 0, 0, 0, 0, 1
0, 0, 0, 0, 0, 0, 0, 0, 6, 129
99, 99, 99, 99, 99, 99, 99, 99, 100, 101
"""

def parse_lines_to_double_arrays(multi_line):
    if not multi_line:
        return []
    lines = multi_line.strip().splitlines()
    result = []
    for line in lines:
        line = line.strip()
        if not line:
            continue
        parts = [float(x.strip()) for x in line.split(',') if x.strip()]
        result.append(parts)
    return result


def percent(data):
    if not data:
        return []
    arr = np.array(data)
    p25 = np.percentile(arr, 25)
    p50 = np.percentile(arr, 50)
    p75 = np.percentile(arr, 75)
    print(f"P25: {p25}, P50: {p50}, P75: {p75}")

    scores = np.zeros_like(arr, dtype=int)
    scores[arr <= p25] = 1
    scores[(arr > p25) & (arr <= p50)] = 2
    scores[(arr > p50) & (arr <= p75)] = 3
    scores[arr > p75] = 4
    return scores.tolist()

def main():
    test_nums = parse_lines_to_double_arrays(TEST_NUMS)
    for row in test_nums:
        print("原始数据:", row)
        scores = percent(row)
        print(scores)
        print("--------")

if __name__ == "__main__":
    main()

跑同样的数据,两边出来的分位数值不一样。问题出在算法上。

Java代码执行结果:
原始数据:[0.0, 0.0, 1.0, 1.0, 1.0, 2.0, 3.0, 4.0, 5.0, 7.0]
P25=0.75, P50=1.50, P75=4.25
[1, 1, 2, 2, 2, 3, 3, 3, 4, 4]
--------
原始数据:[0.0, 0.0, 4.0, 4.0, 6.0, 6.0, 8.0, 10.0, 11.0, 14.0]
P25=3.00, P50=6.00, P75=10.25
[1, 1, 2, 2, 2, 2, 3, 3, 4, 4]
...

Python代码执行结果:
原始数据: [0.0, 0.0, 1.0, 1.0, 1.0, 2.0, 3.0, 4.0, 5.0, 7.0]
P25: 1.0, P50: 1.5, P75: 3.75
[1, 1, 1, 1, 1, 3, 3, 4, 4, 4]
--------
原始数据: [0.0, 0.0, 4.0, 4.0, 6.0, 6.0, 8.0, 10.0, 11.0, 14.0]
P25: 4.0, P50: 6.0, P75: 9.5
[1, 1, 1, 1, 2, 2, 3, 4, 4, 4]
...

差异分析

分位数的计算方法其实不止一种,统计学里一共定义了9种(R语言里叫type 1~9)。不同的库选了不同的默认算法,这就是两边结果对不上的原因。

  • Apache Commons Math 默认用 R-6(NIST推荐,也是Minitab、SPSS的默认)
  • NumPy 默认用 R-7(Excel PERCENTILE.INC 的默认)

两种算法的思路是一样的:先把数据排序,然后根据百分位算出一个”虚拟位置”,最后在相邻两个元素之间做线性插值。区别只在于”虚拟位置”的计算公式不同。

用上面数据的第一行来具体说明,排序后是 [0, 0, 1, 1, 1, 2, 3, 4, 5, 7],现在要算P25:

R-6的做法:

位置 = (n+1) * 0.25 = (10+1) * 0.25 = 2.75

意思是P25落在第2个和第3个元素之间,偏向第3个(0.75的权重)。第2个是0,第3个是1,所以

P25 = 0 + 0.75 × (1-0) = 0.75

R-7的做法:

位置 = 1 + (n-1) * 0.25 = 1 + (10-1) * 0.25 = 3.25

意思是P25落在第3个和第4个元素之间,偏向第4个的权重是 0.25。第3个是1,第4个也是1,所以

P25 = 1 + 0.25 × (1-1) = 1.0

简单的说,R-6假定样本的最大最小值并不是一系列数据的极值,而R-7假定样本的最大最小值就是所有数据的极值。因此,一样的数据,R-6算出来P25是0.75,R-7算出来是1.0,后面的打分结果自然就不一样了。用公式总结一下:

算法 位置公式 谁在用
R-6 pos = p * (N+1) / 100 Apache Commons Math, NIST, Minitab, SPSS
R-7 pos = 1 + p * (N-1) / 100 NumPy(默认), Excel, R语言(默认)

算出位置后,取整数部分 k 和小数部分 d,然后在第 k 个和第 k+1 个元素之间插值:result = v_k + d * (v_{k+1} - v_k)

怎么让两边一致? 指定相同的算法就行。Apache Commons Math 可以通过 .withEstimationType() 切换(R-1到R-9都支持);NumPy 通过 method 参数指定,"weibull" 对应 R-6,"linear"(默认)对应 R-7。

用 Python 手动实现 R-6 算法

为了更好理解,手动实现一个:

"""R-6算法计算百分位数,p取值0~100"""
def percentile_r6(data, p):    
    sorted_data = sorted(data)
    n = len(sorted_data)
    pos = p * (n + 1) / 100.0

    if pos < 1:
        return sorted_data[0]
    if pos >= n:
        return sorted_data[-1]

    k = int(pos)
    d = pos - k
    return sorted_data[k - 1] + d * (sorted_data[k] - sorted_data[k - 1])

Python使用上面这个函数计算后,输出如下,和 Java 端 Apache Commons Math 的结果一致:

原始数据: [0.0, 0.0, 1.0, 1.0, 1.0, 2.0, 3.0, 4.0, 5.0, 7.0]
P25: 0.75, P50: 1.5, P75: 4.25
[1, 1, 2, 2, 2, 3, 3, 3, 4, 4]
--------
原始数据: [0.0, 0.0, 4.0, 4.0, 6.0, 6.0, 8.0, 10.0, 11.0, 14.0]
P25: 3.0, P50: 6.0, P75: 10.25
[1, 1, 2, 2, 2, 2, 3, 3, 4, 4]
...