chronaxis.core.quantile_numpy
源码基线:8904c2f744951857659abd40aaa3535d57faf53c
1"""为已付费的样本数组作原dtype全局排序,然后仅恢复两个标量端点。"""23import numpy as np4from .types import QualityFlag as Q567def ordered_numpy(chunks, index, remainder, denominator):8 """用NumPy原地堆排序得到顺序统计值,并从其精确比值一次舍入插值。910 参数: chunks为原dtype样本块,index为下秩,正分母与余数定义上秩权重。11 返回: (原生数值或None,质量),整数端点不转float。12 异常: 预算由外层预付,数组分配或索引错误直接传播。13 """14 values = None15 try:16 # 连接只产生一个全长副本,堆排序原地执行,不把uint64先改成float64。17 values = np.concatenate(chunks)18 values.sort(kind='heapsort')19 lo = values[index].item()20 if not remainder:21 return lo, Q.NONE22 hi = values[index + 1].item()23 integer = type(lo) is int24 if integer and (int(float(lo)) != lo or int(float(hi)) != hi):25 return None, Q.PRECISION_LOSS26 # 排序和选择均在列上完成。只对这两个端点用整数比计算加权分子,避免27 # max_float - (-max_float)溢出;一次除法给出最终的float64舍入结果。28 a, b = lo.as_integer_ratio()29 c, d = hi.as_integer_ratio()30 numerator = a * d * (denominator - remainder) + c * b * remainder31 divisor = b * d * denominator32 value = numerator / divisor33 if (value == 0 and numerator) or (integer and numerator % divisor == 0 and int(value) != numerator // divisor):34 return None, Q.PRECISION_LOSS35 return value, Q.NONE36 finally:37 # 费用由外层释放;本层回溯只保留错误位置,不继续占用排序副本。38 # chunks 是借用输入,只断开本层参数,不修改或清空调用者的数据。39 chunks = values = None40