工具 hscredit.utils

随机种子、pickle IO、数据描述、Pandas 扩展（df.summary() / df.save() / df.show()）、日志等通用工具。import hscredit 时自动注册 Pandas 扩展方法。

工具函数模块.

提供常用的工具函数，包括随机种子设置、数据IO、特征描述、分箱表美化展示、环境初始化、日志管理等。

hscredit.utils.seed_everything(seed, freeze_torch=False)

固定当前环境随机种子，以保证后续实验可重复。

参数:

• seed (int) -- 随机种子

• freeze_torch (bool) -- 是否固定 pytorch 的随机种子

参考样例

>>> seed_everything(42)
>>> seed_everything(42, freeze_torch=True)

hscredit.utils.load_pickle(file, engine='auto', compression=None)

导入 pickle 文件。

支持多种序列化引擎（joblib/dill/cloudpickle/pickle）和压缩格式 （gzip/bz2/xz/lz4/zstd）。支持根据文件扩展名自动检测。

参数:

• file (str | Path) -- pickle 文件路径，支持 .pkl, .pkl.gz, .joblib, .dill 等格式

• engine (str) -- 使用的序列化引擎，可选： - 'auto': 自动检测（根据文件内容和扩展名推断，默认） - 'joblib': 使用 joblib（推荐用于 numpy/scipy/sklearn 对象） - 'dill': 使用 dill（支持 lambda、嵌套函数等复杂对象） - 'cloudpickle': 使用 cloudpickle（常用于分布式计算如 PyTorch/Spark） - 'pickle': 使用标准库 pickle

• compression (str | None) -- 压缩格式，可选： - None: 根据文件扩展名自动检测（.gz/.bz2/.xz/.lz4/.zst） - 'gzip'/'gz': gzip 压缩 - 'bz2': bzip2 压缩 - 'xz': xz/lzma 压缩 - 'lz4': lz4 压缩（需安装 lz4） - 'zstd'/'zstandard': zstd 压缩（需安装 zstandard）

返回:

反序列化后的对象

返回类型:

Any

参考样例

>>> data = load_pickle('model.pkl')
>>> data = load_pickle('model.pkl.gz')
>>> data = load_pickle('model.dill', engine='dill')
>>> data = load_pickle('model.pkl', engine='cloudpickle')
>>> data = load_pickle('model.pkl', compression='gzip')

hscredit.utils.save_pickle(obj, file, engine='joblib', compression=None, compression_level=None, protocol=None)

保存数据至 pickle 文件。

支持多种序列化引擎（joblib/dill/cloudpickle/pickle）和压缩格式 （gzip/bz2/xz/lz4/zstd），可处理大型模型和复杂对象。

参数:

• obj (Any) -- 需要保存的数据对象

• file (str | Path) -- 文件路径，建议扩展名 .pkl, .joblib, .dill 等

• engine (str) -- 使用的序列化引擎，可选： - 'joblib': joblib（默认，推荐用于 numpy/scipy/sklearn 对象） - 'dill': dill（支持 lambda、嵌套函数等复杂对象） - 'cloudpickle': cloudpickle（常用于分布式计算） - 'pickle': 标准库 pickle

• compression (str | None) -- 压缩格式，可选： - None: 不压缩（默认） - 'gzip'/'gz': gzip 压缩（兼容性好） - 'bz2': bzip2 压缩（压缩率高但较慢） - 'xz': xz/lzma 压缩（最高压缩率） - 'lz4': lz4 压缩（速度最快，需安装 lz4） - 'zstd'/'zstandard': zstd 压缩（速度与压缩率平衡，需安装 zstandard） - 'auto': 根据文件扩展名自动选择

• compression_level (int | None) -- 压缩级别（1-9，数字越大压缩率越高，默认取决于压缩算法）

• protocol (int | None) -- pickle 协议版本（默认使用最高可用版本）

返回:

保存的文件路径

返回类型:

str

参考样例

>>> save_pickle(model, 'model.pkl')
>>> save_pickle(lambda_func, 'func.dill', engine='dill')
>>> save_pickle(model, 'model.pkl', engine='cloudpickle')
>>> save_pickle(model, 'model.pkl.gz')
>>> save_pickle(model, 'model.pkl', compression='zstd', compression_level=3)
>>> save_pickle(model, 'model.pkl.xz', compression='xz')

