规则引擎 `hscredit.core.rules`

规则表达式体系：Rule 支持任意层级嵌套与与/或/非逻辑运算，配套表达式解析、变量提取、美化与优化工具。规则命中指标统一通过 Rule.report 计算。

规则引擎模块.

提供规则定义、评估和报告功能，支持使用 pandas eval 语法的规则表达式编写与组合。

子模块:

rule: 规则定义、评估（Rule类）和报告生成
expr_optimizer: 规则表达式优化与美化

参考样例

>>> from hscredit.core.rules import Rule, get_columns_from_query
>>> rule = Rule("age > 18 and income > 5000", name="优质客群规则")
>>> cols = get_columns_from_query("age > 18 and income < 5000")
>>> print(cols)

class hscredit.core.rules.Rule(expr, name=None, description='', weight=1.0, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]

基类：ParallelizableMixin

规则类。

支持使用 pandas eval 语法的规则定义和评估，支持 &（与）、|（或）、 ~（非）、^（异或）等运算符组合多个规则为复合规则。

属性

参数:

expr (str) -- 规则表达式字符串
name (str | None) -- 规则名称，用于标识和展示，默认为None（使用表达式作为名称）
description (str) -- 规则描述，默认为空字符串
weight (float) -- 规则权重，用于规则集分类器，默认为1.0
n_jobs (int | float | None) -- 并行任务数，默认为-1
parallel_backend (str | None) -- joblib并行后端，默认为None
parallel_config (Mapping[str, Any] | None) -- joblib扩展配置，默认为None

变量:

feature_names_in_ -- 从表达式中解析出的特征名列表
result_ -- 最近一次 predict 的结果 Series
_state -- 当前规则状态（initialized/applied）

参考样例

>>> from hscredit.core.rules import Rule
>>> import pandas as pd
>>> df = pd.DataFrame({'age': [20, 30, 40], 'income': [3000, 8000, 12000]})
>>> rule1 = Rule("age > 18", name="成年规则", description="判断用户是否成年")
>>> rule2 = Rule("income > 5000", name="高收入规则")
>>> # 规则组合
>>> combined = rule1 & rule2
>>> # 应用规则
>>> result = combined.predict(df)
>>> print(result)

filter(X)[源代码]

根据规则过滤数据。

应用规则后返回满足条件（命中）的数据子集。

参数

参数:

X (DataFrame) -- 输入数据 DataFrame

返回:

满足规则的数据子集 DataFrame

抛出:

InputTypeError -- X 不是 DataFrame 时
FeatureNotFoundError -- X 缺少规则表达式所需的列时

返回类型:

DataFrame

参考样例

>>> from hscredit.core.rules import Rule
>>> import pandas as pd
>>> df = pd.DataFrame({'age': [20, 30, 40], 'name': ['A', 'B', 'C']})
>>> rule = Rule("age > 25")
>>> rule.filter(df)

predict(X)[源代码]

应用规则进行预测。

使用 pandas eval 对 DataFrame 执行规则表达式，返回命中的布尔 Series。

参数

参数:

X (DataFrame) -- 输入数据 DataFrame（必须包含规则表达式中引用的全部列）

返回:

规则匹配结果 Series（布尔类型，True表示命中）

抛出:

InputTypeError -- X 不是 DataFrame 时
FeatureNotFoundError -- X 缺少规则表达式所需的列时

返回类型:

Series

参考样例

>>> from hscredit.core.rules import Rule
>>> import pandas as pd
>>> df = pd.DataFrame({'age': [20, 30, 40], 'income': [3000, 8000, 12000]})
>>> rule = Rule("age > 25 and income > 5000")
>>> rule.predict(df)

report(datasets, target='target', overdue=None, dpds=None, del_grey=False, desc='', filter_cols=None, prior_rules=None, amount=None, margins=False, **kwargs)[源代码]

规则效果报告表格输出。

将规则命中与否作为二分类，对数据集计算统计指标，包括样本数、坏账率、LIFT值、风险拒绝比、精确率、召回率、F1分数等。支持金额口径分析与多标签（不同逾期天数定义）联合输出。

