chronaxis.core.quantiles
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""保存精确样本分位数的有限预算状态，结束时定位两个顺序统计值。23q为0或1时只维护原dtype的极值。其余q保留被选中的真实样本，增长前向4本次预算付款，排序临时空间另行预付并释放。标量及数组后端各自完成排序5和线性插值；本对象只共享寿命、秩、预算协议，不读取来源或生成临时文件。6"""78from fractions import Fraction9from math import isfinite10from .types import QualityFlag as Q, SignalDType as D111213SAMPLE_QUANTILES = ('sample_quantile', 'sample_median')141516def quantile_probability(operation, q):17 """校验操作与概率的固定组合，返回精确的配置概率。1819 参数: operation为sample_quantile/sample_median，q为显式数值或None。20 返回: Fraction；中位数固定为1/2，其他概率按配置值的十进制表示解释。21 异常: ValueError为多余、缺失、非有限、越界或布尔概率。22 """23 if operation == 'sample_median':24 if q is not None:25 raise ValueError('sample_median固定使用q=0.5，请省略q')26 return Fraction(1, 2)27 if (operation != 'sample_quantile' or type(q) not in (int, float) or not 0 <= q <= 128 or (type(q) is float and not isfinite(q))):29 raise ValueError('sample_quantile需要有限数值q，范围为0到1')30 return Fraction(str(q))313233class ExactQuantileBuffer:34 """保存单个统计范围的已选样本或端点极值；只持有自己预付的空间。"""3536 def __init__(self, operation, q, dtype, backend, budget=None):37 """配置确定的概率、输入dtype和真实后端，创建空存储。3839 参数: operation/q见概率检查，dtype为数值声明，backend为reference或numpy，40 budget为共享ResourceBudget；独立调用省略时使用项目默认配额。41 返回: 新缓冲。异常: 非法参数在读取样本前失败。42 """43 from ..results.budget import ResourceBudget44 if backend not in ('reference','numpy') or dtype not in (D.INT64,D.UINT64,D.FLOAT64):45 raise ValueError('精确分位数需要数值dtype及已支持后端')46 self._q = quantile_probability(operation, q)47 self._dtype, self._backend = dtype, backend48 self._budget = ResourceBudget() if budget is None else budget49 self._held_cost = self._buffer_cost = self._sort_cost = self._size = 050 self._chunks = [[]] if backend == 'reference' else []51 self._extreme = None52 self._failed = self._closed = self._closing = False5354 def _reserve(self, amount, slot):55 """在保存样本或创建排序副本之前申请空间，保留具体预算错误。5657 参数: amount为本次额外字节，slot为持有本笔费用的字段。58 返回: None。异常: BudgetExceededError或取消原样传播，close可重试清理。59 """60 from ..results.budget import _reserve_owned_ram61 from ..results.errors import BudgetExceededError62 try:63 _reserve_owned_ram(self, amount, slot)64 except BudgetExceededError as error:65 error.add_note('精确样本分位数的样本存储或排序空间超过可用RAM')66 raise6768 def _append_memory_bytes(self, count):69 """给分块调用者计算下一批保留样本的新增费用，不提前占用预算。7071 参数: count为内部已确定的非负样本数，也可为本批可能选中数量的上界。72 返回: 新增字节数；空批和只保留极值的q端点为0。73 异常: 无；实际追加仍通过_reserve检查预算及取消。74 """75 # 与append使用同一个费用来源，避免分块预估与真正付款不一致。76 if not count or self._q in (0, 1):77 return 078 if self._backend == 'reference':79 return (0 if self._size else 256) + count * 6480 return 256 + count * 88182 def append(self, values, mask=None):83 """累加一批真正选中的有效样本；空批直接返回。8485 参数: reference的values为已选原生数值序列；numpy的values为原dtype数组，86 mask为同长布尔选择列。输入所有权属于调用方。87 返回: None。异常: 值/掩码不符、预算不足、关闭后调用或分配失败。88 """89 if self._failed or self._closed or self._closing:90 from ..results.errors import ResultClosedError91 raise ResultClosedError('精确分位数缓冲已关闭或失败')92 try:93 if self._backend == 'reference':94 if mask is not None:95 raise ValueError('标量分位数仅接已选样本')96 count = len(values)97 else:98 import numpy as np99 if mask is None or mask.dtype != np.bool_ or values.shape != mask.shape:100 raise ValueError('数组分位数需要同长布尔掩码')101 count = int(np.count_nonzero(mask))102 if not count:103 return104 if self._q in (0, 1):105 # 端点不需要历史。where+原dtype上限/下限使数组极值不用复制所选列。106 if self._backend == 'reference':107 candidate = (min if self._q == 0 else max)(values)108 else:109 from .numeric import INTEGER_BOUNDS110 lower, upper = INTEGER_BOUNDS.get(self._dtype, (-float('inf'), float('inf')))111 candidate = (np.min(values, where=mask, initial=upper) if self._q == 0 else112 np.max(values, where=mask, initial=lower)).item()113 choose = min if self._q == 0 else max114 self._extreme = candidate if self._extreme is None else choose(self._extreme,candidate)115 else:116 # 预付块头/索引槽、标量对象保留费或数组nbytes，然后取得独占副本。117 if self._backend == 'reference':118 self._reserve(self._append_memory_bytes(count), '_buffer_cost')119 self._chunks[0].extend(values)120 else:121 self._reserve(self._append_memory_bytes(count), '_buffer_cost')122 self._chunks.append(values[mask])123 self._size += count124 except BaseException:125 self._failed = True126 raise127128 def finish(self):129 """读取截至当前批次的分位数快照，不按块分别求分位数再合并。130131 参数: 无。返回: (原生数值或None,质量)。空集合返回None及NONE，最终132 空/部分状态由具有完整观测范围的计算器判定。133 异常: 已关闭/失败时拒绝，排序额度不足或分配失败原样传播。134 """135 from ..results.budget import _release_owned_ram136 if self._failed or self._closed or self._closing:137 from ..results.errors import ResultClosedError138 raise ResultClosedError('精确分位数缓冲已关闭或失败')139 if not self._size or self._q in (0, 1):140 return self._extreme, Q.NONE141 if self._size == 1:142 value = self._chunks[0][0]143 return (value if self._backend == 'reference' else value.item()), Q.NONE144 failure = None145 try:146 # Python排序预付指针列表及排序缓冲。NumPy单一连续副本原地堆排序；147 # 固定项还覆盖列表/数组头与两个端点精确插值所需的临时数值。148 workspace = (32 * self._size if self._backend == 'reference' else149 8 * self._size + 16 * len(self._chunks))150 self._reserve(8192 + workspace, '_sort_cost')151 rank = (self._size - 1) * self._q152 index, remainder = divmod(rank.numerator, rank.denominator)153 if self._backend == 'reference':154 from .quantile_reference import ordered_reference155 return ordered_reference(self._chunks,index,remainder,rank.denominator)156 from .quantile_numpy import ordered_numpy157 return ordered_numpy(self._chunks,index,remainder,rank.denominator)158 except BaseException as error:159 failure = error160 self._failed = True161 raise162 finally:163 try:164 _release_owned_ram(self,self._sort_cost,'_sort_cost')165 except BaseException as cleanup:166 # 退款失败不能遮住先发生的排序/申请错误；实际费用槽由预算助手更新。167 self._failed = True168 original = cleanup if failure is None else failure169 if self._held_cost:170 original.quantile_cleanup_owner = self171 if failure is None:172 raise173 failure.add_note('分位数排序空间释放失败: ' + type(cleanup).__name__)174175 def close(self):176 """放弃样本并退回本实例的全部收费，释放被取消时可再次调用。177178 参数: 无。返回: None。异常: 原预算释放异常传播，不改变其他owner的费用。179 """180 from ..results.budget import _release_owned_ram181 if self._closed:182 return183 self._closing = True184 self._chunks.clear()185 self._extreme = None186 try:187 _release_owned_ram(self,self._sort_cost,'_sort_cost')188 _release_owned_ram(self,self._buffer_cost,'_buffer_cost')189 finally:190 self._closed = self._held_cost == 0191192193def compute_quantile_series(series, spec, sample_kernel, budget):194 """把有时间轴的单块内存输入送入对应后端的精确样本归约器。195196 参数: series为持有值、真实更新及覆盖；spec为quantile/median；sample_kernel197 为该后端的归约器类；budget为空时使用默认配额。198 返回: 包含观测范围、诊断及精确秩结果的CalculationResult。199 异常: 配置、预算或计算器失败传播；正常/异常退出均释放临时所有权。200 """201 from .calculations import make_calculation_result202 from ..results.budget import _cleanup_owner203 reducer = sample_kernel(spec,series.unit,series.values[0].dtype,quantile_budget=budget)204 try:205 facts = reducer.push(series)206 number = reducer.finish()207 actual = facts.actual_window208 span = 0 if actual is None else actual.end_ns - actual.start_ns209 result = make_calculation_result(spec, series, number.value, number.unit, actual,210 facts.value_valid_duration_ns, facts.condition_known_duration_ns, facts.selected_duration_ns,211 facts.selected_value_unknown_duration_ns, facts.valid_sample_count, facts.value_unknown_ranges,212 facts.condition_unknown_ranges, facts.quality | number.quality,213 all_false=actual == spec.window and span > 0 and facts.false_duration_ns == span,214 calculation_failed=number.calculation_failed)215 # 快照已独立，不把排序缓冲带出低层便捷入口。216 reducer.close()217 return result218 except BaseException as error:219 _cleanup_owner(reducer._quantile,error,'quantile_cleanup_owner','内存分位数')220 raise221