class hscredit.utils.ArtifactSerializableMixin

基类：object

hscredit 完整对象制品序列化混入类.

子类无需实现额外方法即可获得统一持久化能力。制品中同时记录对象类型、 制品类别和协议版本，加载时会校验目标类型，避免误加载其他对象。

artifact_kind = '通用制品'

get_artifact_metadata()

返回不包含对象本体的制品元数据.

返回类型:

Dict[str, Any]

classmethod load_artifact(file, engine='auto', compression=None, **kwargs)

加载并校验完整 hscredit 对象.

为兼容旧文件，也接受直接保存、未包装制品元数据的对象。

参数:

• file (str | Path)

• engine (str)

• compression (str | None)

返回类型:

T

save_artifact(file, engine='joblib', compression=None, **kwargs)

保存完整 hscredit 对象.

参数:

• file (str | Path) -- 输出文件路径

• engine (str) -- joblib、pickle、dill 或 cloudpickle

• compression (str | None) -- 可选压缩格式

• kwargs -- 传递给 hscredit.utils.save_pickle()

返回:

保存后的文件路径

返回类型:

str

class hscredit.utils.ParallelBudget(available, depth)

基类：object

并行执行上下文中的可用预算。

参数

available: 当前调用可使用的最大并行预算。 depth: 当前调用所在的嵌套深度。

属性

available: 正整数并行预算。 depth: 非负嵌套深度。

参考样例

ParallelBudget(available=4, depth=1) 表示第一层 worker 可使用 4 个工作预算。

参数:

• available (int)

• depth (int)

available: int

depth: int

class hscredit.utils.ParallelExecutionPlan(requested_workers, workers, backend, adaptive, estimated_work, data_bytes, child_budget, operation)

基类：object

一次并行调用的只读有效执行计划。

参数:

• requested_workers (int)

• workers (int)

• backend (str | None)

• adaptive (bool)

• estimated_work (float)

• data_bytes (int)

• child_budget (int)

• operation (str)

requested_workers: int

workers: int

backend: str | None

adaptive: bool

estimated_work: float

data_bytes: int

child_budget: int

operation: str

class hscredit.utils.ParallelWorkload(task_count, rows=1, columns=1, data_bytes=0, cost_per_item=1.0, capability='process_safe', releases_gil=False, has_parallel_children=False, auto_max_workers=None, operation='批量任务')

基类：object

描述一次批量运算的规模和安全执行能力。

参数:

• task_count (int)

• rows (int)

• columns (int)

• data_bytes (int)

• cost_per_item (float)

• capability (str)

• releases_gil (bool)

• has_parallel_children (bool)

• auto_max_workers (int | None)

• operation (str)

auto_max_workers: int | None = None

capability: str = 'process_safe'

columns: int = 1

cost_per_item: float = 1.0

data_bytes: int = 0

property estimated_work: float

返回不依赖运行时计时的确定性工作量估计。

has_parallel_children: bool = False

operation: str = '批量任务'

releases_gil: bool = False

rows: int = 1

task_count: int

class hscredit.utils.ParallelizableMixin

基类：object

为估计器提供统一并行执行入口的内部混入类。

n_jobs: int | float | None

parallel_backend: str | None

parallel_config: Mapping[str, Any] | None

hscredit.utils.parallel_execute(function, tasks, *, n_jobs=-1, parallel_backend=None, parallel_config=None, task_labels=None, default_backend=None, has_parallel_children=False, workload=None, preserve_exceptions=False)

按提交顺序执行任务，并在线程和进程间传播并行预算。

参数:

• function (Callable[[Task], Result])

• tasks (Iterable[Task])

• n_jobs (int | float | None)

• parallel_backend (str | None)

• parallel_config (Mapping[str, Any] | None)

• task_labels (Iterable[Any] | None)

• default_backend (str | None)

• has_parallel_children (bool)

• workload (ParallelWorkload | None)

• preserve_exceptions (bool)

返回类型:

List[Result]

hscredit.utils.plan_parallel_execution(n_jobs, workload, *, parallel_backend=None, parallel_config=None, default_backend=None, cpu_count=None, available_budget=None)

根据用户预算和工作负载生成确定性的有效执行计划。

参数:

• n_jobs (int | float | None)

• workload (ParallelWorkload)

• parallel_backend (str | None)

• parallel_config (Mapping[str, Any] | None)

