"""检测原语库（B 快捷通道）：可被 YAML 配置场景复用的通用检测/评分模式。 每个原语接收数据行（list[dict]）与参数（dict），返回统一的 PrimitiveResult。 新增"同构"场景时，可在 YAML 中通过 detector 引用原语、params 传参，无需编写 Python。 若遇到原语无法表达的全新算法，则回到 A 方案实现一个 BaseScenario 子类（或新增原语）。 """ from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass, field # 原语统一签名：(rows, params) -> PrimitiveResult PrimitiveFn = Callable[[list[dict], dict], "PrimitiveResult"] @dataclass class PrimitiveResult: """原语输出：是否命中、评分（0-1）、证据、主体、可读简述与可插值指标。""" hit: bool score: float evidence: dict = field(default_factory=dict) subjects: dict = field(default_factory=dict) summary: str = "" # 供 rationale_template 的 {summary} 占位 metrics: dict = field(default_factory=dict) # 供 rationale_template 的其它占位 # 全局原语注册表：name -> 原语函数 PRIMITIVE_REGISTRY: dict[str, PrimitiveFn] = {} def register_primitive(name: str) -> Callable[[PrimitiveFn], PrimitiveFn]: """函数装饰器：把检测原语注册到全局表。""" def deco(fn: PrimitiveFn) -> PrimitiveFn: PRIMITIVE_REGISTRY[name] = fn return fn return deco def get_primitive(name: str) -> PrimitiveFn | None: """按名称查找已注册的检测原语。""" return PRIMITIVE_REGISTRY.get(name) def _to_float(value: object, default: float = 0.0) -> float: try: return float(value) # type: ignore[arg-type] except (TypeError, ValueError): return default @register_primitive("threshold_edge") def threshold_edge(rows: list[dict], params: dict) -> PrimitiveResult: """金额集中在审批阈值边缘 [edge_ratio*threshold, threshold) 的拆单型模式。 params: amount_field (str): 金额字段名 threshold (float): 审批阈值（>0） edge_ratio (float): 边缘下界比例，默认 0.8 min_count (int): 触发命中的最小记录数，默认 3 key_field (str, 可选): 主体字段名（用于汇总涉及主体） """ amount_field = params["amount_field"] threshold = _to_float(params.get("threshold")) edge_ratio = _to_float(params.get("edge_ratio", 0.8), 0.8) min_count = int(params.get("min_count", 3)) if threshold <= 0: raise ValueError("threshold 必须为正数") lower = edge_ratio * threshold near = [r for r in rows if lower <= _to_float(r.get(amount_field)) < threshold] ratio = (len(near) / len(rows)) if rows else 0.0 total = sum(_to_float(r.get(amount_field)) for r in near) hit = len(near) >= min_count score = 0.0 if hit: score = min(0.6, 0.1 * len(near)) + 0.2 * ratio score = round(min(score, 1.0), 3) subjects: dict = {} key_field = params.get("key_field") if key_field: subjects = { "keys": sorted( {str(r.get(key_field)) for r in near if r.get(key_field) is not None} ) } return PrimitiveResult( hit=hit, score=score, evidence={ "near_count": len(near), "edge_ratio": round(ratio, 3), "near_amount": total, "threshold": threshold, }, subjects=subjects, summary=( f"检测到 {len(near)} 条记录金额集中在审批阈值 {threshold:.0f} 的边缘区间" f"（占比 {ratio:.0%}），合计约 {total:.0f}" ), metrics={ "near_count": len(near), "ratio": round(ratio, 3), "near_amount": total, "threshold": threshold, }, ) @register_primitive("cliff_drop") def cliff_drop(rows: list[dict], params: dict) -> PrimitiveResult: """时序断崖：按 index 排序后，相邻点 value 骤降超过 cliff_drop 即命中。 params: index_field (str): 序号/时间字段名 value_field (str): 取值字段名（如留存率） cliff_drop (float): 触发断崖的最小跌幅，默认 0.5 """ index_field = params["index_field"] value_field = params["value_field"] cliff = _to_float(params.get("cliff_drop", 0.5), 0.5) ordered = sorted(rows, key=lambda r: _to_float(r.get(index_field))) max_drop = 0.0 cliff_at: object = None for prev, cur in zip(ordered, ordered[1:], strict=False): drop = _to_float(prev.get(value_field)) - _to_float(cur.get(value_field)) if drop > max_drop: max_drop = drop if drop >= cliff: cliff_at = cur.get(index_field) hit = cliff_at is not None max_drop = round(max_drop, 3) score = round(min(0.4 + 0.5 * max_drop, 1.0), 3) if hit else round(0.3 * max_drop, 3) return PrimitiveResult( hit=hit, score=score, evidence={"cliff_at": cliff_at, "max_drop": max_drop}, summary=( f"在 {cliff_at} 处出现断崖式骤降（最大跌幅 {max_drop:.0%}）" if hit else f"未见明显断崖（最大跌幅 {max_drop:.0%}）" ), metrics={"cliff_at": cliff_at, "max_drop": max_drop}, ) @register_primitive("ratio_threshold") def ratio_threshold(rows: list[dict], params: dict) -> PrimitiveResult: """聚合比率/阈值判定：对某字段做聚合并与阈值比较，命中则给定分。 params: value_field (str): 取值字段名 agg (str): 聚合方式 mean/sum/max/min/count，默认 mean op (str): 比较运算 >=,>,<=,<，默认 >= threshold (float): 比较阈值 score (float): 命中评分，默认 0.8 """ value_field = params["value_field"] agg = str(params.get("agg", "mean")) op = str(params.get("op", ">=")) threshold = _to_float(params.get("threshold")) hit_score = _to_float(params.get("score", 0.8), 0.8) values = [_to_float(r.get(value_field)) for r in rows] if not values and agg != "count": return PrimitiveResult(hit=False, score=0.0, summary="无数据，未触发") if agg == "sum": actual = sum(values) elif agg == "max": actual = max(values) elif agg == "min": actual = min(values) elif agg == "count": actual = float(len(values)) else: # mean actual = sum(values) / len(values) comparators = { ">=": lambda a, b: a >= b, ">": lambda a, b: a > b, "<=": lambda a, b: a <= b, "<": lambda a, b: a < b, } cmp = comparators.get(op) if cmp is None: raise ValueError(f"不支持的比较运算: {op}") hit = cmp(actual, threshold) score = round(hit_score, 3) if hit else 0.0 return PrimitiveResult( hit=hit, score=score, evidence={"agg": agg, "actual": round(actual, 3), "op": op, "threshold": threshold}, summary=( f"{value_field} 的{agg}为 {actual:.3g}，{op} 阈值 {threshold:.3g}" if hit else f"{value_field} 的{agg}为 {actual:.3g}，未达阈值 {threshold:.3g}" ), metrics={"actual": round(actual, 3), "threshold": threshold}, )