参数

参数:

datasets (DataFrame) -- 数据集 DataFrame，需要包含目标变量列或逾期天数列
target (str) -- 目标变量列名，默认为"target"，0=好样本，1=坏样本
overdue (str | List[str] | None) -- 逾期天数字段名（可选，传入时以逾期天数>DPD定义坏样本，支持多标签多DPD联合分析）
dpds (int | List[int] | None) -- 逾期定义方式，逾期天数 > DPD 为坏样本，默认为0；传入列表时支持多DPD联合分析
del_grey (bool) -- 是否删除逾期天数在(0, DPD]区间内的灰度样本，默认为False
desc (str) -- 规则描述，用于报告的"指标含义"列，默认为空字符串
filter_cols (List[str] | None) -- 指定返回的字段列表（可选）
prior_rules (Rule | None) -- 先验规则（可选），先对数据应用先验规则排除部分样本，再对当前规则进行评估
amount (str | None) -- 金额字段名（可选），传入时以金额口径而非样本数口径进行统计
margins (bool) -- 是否在报告末尾添加合计行，默认为False

返回:

规则效果评估表DataFrame。单标签时返回单层列结构，多标签时返回多层列结构（MultiIndex）；列包括：规则分类、指标名称、指标含义、分箱、样本总数、样本占比、好样本数、好样本占比、坏样本数、坏样本占比、坏账率、LIFT值、坏账改善、风险拒绝比、准确率、精确率、召回率、F1分数

抛出:

FeatureNotFoundError -- 数据集缺少规则表达式所需的列时
KeyError -- overdue字段在数据集中不存在时

返回类型:

DataFrame

参考样例

>>> from hscredit.core.rules import Rule
>>> import pandas as pd
>>> df = pd.DataFrame({
...     'age': [20, 30, 40, 50],
...     'income': [3000, 8000, 12000, 5000],
...     'target': [0, 1, 1, 0]
... })
>>> rule = Rule("age > 25 and income > 5000")
>>> report = rule.report(df, target='target')
>>> print(report)

result()[源代码]

获取规则预测结果。

返回最近一次调用 predict() 的结果。必须先调用 predict() 才能使用此方法。

返回:: 最近一次预测的布尔 Series
抛出:: RuleUnAppliedError -- 尚未调用 predict() 时

参考样例

>>> from hscredit.core.rules import Rule
>>> import pandas as pd
>>> df = pd.DataFrame({'age': [20, 30, 40]})
>>> rule = Rule("age > 25")
>>> rule.predict(df)
>>> rule.result()

static save(report, excel_writer, sheet_name=None, excel_params=None)[源代码]

保存规则报告到 Excel。

参数

参数:

report (DataFrame) -- 规则报告 DataFrame（由 report() 方法生成）
excel_writer (str | PathLike | ExcelWriter) -- Excel 文件路径（字符串或 PathLike）或 ExcelWriter 对象；传入路径时会自动创建并写入后关闭，传入对象时追加写入不关闭
sheet_name (str | None) -- 工作表名称，默认为None（使用默认名称Sheet1）
excel_params (Dict | None) -- 额外的 dataframe2excel 写入参数（可选）

返回:

ExcelWriter 对象

抛出:

TypeError -- excel_writer 类型不正确时

返回类型:

ExcelWriter

参考样例

>>> from hscredit.core.rules import Rule
>>> import pandas as pd
>>> df = pd.DataFrame({'age': [20, 30], 'income': [3000, 8000], 'target': [0, 1]})
>>> rule = Rule("age > 25")
>>> report = rule.report(df)
>>> writer = Rule.save(report, "rule_report.xlsx", sheet_name="规则报告")

class hscredit.core.rules.RuleFlow(rules, mode='serial', name=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]

基类：ParallelizableMixin

生产规则流转校验器。

按给定规则顺序计算每笔样本的规则命中结果，支持串行和并行两种执行模式。

参数

参数:

