我正在使用PYDANIC创建一个基于PANDA TIMESTAMP(start,end)和Timedelta(period)对象的时间序列模型.该模型将用于一个具有多个配置/场景的小型数据分析程序.

我需要基于两个bool(include_end_period,allow_future)和一个可选int(max_periods)配置参数实例化和验证TimeSeries模型的各个方面.然后,我需要派生三个新字段(timezonetotal_durationtotal_periods)并执行一些额外的验证.

由于在验证另一个值时需要使用一个值的几个实例,我无法使用典型的@validator个方法获得所需的结果.特别是,我经常会得到一个丢失的KeyError,而不是预期的ValueError.我找到的最好的解决方案是创建一个长@root_validator(pre=True)方法.

from pydantic import BaseModel, ValidationError, root_validator, conint
from pandas import Timestamp, Timedelta


class Timeseries(BaseModel):
    start: Timestamp
    end: Timestamp
    period: Timedelta
    include_end_period: bool = False
    allow_future: bool = True
    max_periods: conint(gt=0, strict=True) | None = None
    
    # Derived values, do not pass as params
    timezone: str | None
    total_duration: Timedelta
    total_periods: conint(gt=0, strict=True)
    
    class Config:
        extra = 'forbid'
        validate_assignment = True
    
    
    @root_validator(pre=True)
    def _validate_model(cls, values):
        
        # Validate input values
        if values['start'] > values['end']:
            raise ValueError('Start timestamp cannot be later than end')
        if values['start'].tzinfo != values['end'].tzinfo:
            raise ValueError('Start, end timezones do not match')
        if values['period'] <= Timedelta(0):
            raise ValueError('Period must be a positive amount of time')
        
        # Set timezone
        timezone = values['start'].tzname()
        if 'timezone' in values and values['timezone'] != timezone:
            raise ValueError('Timezone param does not match start timezone')
        values['timezone'] = timezone
        
        # Set duration (add 1 period if including end period)
        total_duration = values['end'] - values['start']
        if values['include_end_period']:
            total_duration += values['period']
        if 'total_duration' in values and values['total_duration'] != total_duration:
            error_context = ' + 1 period (included end period)' if values['include_end_period'] else ''
            raise ValueError(f'Duration param does not match end - start timestamps{error_context}')
        values['total_duration'] = total_duration
        
        # Set total_periods
        periods_float: float = values['total_duration'] / values['period']
        if periods_float != int(periods_float):
            raise ValueError('Total duration not divisible by period length')
        total_periods = int(periods_float)
        if 'total_periods' in values and values['total_periods'] != total_periods:
            raise ValueError('Total periods param does not match')
        values['total_periods'] = total_periods
        
        # Validate future
        if not values['allow_future']:
            # Get current timestamp to floor of period (subtract 1 period if including end period)
            max_end: Timestamp = Timestamp.now(tz=values['timezone']).floor(freq=values['period'])
            if values['include_end_period']:
                max_end -= values['period']
            if values['end'] > max_end:
                raise ValueError('End period is future or current (incomplete)')
        
        # Validate derived values
        if values['total_duration'] < Timedelta(0):
            raise ValueError('Total duration must be positive amount of time')
        if values['max_periods'] and values['total_periods'] > values['max_periods']:
            raise ValueError('Total periods exceeds max periods param')
        
        return values

在令人满意的情况下实例化模型,使用所有配置判断:

start = Timestamp('2023-03-01T00:00:00Z')
end = Timestamp('2023-03-02T00:00:00Z')
period = Timedelta('5min')

try:
    ts = Timeseries(start=start, end=end, period=period,
                    include_end_period=True, allow_future=False, max_periods=10000)
    print(ts.dict())
except ValidationError as e:
    print(e)

输出:

"""
{'start': Timestamp('2023-03-01 00:00:00+0000', tz='UTC'),
 'end': Timestamp('2023-03-02 00:00:00+0000', tz='UTC'),
 'period': Timedelta('0 days 00:05:00'),
 'include_end_period': True,
 'allow_future': False,
 'max_periods': 10000,
 'timezone': 'UTC',
 'total_duration': Timedelta('1 days 00:05:00'),
 'total_periods': 289}
"""

在这里,我相信我的所有验证都像预期的那样工作,并提供了预期的ValueErrors,而不是帮助较小的KeyErrors.Is this approach reasonable?这似乎与典型的/推荐的方法背道而驰,与@validator相比,@root_validator的文档相当简短.

我还不满意需要在模型顶部列出派生值(timezonetotal_durationtotal_periods).这意味着它们可以/应该在实例化时传递,并且在我的验证器脚本中需要额外的逻辑来判断它们是否被传递,以及它们是否与派生的值匹配.如果省略它们,它们将不会从类型、约束等的默认验证中受益,并且会迫使我将配置更改为extra='allow'.如果有任何关于如何改进这方面的建议,我将不胜感激.

谢谢!

推荐答案

出于测试目的,拥有如此大的功能通常是个好主意.即使您想采用root_validator方法,您仍然可以(并且IMO应该)将逻辑划分为不同的、在语义上有意义的方法.

