关于Hypothesis Pandas扩展生成独热编码与纵向数据的技术问询

问题1：直接生成独热编码列 #

无需先手动生成整数列再编码，可通过data_frames().map()结合自定义函数实现自动独热编码，同时保留缺失值：

• 定义onehot_multicategorical_column函数，接收目标列前缀，返回一个处理DataFrame的嵌套函数
• 嵌套函数内用pd.get_dummies生成独热编码，开启dummy_na=True识别缺失值
• 将原缺失值对应的独热编码列设为NaN，再删除临时生成的prefix_nan列
• 最后通过.map()将该函数应用到生成的DataFrame上，完成自动编码

问题2：生成带多级索引的纵向数据 #

通过st.builds配合自定义展开函数，构建符合纵向数据结构的多级索引：

• 用st.tuples(st.integers(0), st.lists(st.integers(0), min_size=1, max_size=5))生成（样本ID，时间点列表）的元组集合
• 用unpack_tuples函数将嵌套结构展平为（ID，单个时间点）的扁平元组列表
• 通过pd.MultiIndex.from_tuples将扁平元组转为多级索引，传给data_frames的index参数

问题3：精准控制缺失值 #

Hypothesis基础整数策略不支持缺失值，但可以通过st.one_of(st.none(), 原元素策略)给列添加缺失值：

• 二进制列示例：column("bin", elements=st.one_of(st.none(), st.integers(0, 1)))
• 多分类列示例：column("mult", elements=st.one_of(st.none(), st.sampled_from([0, 1, 2])))
• 配合独热编码的自定义函数，可直接保留缺失值，无需后续手动处理

问题4：生成列组合可变的数据集 #

通过st.lists动态生成列列表，指定采样范围和最小列数，即可生成列组合不固定的数据集：

columns_strategy = st.lists(
    st.sampled_from([
        column("ctn", dtype=float),
        column("bin", elements=st.one_of(st.none(), st.integers(0, 1))),
        column("mult", elements=st.one_of(st.none(), st.sampled_from([0, 1, 2])))
    ]),
    min_size=1,  # 确保至少包含一列
    unique=True  # 避免重复列
)

# 在测试中使用该策略
@given(data_frames(columns=columns_strategy, ...))

当前实现代码 #

from hypothesis import given, strategies as st
from hypothesis.extra.pandas import data_frames, column
import pandas as pd
import unittest

class TestTransforms(unittest.TestCase):
    @given(
        data_frames(
            columns=[
                # 连续变量
                column("ctn", dtype=float),
                # 二进制变量
                column("bin", elements=st.integers(0, 1)),
                # 多分类变量（数值编码）
                column("mult", elements=st.integers(0, 2)),
            ]
        )
    )
    def test_hypothesis(self, df):
        # 手动独热编码多分类列
        df = pd.concat(
            [
                df.drop(["mult"], axis=1),
                pd.get_dummies(df["mult"], prefix="mult"),
            ],
            axis=1,
        )

if __name__ == "__main__":
    unittest.main()

最终可用版本代码（修正重复函数+补全依赖导入） #

from hypothesis import given, strategies as st
from hypothesis.extra.pandas import data_frames, column
import pandas as pd
import numpy as np
import unittest
from typing import Callable


def onehot_multicategorical_column(
    prefix: str,
) -> Callable[[pd.DataFrame], pd.DataFrame]:
    def integrate_onehots(df: pd.DataFrame) -> pd.DataFrame:
        if df[prefix].empty:
            return df
        dummies = pd.get_dummies(df, columns=[prefix], prefix=prefix, dummy_na=True)
        # 保留缺失值
        dummies.loc[
            dummies[f"{prefix}_nan"].astype(bool),
            dummies.columns.str.startswith(prefix),
        ] = np.nan
        return dummies.drop(f"{prefix}_nan", axis=1)

    return integrate_onehots


def unpack_tuples(nested_tuples):
    """
    输入格式：List[Tuple[int, List[int]]]
    第一个int是样本ID，第二个是时间点列表
    将其展平为每个ID对应多个时间点的扁平元组列表
    示例：[(0,[0,1,2]), (1,[0,2])] => [(0,0), (0,1), (0,2), (1,0), (1,2)]
    """
    return [
        (pt_id, time_pt) for pt_id, time_pts in nested_tuples for time_pt in time_pts
    ]

class TestTransforms(unittest.TestCase):
    @given(
        data_frames(
            columns=[
                column("ctn", dtype=float),
                column("bin", elements=st.one_of(st.none(), st.integers(0, 1))),
                column(
                    "mult", elements=st.one_of(st.none(), st.sampled_from([0, 1, 2]))
                ),
            ],
            index=st.builds(
                pd.MultiIndex.from_tuples,
                st.lists(
                    st.tuples(
                        st.integers(0), st.lists(st.integers(0), min_size=1, max_size=5)
                    ),
                    min_size=2,
                ).map(unpack_tuples),
            ),
        ).map(onehot_multicategorical_column("mult"))
    )
    def test_hypothesis(self, df):
        # 编写你的测试逻辑
        pass


if __name__ == "__main__":
    unittest.main()

内容的提问来源于stack exchange，提问作者davzaman