rules (Rule | Sequence[Rule]) -- 单条 Rule 或规则列表，规则顺序即生产执行顺序
mode (str) -- 执行模式，"serial"/"串行" 表示命中第一条规则后停止， "parallel"/"并行" 表示所有规则都参与判断
name (str | None) -- 规则流名称，默认 "RuleFlow"
n_jobs (int | float | None) -- 并行任务数，默认为-1
parallel_backend (str | None) -- joblib并行后端，默认为None
parallel_config (Mapping[str, Any] | None) -- joblib扩展配置，默认为None

属性

变量:

rules -- 规则列表
mode -- 标准化后的执行模式，取值为 "serial" 或 "parallel"
feature_names_in_ -- 全部规则表达式引用的字段列表

参数:

rules (Rule | Sequence[Rule])
mode (str)
name (str | None)
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)

参考样例

>>> from hscredit.core.rules import Rule, RuleFlow
>>> flow = RuleFlow([Rule("score < 500", name="低分"), Rule("cnt > 5", name="多头")])
>>> result = flow.predict(data)
>>> report = flow.report(data, date_col="放款时间", freq="M", group_cols="商品类别")
>>> summary = flow.summary(data, group_cols="商品类别")

compare(data, production_hits, hit_col=None, order_id_col=None, include_data=True)[源代码]

对比线上生产命中结果与线下规则执行结果。

production_hits 支持三种格式：

单列命中规则：每行是 rule.name、[rule.name]、空值或多个规则名列表
指定 hit_col：从 DataFrame 的该列读取命中规则
规则命中矩阵：列名为规则名称/表达式/展示名，值为 0/1 或 True/False

参数:

data (DataFrame) -- 生产订单原始字段数据 DataFrame
production_hits (DataFrame | Series) -- 每笔订单线上生产命中规则
hit_col (str | None) -- 命中规则列名；不传时优先识别 "命中规则"，再识别规则命中矩阵
order_id_col (str | None) -- 订单 ID 字段；传入时按订单 ID 对齐，否则按行顺序/索引对齐
include_data (bool) -- 差异明细是否拼接原始订单字段，默认为 True

返回:

(report, diff_detail)。report 为规则一致性报告，diff_detail 为差异订单明细

返回类型:

Tuple[DataFrame, DataFrame]

predict(data)[源代码]

计算每笔样本在每条规则上的命中结果。

串行模式下，前序规则已命中的样本不会继续流转到后续规则，后续规则列填充 <NA>；并行模式下，每条规则都会对全量样本计算命中结果。返回结果保留输入数据索引。

参数:: data (DataFrame) -- 生产拼表后的变量取值明细
返回:: 每条规则命中明细以及规则集汇总列
返回类型:: DataFrame

report(data, date_col=None, freq='M', group_cols=None, dropna=True)[源代码]

输出每条规则的流转命中报表。

参数:

data (DataFrame) -- 生产拼表后的变量取值明细
date_col (str | None) -- 日期列名，传入后按 freq 生成 统计周期 分组
freq (str) -- pandas Period 频率，默认 "M" 月
group_cols (str | Sequence[str] | None) -- 类别分组字段，支持单列或多列；可与 date_col 同时使用
dropna (bool) -- 是否丢弃分组字段缺失样本，默认为 True；False 时归入 "缺失"

返回:

规则级流转命中报表

返回类型:

DataFrame

summary(data, date_col=None, freq='M', group_cols=None, dropna=True)[源代码]

输出规则集整体通过与命中汇总。

分组参数与 report() 一致；未传入分组时返回一行整体汇总。

参数:

data (DataFrame)
date_col (str | None)
freq (str)
group_cols (str | Sequence[str] | None)
dropna (bool)

返回类型:

DataFrame

hscredit.core.rules.get_columns_from_query(query_str)[源代码]

获取 pandas query 语句使用的列。

解析 query 语法树，提取其中涉及的全部列名，返回去重排序后的列表。

参数

参数:: query_str (str) -- pandas query 支持的查询语句，如 "age > 18 and income < 5000"
返回:: query 语句使用的列名列表（去重后按字母排序）
返回类型:: List[str]