但我会建议一种完全不同的方法.由于timezonetotal_durationtotal_periods是从其他字段派生的,且该过程不是非常昂贵,因此我将为这些字段定义属性,而不是将它们作为字段.

这样做的好处是您不需要预先计算它们的值,这意味着您不需要使用pre=True方法,并且可以在field-specific验证器中使用以前验证过的字段值.

当您确实需要确保许多不同的字段一起遵循某个逻辑时,根验证器仍然是有意义的.

以下是我的建议:

from collections.abc import Mapping
from typing import Any

from pandas import Timedelta, Timestamp
from pydantic import BaseModel, conint, root_validator, validator

AnyMap = Mapping[str, Any]


class Timeseries(BaseModel):
    start: Timestamp
    end: Timestamp
    period: Timedelta
    include_end_period: bool = False
    allow_future: bool = True
    max_periods: conint(gt=0, strict=True) | None = None

    @validator("end")
    def ensure_end_consistent_with_start(
        cls,
        v: Timestamp,
        values: AnyMap,
    ) -> Timestamp:
        val_start: Timestamp = values["start"]
        if v < val_start:
            raise ValueError("Start timestamp cannot be later than end")
        if val_start.tzinfo != v.tzinfo:
            raise ValueError("Start, end timezones do not match")
        return v

    @validator("period")
    def ensure_period_is_positive(cls, v: Timedelta) -> Timedelta:
        if v <= Timedelta(0):
            raise ValueError("Period must be a positive amount of time")
        return v

    @validator("period")
    def ensure_period_divides_duration(
        cls,
        v: Timedelta,
        values: AnyMap,
    ) -> Timedelta:
        duration: float = (values["end"] - values["start"]) / v
        if int(duration) != duration:
            raise ValueError("Total duration not divisible by period length")
        return v

    @root_validator
    def ensure_end_is_allowed(cls, values: AnyMap) -> AnyMap:
        if values["allow_future"]:
            return values
        val_period: Timedelta = values["period"]
        val_end: Timestamp = values["end"]
        max_end = Timestamp.now(tz=val_end.tzname()).floor(freq=val_period)
        if values["include_end_period"]:
            max_end -= val_period
        if val_end > max_end:
            raise ValueError("End period is future or current (incomplete)")
        return values

    @root_validator
    def ensure_num_periods_allowed(cls, values: AnyMap) -> AnyMap:
        periods = int((values["end"] - values["start"]) / values["period"])
        if values["include_end_period"]:
            periods += 1
        if values["max_periods"] and periods > values["max_periods"]:
            raise ValueError("Total periods exceeds max periods param")
        return values

    @property
    def timezone(self) -> str | None:
        return self.start.tzname()

    @property
    def total_duration(self) -> Timedelta:
        total_duration = self.end - self.start
        if self.include_end_period:
            total_duration += self.period
        return total_duration

    @property
    def total_periods(self) -> int:
        return int(self.total_duration / self.period)

我想这是个人喜好的问题,何时从字段验证器切换到根验证器.例如,您可以争辩说ensure_period_divides_duration应该是根验证器,因为它使用三个字段的值.

当然,您的示例数据也适用于此模型.

需要注意的一点是,当您验证endstart之后(period平均分配总持续时间)时,对total_periods的范围限制无论如何都是多余的.

你也可以争辩说,即使是total_duration这样简单的东西也不应该是一处房产.在这种情况下,您可以将其设置为名为get_total_duration的方法.

但是,如果您有那些"派生"的字段,您总是会遇到这样的问题:必须判断用户传递的内容是否与其余数据一致.

我相信,一旦平丹蒂克v2下降,大部分令人头疼的问题将会消失,它promise computed fields(见plan for v2).

Python相关问答推荐

为什么图像结果翻转了90度?

如何对行使用分段/部分.diff()或.pct_change()?

避免循环的最佳方法

如何在超时的情况下同步运行Matplolib服务器端?该过程随机挂起

当值是一个integer时,在Python中使用JMESPath来验证字典中的值(例如:1)

从管道将Python应用程序部署到Azure Web应用程序,不包括需求包

添加包含中具有任何值的其他列的计数的列

计算相同形状的两个张量的SSE损失

Pythind 11无法弄清楚如何访问tuple元素

Python daskValue错误:无法识别的区块管理器dask -必须是以下之一:[]

Deliveryter Notebook -无法在for循环中更新matplotlib情节(保留之前的情节),也无法使用动画子功能对情节进行动画

根据在同一数据框中的查找向数据框添加值

Django mysql图标不适用于小 case

PMMLPipeline._ fit()需要2到3个位置参数,但给出了4个位置参数

切片包括面具的第一个实例在内的眼镜的最佳方法是什么?

如何在给定的条件下使numpy数组的计算速度最快?

Python导入某些库时非法指令(核心转储)(beautifulsoup4."" yfinance)

在pandas/python中计数嵌套类别

如何使用OpenGL使球体遵循Python中的八样路径?

PYTHON、VLC、RTSP.屏幕截图不起作用