• default_backend (str | None)

• cpu_count (int | None)

• available_budget (int | None)

返回类型:

ParallelExecutionPlan

hscredit.utils.resolve_n_jobs(n_jobs, task_count=None, *, cpu_count=None, available_budget=None)

解析并行工作数。

-1 大约使用物理 CPU 的 80%，并在多核环境中保留一个 CPU。 正整数表示固定工作数，0 到 1 之间的小数表示物理 CPU 比例。

参数:

• n_jobs (int | float | None)

• task_count (int | None)

• cpu_count (int | None)

• available_budget (int | None)

返回类型:

int | None

hscredit.utils.resolve_native_workers(n_jobs, native_workers=None)

统一解析 thread_count、num_workers 等原生线程参数。

参数:

• n_jobs (int | float | None)

• native_workers (int | None)

返回类型:

int

hscredit.utils.get_physical_cpu_count()

返回可用的物理 CPU 数，无法识别时使用保守回退值。

返回类型:

int

hscredit.utils.split_parallel_budget(available, task_count, has_parallel_children)

计算当前层和真实并行子层的工作预算。

参数:

• available (int)

• task_count (int)

• has_parallel_children (bool)

返回类型:

Tuple[int, int]

hscredit.utils.validate_parallel_config(parallel_backend, parallel_config)

校验 joblib 并行配置并返回独立的配置字典。

工作数和后端由公共参数统一管理，不能在 parallel_config 中重复声明。 backend_kwargs 用于承载后端专属参数，且会复制为独立字典。

参数:

• parallel_backend (str | None)

• parallel_config (Mapping[str, Any] | None)

返回类型:

Dict[str, Any]

hscredit.utils.feature_describe(data, feature=None, percentiles=None, missing=None, cardinality=None)

特征描述统计（数值型与类别型自动区分）。

对数值型特征输出 样本数/非空数/查得率/最小值/平均值/各分位数/最大值； 对类别型特征（或唯一值数 ≤ cardinality）输出 样本数/非空数/查得率 及各取值占比。

参数:

• data (DataFrame) -- 输入数据，DataFrame``（需配合 ``feature 指定列）或单列 Series

• feature (str | None) -- 特征列名，仅当 data 为 DataFrame 时需要； 为 None 时把 data 整体当作 Series 处理

• percentiles (List[float] | None) -- 分位数列表（0~1 之间），默认 [0.01, 0.02, 0.03, 0.05, 0.1, 0.2, ..., 0.95, 0.97, 0.98, 0.99]

• missing -- 缺失值标记（标量或列表），统计前会被替换为 np.nan；默认 None 不替换

• cardinality (int | None) -- 基数阈值（正整数），唯一值数 ≤ 该值时强制按类别型统计； 默认 None 表示仅按 dtype 判断

返回:

描述统计结果 Series``（``name 为特征名）

抛出:

ValueError -- feature 不在 data 列中，或 cardinality < 1 时

返回类型:

Series

参考样例

>>> feature_describe(df, feature='age')          # 指定列
>>> feature_describe(df['age'])                  # 直接传 Series
>>> feature_describe(df, feature='city', cardinality=20)   # 强制按类别统计
>>> feature_describe(df, feature='income', missing=-999)   # -999 视为缺失

hscredit.utils.groupby_feature_describe(data, by=None, n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)

按分组进行特征描述统计。

参数:

• data (DataFrame) -- 数据DataFrame

• by -- 分组字段或字段列表

• n_jobs (int) -- 并行任务数，-1 根据数据规模自动选择

• parallel_backend (str | None) -- joblib 后端；默认使用避免复制字符串列的线程后端

• parallel_config (Dict[str, Any] | None) -- joblib 扩展配置

• kwargs -- 传递给feature_describe的其他参数

返回:

描述统计结果DataFrame

返回类型:

DataFrame

参考样例

>>> groupby_feature_describe(df, by='gender')
>>> groupby_feature_describe(df, by=['gender', 'age_group'])

hscredit.utils.germancredit()

加载德国信贷数据集 German Credit Data。

数据来源：https://archive.ics.uci.edu/dataset/144/statlog+german+credit+data

返回:

pd.DataFrame

参考样例

>>> df = germancredit()
>>> print(df.shape)
(1000, 21)

hscredit.utils.round_float(num, decimal=4)

调整数值分箱的上下界小数点精度，如未超出精度保持原样输出。