参考样例

>>> from hscredit.core.rules import get_columns_from_query
>>> get_columns_from_query("age > 18 and income < 5000")
['age', 'income']
>>> get_columns_from_query("salary >= 3000 & age.between(20, 60)")
['age', 'salary']
>>> get_columns_from_query("`衡枢鉴真分老客版` < 600 & `逾期(天)` > 7")
['衡枢鉴真分老客版', '逾期(天)']

hscredit.core.rules.optimize_expr(expr)[源代码]

简化规则表达式字符串。

解析表达式为表达式树后，应用以下布尔代数定律做等价化简，并去除冗余括号、将 and/or 统一为 &/|：

幂等律（Idempotent）：A & A → A，A | A → A
吸收律（Absorption）：A | (A & B) → A，A & (A | B) → A
双重否定（Double negation）：~~A → A

备注

化简基于子表达式字符串的规范化比较，仅识别字面等价的原子条件，不做跨变量的逻辑推理（如 age > 18 与 age >= 19 不会被判定为等价）。

参数:: expr (str) -- 原始规则表达式字符串，支持 &/|/~/and/or/not
返回:: 化简后的等价表达式字符串
返回类型:: str

参考样例

>>> optimize_expr("(age > 18) & (age > 18)")    # 幂等律
'age > 18'
>>> optimize_expr("~~(age > 18)")               # 双重否定
'age > 18'

引用

布尔代数化简定律：https://en.wikipedia.org/wiki/Boolean_algebra#Laws

hscredit.core.rules.beautify_expr(expr)[源代码]

美化规则表达式字符串。

在不改变逻辑的前提下规范化表达式的书写：将 and/or/not 统一为 &/|/~ 符号形式，按运算符结合律去除同级冗余括号，得到格式一致、便于展示与比较的表达式。与 optimize_expr() 的区别在于不做幂等/吸收等化简。

参数:: expr (str) -- 原始规则表达式字符串
返回:: 美化后的等价表达式字符串
返回类型:: str

参考样例

>>> beautify_expr("(age > 18) & (income > 5000)")
'age > 18 & income > 5000'
>>> beautify_expr("age > 18 and income > 5000")
'age > 18 & income > 5000'

hscredit.core.rules.get_expr_variables(expr)[源代码]

提取规则表达式中引用的变量（列）名。

用正则匹配表达式中的标识符，剔除 and/or/not/True/False/ None/inf/nan 等保留字，返回去重后的变量名列表。

备注

返回顺序 **不保证稳定**（基于集合去重）。如需有序且能正确处理含空格/中文/ 反引号的列名，请使用 hscredit.core.rules.get_columns_from_query() （返回去重并按字母排序的列表）。

参数:: expr (str) -- 规则表达式字符串
返回:: 表达式引用的变量名列表（去重，顺序不保证）
返回类型:: List[str]

参考样例

>>> sorted(get_expr_variables("(age > 18) & (income > 5000)"))
['age', 'income']

class hscredit.core.rules.RuleState(value)[源代码]

基类：str, Enum

规则生命周期状态枚举。

继承 str，可直接与字符串比较。Rule 用它标记是否已执行过 predict，从而约束 Rule.result() 等依赖结果的方法。

枚举值

INITIALIZED ("initialized")：规则已创建但尚未调用 Rule.predict()，此时无可用结果
APPLIED ("applied")：规则已对某数据集执行过 Rule.predict()， result_ 中存有最近一次命中结果

INITIALIZED = 'initialized'

APPLIED = 'applied'

exception hscredit.core.rules.RuleStateError[源代码]

基类：StateError

规则状态异常基类。

当在规则不允许的状态下调用方法时抛出，继承自 StateError。

exception hscredit.core.rules.RuleUnAppliedError[源代码]

基类：RuleStateError

规则尚未应用异常。

在未先调用 Rule.predict() 的情况下访问 Rule.result() 等依赖预测结果的方法时抛出。

规则引擎 hscredit.core.rules

规则引擎 `hscredit.core.rules`