selfrelease
74e35fed96
- 新增场景基类、配置加载、注册表与原语模块 - 添加 r10_refund_split 规则及场景 JSON Schema - 扩展 scan 引擎与 scenarios API - 新增场景注册表/配置/集成测试 - 更新前端 App、api、labels 支持新场景
211 lines
7.2 KiB
Python
211 lines
7.2 KiB
Python
"""检测原语库（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},
|
||
)
|