参数:

• num -- 分箱的上界或者下界

• decimal (int) -- 小数点保留的精度

返回:

精度调整后的数值

参考样例

>>> round_float(3.14159265, decimal=4)
3.1416

hscredit.utils.reload(module_name)

在jupyter中强制重载模块，忽略所有缓存

参数:

module_name -- 模块名称

返回:

重新导入的模块

参考样例

>>> import hscredit.utils.misc
>>> hscredit.utils.misc.reload('hscredit.utils.misc')
<module 'hscredit.utils.misc' from '...'>

hscredit.utils.trapz(y, x=None, dx=1.0, axis=-1)

梯形法则数值积分（跨 NumPy 版本兼容）.

NumPy 2.0 起将 np.trapz 重命名为 np.trapezoid，旧名在 2.0 中弃用、 在更高版本中移除。本函数按版本自动选择可用实现，保证在新旧 NumPy 下行为一致。

参数:

• y -- 被积函数值数组

• x -- 采样点坐标，可选

• dx -- 采样间隔（x 未提供时使用），默认 1.0

• axis -- 积分所沿的轴，默认 -1

返回:

积分结果

参考样例

>>> import numpy as np
>>> from hscredit.utils import trapz
>>> trapz([0, 1, 1], [0, 0.5, 1.0])
0.75

hscredit.utils.init_setting(font_path=None, seed=None, freeze_torch=False, logger=False, **kwargs)

初始化环境配置。

去除警告信息、修改 pandas 默认配置、固定随机种子。

参数:

• font_path -- 画图时图像使用的字体，支持系统已注册字体名称或本地 .ttf 字体文件路径；为 None 时自动安装并使用包内置字体，安装不可用时回退到“楷体”

• seed -- 随机种子，默认为 None（不固定）。非 None 时调用 seed_everything()

• freeze_torch -- 是否同时固定 PyTorch 随机种子，默认 False（仅 seed 非 None 时生效）

• logger -- 是否返回一个日志器，默认为 False

• kwargs -- 当 logger 为 True 时传给 logging.getLogger 的参数

返回:

当 logger 为 True 时返回 logging.Logger，否则返回 None

注意

本函数在 import hscredit 时被自动调用，会尝试将内置字体安装到当前用户字体目录， 全局执行 warnings.filterwarnings("ignore") 屏蔽所有警告，并修改 pandas/matplotlib 全局配置。 字体安装失败不会阻断导入，系统不存在品牌字体时将回退到“楷体”。

参考样例

>>> from hscredit.utils import init_setting
>>> init_setting()                       # 默认配置（内置中文字体）
>>> init_setting(seed=42)                # 同时固定随机种子
>>> init_setting(font_path='SimHei')     # 指定系统字体
>>> logger = init_setting(logger=True)   # 返回日志器

hscredit.utils.get_bundled_font_path()

返回 hscredit 包内置字体文件路径.

返回类型:

Path

hscredit.utils.install_bundled_font(force=False, system=None)

安装 hscredit 内置字体到当前用户字体库.

参数:

• force (bool) -- 是否强制覆盖已有字体文件

• system (str | None) -- 操作系统名称，仅用于测试或显式覆盖自动识别结果

返回:

(安装路径, 是否实际更新字体文件)

返回类型:

Tuple[Path, bool]

hscredit.utils.init_logger(name='hscredit', level=20, log_file=None, format=None, console=True)

初始化日志记录器。

参数:

• name (str) -- logger 名称，默认为 "hscredit"

• level (int) -- 日志级别，默认为 logging.INFO

• log_file (str) -- 日志文件路径，默认为 None（不写入文件）

• format (str) -- 日志格式，默认为 "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

• console (bool) -- 是否输出到控制台，默认为 True

返回:

配置好的 logger 对象

返回类型:

Logger

参考样例

>>> from hscredit.utils import init_logger
>>> logger = init_logger()                       # 控制台 INFO 日志
>>> logger = init_logger(level=10)               # DEBUG 级别
>>> logger = init_logger(log_file='logs/run.log')  # 同时写入文件
>>> logger.info("模型训练开始")

hscredit.utils.get_logger(name='hscredit')

获取已存在的 logger（logging.getLogger 的轻封装）。

参数:

name (str) -- logger 名称，默认 "hscredit"。应与 init_logger() 创建时使用的名称一致，以复用其 handler 配置

