"""Ordinal Encoder (序数编码器).
将类别特征转换为整数编码。
"""
from typing import Optional, List, Dict, Any, Union
import numpy as np
import pandas as pd
from .base import BaseEncoder
from ...exceptions import NotFittedError
[文档]
class OrdinalEncoder(BaseEncoder):
"""序数编码器.
将每个类别映射为一个整数，保留类别的顺序关系（如果存在）。
适用于树模型和需要保留单一特征维度的场景。
**参数**
:param cols: 需要编码的列名列表。如果为None，则自动识别所有列（支持类别型和数值型）
:param mapping: 自定义映射字典，如{'col': {'a': 1, 'b': 2}}，默认为None
:param handle_unknown: 处理未知类别的方式，默认为'value'
:param handle_missing: 处理缺失值的方式，默认为'value'
:param drop_invariant: 是否删除方差为0的列，默认为False
:param return_df: 是否返回DataFrame，默认为True
**属性**
- mapping_: 序数编码映射字典，格式为 {col: {category: integer}}
**参考样例**
>>> from hscredit.core.encoders import OrdinalEncoder
>>> encoder = OrdinalEncoder(cols=['education'])
>>> X_encoded = encoder.fit_transform(X)
>>>
>>> # 自定义映射
>>> mapping = {'education': {'high': 3, 'medium': 2, 'low': 1}}
>>> encoder = OrdinalEncoder(cols=['education'], mapping=mapping)
>>> X_encoded = encoder.fit_transform(X)
**注意**
默认整数映射不含真实序关系，仅对天然有序的类别（如学历 高/中/低）通过 ``mapping``
显式指定顺序才有意义；无序类别用于线性模型时应改用 OneHot 或 WOE 编码，否则会引入
虚假的大小关系。
**引用**
序数编码参见 sklearn ``OrdinalEncoder``：
https://scikit-learn.org/stable/modules/generated/sklearn.preprocessing.OrdinalEncoder.html
"""
def _get_category_cols(self, X: pd.DataFrame) -> List[str]:
"""自动识别需要编码的列。
OrdinalEncoder支持数值型和类别型列，因此返回所有列。
:param X: 输入数据
:return: 列名列表
"""
return X.columns.tolist()
def __init__(
self,
cols: Optional[List[str]] = None,
mapping: Optional[Dict[str, Dict[Any, int]]] = None,
handle_unknown: str = "value",
handle_missing: str = "value",
drop_invariant: bool = False,
return_df: bool = True,
target: Optional[str] = None,
n_jobs: Optional[Union[int, float]] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
):
"""初始化序数编码器。
:param cols: 需要编码的列名列表
:param mapping: 自定义映射字典，如{'col': {'a': 1, 'b': 2}}，默认为None
:param handle_unknown: 处理未知类别的方式，默认为'value'
:param handle_missing: 处理缺失值的方式，默认为'value'
:param drop_invariant: 是否删除方差为0的列，默认为False
:param return_df: 是否返回DataFrame，默认为True
:param target: scorecardpipeline风格的目标列名。序数编码器不使用此参数，仅为API一致性保留
"""
super().__init__(
cols=cols,
drop_invariant=drop_invariant,
return_df=return_df,
handle_unknown=handle_unknown,
handle_missing=handle_missing,
target=target,
n_jobs=n_jobs,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
)
self.mapping = mapping
def _fit(self, X: pd.DataFrame, y: Optional[pd.Series] = None):
"""拟合序数编码器。
:param X: 输入数据，shape (n_samples, n_features)
:param y: 目标变量（可选），序数编码器不需要
"""
self._fit_columns(X, y)
def _fit_column(self, column, values, y=None):
configured_mapping = self.mapping or {}
if column in configured_mapping:
mapping = configured_mapping[column].copy()
else:
categories = values.dropna().unique()
categories = self._sort_categories([c for c in categories if c is not np.nan])
mapping = {cat: i for i, cat in enumerate(categories)}
if self.handle_missing == "value":
mapping[np.nan] = -1
elif self.handle_missing == "return_nan":
mapping[np.nan] = np.nan
if self.handle_unknown == "value":
mapping["__UNKNOWN__"] = -1
elif self.handle_unknown == "return_nan":
mapping["__UNKNOWN__"] = np.nan
return {"mapping_": mapping}
def _transform(self, X: pd.DataFrame, y: Optional[pd.Series] = None) -> pd.DataFrame:
"""转换数据。
:param X: 输入数据，shape (n_samples, n_features)
:param y: 目标变量（可选），序数编码器不需要
:return: 编码后的数据
"""
return self._transform_columns(X, y)
def _transform_column(self, column, values, y=None, context=None):
mapping = self.mapping_[column]
result = values.map(mapping)
if self.handle_unknown == "value":
result = result.fillna(-1)
elif self.handle_unknown == "error" and result.isna().any():
raise ValueError(f"列'{column}'包含未知类别")
return result