Press "Enter" to skip to content

类型提示数据框进行静态分析和运行时验证

如何使用StaticFrame实现完整的DataFrame类型提示

作者提供的照片

自从Python 3.5引入了类型提示功能以来,对DataFrame进行静态类型化通常只限于指定类型本身:

def process(f: DataFrame) -> Series: ...

这是不足够的,因为它忽略了容器内包含的类型。DataFrame可能具有字符串列标签和三列整数、字符串和浮点值;这些特征定义了类型。使用这样的类型提示的函数参数可以为开发人员、静态分析器和运行时检查器提供了解接口期望的全部信息。StaticFrame是一个开源项目,我是其中的主要开发者。它现在支持这种类型提示:

from typing import Anyfrom static_frame import Frame, Index, TSeriesAnydef process(f: Frame[   # 容器的类型        Any,            # 索引标签的类型        Index[np.str_], # 列标签的类型        np.int_,        # 第一列的类型        np.str_,        # 第二列的类型        np.float64,     # 第三列的类型        ]) -> TSeriesAny: ...

所有核心的StaticFrame容器现在都支持泛型的规范。虽然可以静态检查,但新的装饰器@CallGuard.check允许在函数接口上运行时验证这些类型提示。此外,使用Annotated泛型,新的Require类定义了一系列强大的运行时验证器,允许对每个列或每行的数据进行检查。最后,每个容器都提供了一个新的via_type_clinic接口来推导和验证类型提示。这些工具共同提供了一种用于类型提示和验证DataFrame的一致方法。

泛型DataFrame的需求

Python的内置泛型类型(如tupledict)需要指定组件类型(如tuple[int, str, bool]dict[str, int])。定义组件类型可以实现更准确的静态分析。虽然对于DataFrames也是如此,但很少有人尝试为DataFrames定义全面的类型提示。

Pandas即使使用<pandas-stubs包,也不允许指定DataFrame组件的类型。Pandas DataFrame允许广泛的原地变异,在静态类型上可能不合理。幸运的是,StaticFrame中提供了不可变的DataFrame。</pandas-stubs

此外,Python用于定义泛型的工具直到最近才适合于DataFrames。一个DataFrame具有可变数量的异构列类型,这对于泛型的规范提出了挑战。使用Python 3.11中引入的新的TypeVarTuple(在typing_extensions包中进行了反向移植),对此结构进行类型定义变得更容易。

TypeVarTuple允许定义接受可变数量类型的泛型(详细信息请参见PEP 646)。有了这个新的类型变量,StaticFrame可以定义一个带有索引类型TypeVar、列类型TypeVar和零个或多个列类型的TypeVarTuple的泛型Frame

一个泛型Series由索引类型TypeVar和值类型TypeVar定义。StaticFrame的IndexIndexHierarchy也是泛型的,后者再次利用TypeVarTuple为每个深度级别定义了变量数量的组件Index

StaticFrame使用NumPy类型来定义Frame的列类型或SeriesIndex的值类型。这允许准确地指定大小的数字类型,如np.uint8np.complex128;或广泛地指定类型的类别,如np.integernp.inexact。由于StaticFrame支持所有NumPy类型,它们之间的对应关系是直接的。

用泛型 DataFrame 定义的接口

扩展上面的例子,下面的函数接口展示了一个具有三列的 Frame,被转换成一个 Series 字典。由于组件类型的提示提供了更多的信息,函数的目的几乎是显而易见的。

from typing import Anyfrom static_frame import Frame, Series, Index, IndexYearMonthdef process(f: Frame[        Any,        Index[np.str_],        np.int_,        np.str_,        np.float64,        ]) -> dict[                int,                Series[                 # 容器的类型                        IndexYearMonth, # 索引标签的类型                        np.float64,     # 值的类型                        ],                ]: ...

该函数处理来自一个 开源资产定价(OSAP)数据集(公司级特征 / 个体 / 预测变量)的信号表格。每个表格有三列:证券标识符(标有“permno”),年份和月份(标有“yyyymm”),以及信号(具有特定信号的名称)。

该函数忽略了提供的 Frame 的索引(类型为 Any),并根据第一列“permno” np.int_ 值创建了分组。返回一个以“permno”为键的字典,其中每个值是“permno”对应的 np.float64 值的 Series;索引是从 np.str_ 的“yyyymm”列创建的 IndexYearMonth。(StaticFrame 使用 NumPy 的 datetime64 值来定义单元类型的索引: IndexYearMonth 存储 datetime64[M] 标签。)

与返回一个 dict 不同,下面的函数返回一个具有层次化索引的 SeriesIndexHierarchy 泛型为每个深度级别指定了一个组件 Index;这里,外部深度是 Index[np.int_](根据“permno”列派生),内部深度是 IndexYearMonth(根据“yyyymm”列派生)。

from typing import Anyfrom static_frame import Frame, Series, Index, IndexYearMonth, IndexHierarchydef process(f: Frame[        Any,        Index[np.str_],        np.int_,        np.str_,        np.float64,        ]) -> Series[                    # 容器的类型                IndexHierarchy[          # 索引标签的类型                        Index[np.int_],  # 索引深度 0 的类型                        IndexYearMonth], # 索引深度 1 的类型                np.float64,              # 值的类型                ]: ...

丰富的类型提示提供了一个自我说明的接口,使功能变得明确。更好的是,这些类型提示可以与 Pyright(现在)和 Mypy(待 TypeVarTuple 完全支持)一起用于静态分析。例如,用包含两列 np.float64Frame 调用该函数将无法通过静态分析类型检查或在编辑器中生成警告。

运行时类型验证

静态类型检查可能还不够:运行时评估提供了更强的约束,特别适用于动态或不完整(或错误的)类型提示的值。

基于新的运行时类型检查器 TypeClinic,StaticFrame 2 引入了 @CallGuard.check,一个用于运行时验证类型提示接口的装饰器。支持所有 StaticFrame 和 NumPy 泛型,大多数内置的 Python 类型也都得到支持,即使是深度嵌套的情况。下面的函数添加了 @CallGuard.check 装饰器。

from typing import Anyfrom static_frame import Frame, Series, Index, IndexYearMonth, IndexHierarchy, CallGuard@CallGuard.checkdef process(f: Frame[        Any,        Index[np.str_],        np.int_,        np.str_,        np.float64,        ]) -> Series[                IndexHierarchy[Index[np.int_], IndexYearMonth],                np.float64,                ]: ...

现在使用 @CallGuard.check 进行装饰,如果以上函数使用未标记的 Frame 调用,其中包含两列 np.float64 ,则会引发 ClinicError 异常,说明预期有三列,但只提供了两列,并且预期为字符串列标签,但提供了整数标签。(要发出警告而不是引发异常,请使用 @CallGuard.warn 装饰器。)

ClinicError:In args of (f: Frame[Any, Index[str_], int64, str_, float64]) -> Series[IndexHierarchy[Index[int64], IndexYearMonth], float64]└── Frame[Any, Index[str_], int64, str_, float64]    └── Expected Frame has 3 dtype, provided Frame has 2 dtypeIn args of (f: Frame[Any, Index[str_], int64, str_, float64]) -> Series[IndexHierarchy[Index[int64], IndexYearMonth], float64]└── Frame[Any, Index[str_], int64, str_, float64]    └── Index[str_]        └── Expected str_, provided int64 invalid

运行时数据验证

其他特征可以在运行时进行验证。例如,验证 shapename 属性,以及索引或列上的标签序列。StaticFrame 的 Require 类提供了一系列可配置的验证器。

  • Require.Name:验证容器的 ``name`` 属性。
  • Require.Len:验证容器的长度。
  • Require.Shape:验证容器的 ``shape`` 属性。
  • Require.LabelsOrder:验证标签的顺序。
  • Require.LabelsMatch:验证标签的包含性,与顺序无关。
  • Require.Apply:对容器应用返回布尔值的函数。

符合不断增长的趋势,这些对象作为一个或多个额外参数提供给 Annotated 泛型的类型提示中。 (有关详细信息,请参阅 PEP 593)第一个 Annotated 参数引用的类型是后续参数验证器的目标。例如,如果将 Index[np.str_] 类型提示替换为 Annotated[Index[np.str_], Require.Len(20)] 类型提示,则会将运行时长度验证应用于与第一个参数关联的索引。

扩展处理 OSAP 信号表的示例,我们可以验证对列标签的期望。 Require.LabelsOrder 验证器可以定义一个标签序列,可选择使用 来表示零个或多个未指定的连续标签。要指定表的前两列标签为“permno”和“yyyymm”,而第三个标签是变量(取决于信号),可以在 Annotated 泛型中定义以下 Require.LabelsOrder

from typing import Any, Annotatedfrom static_frame import Frame, Series, Index, IndexYearMonth, IndexHierarchy, CallGuard, Require@CallGuard.checkdef process(f: Frame[        Any,        Annotated[                Index[np.str_],                Require.LabelsOrder('permno', 'yyyymm', ...),                ],        np.int_,        np.str_,        np.float64,        ]) -> Series[                IndexHierarchy[Index[np.int_], IndexYearMonth],                np.float64,                ]: ...

如果接口期望一个小的 OSAP 信号表集合,可以使用 Require.LabelsMatch 验证器验证第三列。此验证器可以指定所需的标签、标签集(其中至少一个必须匹配)和正则表达式模式。如果只期望来自三个文件的表(即“Mom12m.csv”、“Mom6m.csv”和“LRreversal.csv”),可以使用一个集合定义第三列标签的 Require.LabelsMatch

@CallGuard.checkdef process(f: Frame[        Any,        Annotated[                Index[np.str_],                Require.LabelsOrder('permno', 'yyyymm', ...),                Require.LabelsMatch({'Mom12m', 'Mom6m', 'LRreversal'}),                ],        np.int_,        np.str_,        np.float64,        ]) -> Series[                IndexHierarchy[Index[np.int_], IndexYearMonth],                np.float64,                ]: ...

Both Require.LabelsOrder and Require.LabelsMatch can associate functions with label specifiers to validate data values. If the validator is applied to column labels, a Series of column values will be provided to the function; if the validator is applied to index labels, a Series of row values will be provided to the function.

Annotated的用法类似,标签指定符被替换为一个列表,其中第一项是标签指定符,剩余项是返回布尔值的行或列处理函数。

扩展以上示例,我们可以验证所有“permno”值是否大于零,以及所有信号值(“Mom12m”,“Mom6m”,“LRreversal”)是否大于等于-1。

from typing import Any, Annotatedfrom static_frame import Frame, Series, Index, IndexYearMonth, IndexHierarchy, CallGuard, Require@CallGuard.checkdef process(f: Frame[        Any,        Annotated[                Index[np.str_],                Require.LabelsOrder(                        ['permno', lambda s: (s > 0).all()],                        'yyyymm',                        ...,                        ),                Require.LabelsMatch(                        [{'Mom12m', 'Mom6m', 'LRreversal'}, lambda s: (s >= -1).all()],                        ),                ],        np.int_,        np.str_,        np.float64,        ]) -> Series[                IndexHierarchy[Index[np.int_], IndexYearMonth],                np.float64,                ]: ...

如果验证失败,@CallGuard.check将引发异常。例如,如果上述函数被调用并传入一个具有意外的第三列标签的Frame,将引发以下异常:

ClinicError:In args of (f: Frame[Any, Annotated[Index[str_], LabelsOrder(['permno', <lambda>], 'yyyymm', ...), LabelsMatch([{'Mom12m', 'LRreversal', 'Mom6m'}, <lambda>])], int64, str_, float64]) -> Series[IndexHierarchy[Index[int64], IndexYearMonth], float64]└── Frame[Any, Annotated[Index[str_], LabelsOrder(['permno', <lambda>], 'yyyymm', ...), LabelsMatch([{'Mom12m', 'LRreversal', 'Mom6m'}, <lambda>])], int64, str_, float64]    └── Annotated[Index[str_], LabelsOrder(['permno', <lambda>], 'yyyymm', ...), LabelsMatch([{'Mom12m', 'LRreversal', 'Mom6m'}, <lambda>])]        └── LabelsMatch([{'Mom12m', 'LRreversal', 'Mom6m'}, <lambda>])            └── 预期的标签与frozenset({'Mom12m', 'LRreversal', 'Mom6m'})不匹配

使用TypeVarTuple的表达能力

如上所示,TypeVarTuple允许指定具有零个或多个异构列类型的Frame。例如,我们可以为包含两个浮点类型或六个混合类型的Frame提供类型提示:

>>> from typing import Any>>> from static_frame import Frame, Index>>> f1: sf.Frame[Any, Any, np.float64, np.float64]>>> f2: sf.Frame[Any, Any, np.bool_, np.float64, np.int8, np.int8, np.str_, np.datetime64]

虽然这可以适应各种DataFrame,但为具有数百列的宽DataFrame提供类型提示将变得笨拙。Python 3.11引入了一种新的语法,以在TypeVarTuple泛型中提供可变范围的类型:使用全部的星表达式。例如,要为具有日期索引、字符串列标签和任意列类型组合的Frame提供类型提示,可以对零个或多个全部进行星解包:

>>> from typing import Any>>> from static_frame import Frame, Index>>> f: sf.Frame[Index[np.datetime64], Index[np.str_], *tuple[All, ...]]

元组星号表达式可以在类型列表的任何位置使用,但只能使用一个。例如,下面的类型提示定义了一个Frame,它必须以布尔和字符串列开头,但对于后续的任意数量的np.float64列具有灵活的规范。

>>> from typing import Any>>> from static_frame import Frame>>> f: sf.Frame[Any, Any, np.bool_, np.str_, *tuple[np.float64, ...]]

类型提示的工具

使用这样详细的类型提示可能会有挑战性。为了帮助用户,StaticFrame提供了方便的运行时类型提示和检查工具。所有的StaticFrame 2容器现在都具有via_type_clinic接口,可以访问TypeClinic的功能。

首先,提供了一些实用工具,用于将容器(例如一个完整的Frame)转换为类型提示。容器的via_type_clinic接口的字符串表示提供了容器类型提示的字符串表示;或者,to_hint()方法返回一个完整的泛型别名对象。

>>> import static_frame as sf>>> f = sf.Frame.from_records(([3, '192004', 0.3], [3, '192005', -0.4]), columns=('permno', 'yyyymm', 'Mom3m'))>>> f.via_type_clinicFrame[Index[int64], Index[str_], int64, str_, float64]>>> f.via_type_clinic.to_hint()static_frame.core.frame.Frame[static_frame.core.index.Index[numpy.int64], static_frame.core.index.Index[numpy.str_], numpy.int64, numpy.str_, numpy.float64]

其次,提供了用于运行时类型提示测试的实用工具。via_type_clinic.check()函数允许根据提供的类型提示验证容器。

>>> f.via_type_clinic.check(sf.Frame[sf.Index[np.str_], sf.TIndexAny, *tuple[tp.Any, ...]])ClinicError:In Frame[Index[str_], Index[Any], Unpack[Tuple[Any, ...]]]└── Index[str_]    └── Expected str_, provided int64 invalid

为了支持逐步类型化,StaticFrame定义了几个通用别名,对于每个组件类型都配置有Any。例如,TFrameAny可以用于任何FrameTSeriesAny可以用于任何Series。顾名思义,TFrameAny将验证上面创建的Frame

>>> f.via_type_clinic.check(sf.TFrameAny)

结论

对于DataFrames来说,更好的类型提示已经过时了。借助现代Python类型提示工具和基于不可变数据模型构建的DataFrame,StaticFrame 2满足了这个需求,为那些优先考虑可维护性和可验证性的工程师提供了强大的资源。

Leave a Reply

Your email address will not be published. Required fields are marked *