返回:

对应名称的 logging.Logger 对象

返回类型:

Logger

参考样例

>>> from hscredit.utils import init_logger, get_logger
>>> init_logger(name='hscredit')
>>> logger = get_logger('hscredit')   # 在其他模块复用同一 logger

hscredit.utils.style_bin_table(df, max_rows=None, highlight_iv=True, highlight_bad_rate=True, highlight_lift=True, highlight_ks=True, compact=False, precision=None, index_as_bin=False, percent_format=True, high_tech_style=False)

美化分箱表展示.

使用 pandas Styler 对分箱表进行格式化和高亮，使其在 Jupyter 中更易读。

参数:

• df (DataFrame) -- 分箱统计表 DataFrame

• max_rows (int | None) -- 最大显示行数，None 表示显示全部

• highlight_iv (bool) -- 是否高亮 IV 值

• highlight_bad_rate (bool) -- 是否高亮坏样本率（进度条）

• highlight_lift (bool) -- 是否高亮 LIFT 值（进度条）

• highlight_ks (bool) -- 是否高亮 KS 值（进度条）

• compact (bool) -- 是否使用紧凑模式（隐藏部分列）

• precision (Dict[str, int] | None) -- 自定义小数位数，格式为 {'列名': 位数}

• index_as_bin (bool) -- 是否将分箱作为索引显示

• percent_format (bool) -- 是否将百分比相关列显示为百分比格式（默认True）

• high_tech_style (bool) -- 是否使用高科技/AI风格样式

返回:

格式化后的 Styler 对象

返回类型:

Any

参考样例

>>> table = feature_bin_stats(data, 'score', target='target')
>>> style_bin_table(table).show()

hscredit.utils.style_rule_table(df, overall_badrate=None, precision=None)

美化决策树规则表展示。

使用 pandas Styler 对规则表进行格式化、高亮和颜色标注， 使其在 Jupyter 中更易读。

参数:

• df (DataFrame) --

  规则表 DataFrame，兼容两种来源：

  • ManualTreeExtractor.get_rule_table() / report() 的报告格式 （节点编号、是否叶子、指标含义、样本总数、样本占比、坏样本率、LIFT值、…）

  • 旧版精简格式（节点编号、是否叶子、规则表达式、样本数、坏账率、LIFT值、…）

• overall_badrate (float | None) -- 全局坏样本率（用于计算颜色梯度），默认从数据推断

• precision (dict | None) -- 自定义小数位数，格式为 {'列名': 位数}

返回:

格式化后的 Styler 对象

返回类型:

Any

参考样例

>>> from hscredit.report.mining import ManualTreeExtractor
>>> ext = ManualTreeExtractor(target='IS_BAD')
>>> ext.fit(df, features=['age', 'income'])
>>> styled = style_rule_table(ext.get_rule_table())
>>> styled.show()

class hscredit.utils.BinTableDisplay(df)

基类：object

分箱表展示器.

提供链式调用接口，方便在 Jupyter 中展示美观的分箱表。

参考样例

>>> table = feature_bin_stats(data, 'score', target='target')
>>> table.show()
>>> table.show(compact=True)
>>> table.show(highlight_iv=False)

参数:

df (pd.DataFrame)

export_html(filename)

导出为 HTML 文件.

参数:

filename (str) -- 文件名

返回:

self，支持链式调用

返回类型:

BinTableDisplay

highlight_bins(bins, color='#e3f2fd')

高亮指定的分箱行.

参数:

• bins (int | List[int]) -- 要高亮的分箱索引或索引列表

• color (str) -- 高亮颜色

返回:

self，支持链式调用

返回类型:

BinTableDisplay

show(max_rows=None, highlight_iv=True, highlight_bad_rate=True, highlight_lift=True, highlight_ks=True, compact=False, precision=None, index_as_bin=False, percent_format=True, high_tech_style=False, **kwargs)

展示美化的分箱表.

参数:

• max_rows (int | None) -- 最大显示行数

• highlight_iv (bool) -- 是否高亮 IV 值

• highlight_bad_rate (bool) -- 是否高亮坏样本率

• highlight_lift (bool) -- 是否高亮 LIFT 值

• highlight_ks (bool) -- 是否高亮 KS 值

• compact (bool) -- 是否使用紧凑模式

• precision (Dict[str, int] | None) -- 自定义小数位数

• index_as_bin (bool) -- 是否将分箱作为索引显示

• percent_format (bool) -- 是否将百分比相关列显示为百分比格式（默认True）

• high_tech_style (bool) -- 是否使用高科技/AI风格样式

• kwargs -- 其他参数

返回:

self，支持链式调用

返回类型:

BinTableDisplay

to_excel(filename, sheet_name='分箱统计')

导出为 Excel 文件.

参数:

• filename (str) -- 文件名

• sheet_name (str) -- 工作表名称

返回:

self，支持链式调用

返回类型:

BinTableDisplay

hscredit.utils.register_extensions()

注册 pandas DataFrame/Series 扩展方法.

在导入 hscredit 时自动调用，将以下方法添加到 pandas: - df.summary(): 综合特征描述统计 - df.eda_info(): EDA基础信息 - df.missing_analysis(): 缺失值分析 - df.show(): 美化展示分箱表 - df.save(): 保存到Excel - s.summary(): 单字段综合特征描述统计 - s.save(): Series保存到Excel - df/s/groupby.hscredit.apply() / hscredit(...).apply(): 严格单次并行 apply

幂等：重复调用不会重复注册（已存在同名属性时跳过）。导入 hscredit 时已自动执行， 通常无需手动调用。

参考样例

>>> import pandas as pd
>>> import hscredit  # 导入即自动注册扩展
>>>
>>> df = pd.DataFrame({'age': [25, 40], 'target': [0, 1]})
>>> summary = df.summary(y='target')
>>> df.save("report.xlsx")
>>> default_result = df.hscredit.apply(func, axis=1)
>>> result = df.hscredit(n_jobs=-1, bar=False).apply(func, axis=1)

class hscredit.utils.HSCreditApplyProxy(_obj, n_jobs=-1, bar=True, parallel_backend=None, parallel_config=None)

基类：object

保存一次 pandas apply 调用的并行配置。

参数

_obj: DataFrame、Series 或相应的 GroupBy 对象。 n_jobs: 总并行预算，默认 -1 表示自动使用约 80% 的物理核心。 bar: 是否显示按真实完成项累计的进度条。 parallel_backend: 可选的 joblib 后端；未指定时按 callable 能力静态选择。 parallel_config: batch_size、timeout 等统一并行运行参数。

属性

所有配置均只属于当前代理；创建代理不会修改原 pandas 对象。

参考样例

df.hscredit.apply(func, axis=1) df.hscredit(n_jobs=-1, bar=True).apply(func, axis=1)

参数:

• _obj (Any)

• n_jobs (Any)

• bar (bool)

• parallel_backend (str | None)

• parallel_config (Mapping[str, Any] | None)

apply(func, *args, **kwargs)

使用已配置的 hscredit 执行策略调用 pandas apply。

bar: bool = True

n_jobs: Any = -1

parallel_backend: str | None = None

parallel_config: Mapping[str, Any] | None = None

hscredit.utils.create_hscredit_apply_proxy(self, n_jobs=-1, bar=True, parallel_backend=None, parallel_config=None)

为当前 pandas 对象创建独立的 apply 配置代理。

此函数由 pandas 对象的 hscredit 访问器调用。它只保存配置，不读取样本、 不调用用户函数，也不会改变原对象。

参数:

• bar (bool)

• parallel_backend (str | None)

• parallel_config (Mapping[str, Any] | None)

返回类型:

HSCreditApplyProxy

hscredit.utils.check_xy_inputs(X, y=None, target='target', accept_numpy=True)

统一检查并处理 X 和 y 输入（适配双 API 调用风格）。

支持两种 API 风格，并按 传入的 y > 从 X 提取 target 列 的优先级确定目标变量：

• sklearn 风格：fit(X, y)，X、y 分别传入

• scorecardpipeline 风格：fit(df)，目标列包含在 df 中、由 target 指定

参数:

• X (ndarray | DataFrame | Series | List) -- 输入数据，DataFrame 或 numpy 数组（自动转为 DataFrame）

• y (ndarray | Series | List | None) -- 目标变量（可选），提供时优先于从 X 提取 target 列

• target (str) -- 目标列名，当 y 为 None 时从 X 中提取，默认 'target'

• accept_numpy (bool) -- 是否接受 numpy 数组输入，默认 True

返回:

三元组 (X_df, y_series, feature_names)：

• X_df (DataFrame)：处理后的特征数据

• y_series (Series)：目标变量

• feature_names (list)：特征名称列表

抛出:

• InputValidationError -- X 与 y 长度不匹配，或数据为空时

• FeatureNotFoundError -- y 为 None 且 X 中无 target 列时

• InputTypeError -- 输入类型不受支持时

返回类型:

Tuple[DataFrame, Series, List[str]]

参考样例

>>> # sklearn 风格
>>> X_df, y_series, features = check_xy_inputs(X_train, y_train)
>>>
>>> # scorecardpipeline 风格
>>> X_df, y_series, features = check_xy_inputs(df, target='target')

hscredit.utils.convert_to_dataframe(X, columns=None)

将输入转换为 DataFrame。

支持 DataFrame / numpy 数组（1 维或 2 维）/ Series / list / tuple。 1 维输入转为单列 DataFrame；DataFrame 与 Series 会复制后返回。

参数:

• X (ndarray | DataFrame | Series | List) -- 输入数据，不能为 None

• columns (List[str] | None) -- 列名列表，仅当 X 不是 DataFrame 时使用；为 None 时自动生成 feature_0、feature_1 …

返回:

转换后的 DataFrame

抛出:

• InputValidationError -- X 为 None、维度 >2，或列名数量与列数不匹配时

• InputTypeError -- 输入类型不受支持时

返回类型:

DataFrame

参考样例

>>> df = convert_to_dataframe(np.array([[1, 2], [3, 4]]), columns=['a', 'b'])
>>> df = convert_to_dataframe([[1, 2], [3, 4]])

hscredit.utils.extract_target_from_df(df, target='target', drop=True)

从 DataFrame 中提取目标变量。

参数:

• df (DataFrame) -- 输入 DataFrame

• target (str) -- 目标列名，默认 'target'

• drop (bool) -- 是否从返回的特征表中删除 target 列，默认 True

返回:

二元组 (X, y)，X 为特征 DataFrame，y 为目标 Series

抛出:

• InputTypeError -- 输入不是 DataFrame 时

• FeatureNotFoundError -- target 列不存在时

返回类型:

Tuple[DataFrame, Series]

参考样例

>>> X, y = extract_target_from_df(df, target='target')
>>> X, y = extract_target_from_df(df, target='label', drop=False)

hscredit.utils.check_array_1d(arr, name='array')

检查并转换为一维 Series。

参数:

• arr (ndarray | DataFrame | Series | List) -- 输入数组，ndarray / Series / list / tuple； 二维数组会被展平为一维

• name (str) -- 数组名称，用于生成错误信息与 Series 名称，默认 'array'

返回:

一维 Series

抛出:

• InputValidationError -- 数组为空时

• InputTypeError -- 类型不受支持时

返回类型:

Series

参考样例

>>> s = check_array_1d([0, 1, 1, 0], name='y')

hscredit.utils.get_feature_dtypes(X)

获取特征的数据类型分类（数值型 / 类别型）。

参数:

X (DataFrame) -- 特征 DataFrame

返回:

字典，含三个键：

• 'numeric'：数值型特征名列表（int/uint/float 各精度）

• 'categorical'：非数值型特征名列表（object、category、bool 等）

• 'all'：全部特征名列表

返回类型:

dict

参考样例

>>> dtypes = get_feature_dtypes(df)
>>> numeric_features = dtypes['numeric']
>>> categorical_features = dtypes['categorical']

hscredit.utils.check_missing_values(X, y=None, raise_error=False)

检查缺失值情况。

参数:

• X (DataFrame) -- 特征 DataFrame

• y (Series | None) -- 目标 ``Series``（可选），提供时一并统计其缺失情况

• raise_error (bool) -- 存在缺失值时是否抛出异常，默认 False（仅统计不报错）

返回:

缺失值统计字典，含键 X_missing``（总缺失数）、 ``X_missing_by_col``（各列缺失数）、``X_missing_ratio``（整体缺失率）； 当 ``y 非 None 时另含 y_missing、y_missing_ratio

抛出:

InputValidationError -- raise_error=True 且 X 或 y 存在缺失值时

返回类型:

dict

参考样例

>>> stats = check_missing_values(X, y)
>>> print(stats['X_missing_ratio'])
>>> check_missing_values(X, raise_error=True)   # 有缺失则报错