基于dataclasses、pandas的Python函数式设计与继承架构问询

结合你函数式风格（不可变数据、纯函数）的需求，以及时间序列对象的语义层级约束，推荐以下两种可行方案：

方案1：语义化继承+结构验证 #

虽然子类的字段与父类完全一致，但继承在这里的核心作用是标记语义约束（即不同子类对应DataFrame的不同列结构），而非复用字段，这种方式完全合理且适配你的场景。

实现示例 #

from dataclasses import dataclass
import pandas as pd

# 基础抽象类，定义通用字段
@dataclass(frozen=True)
class BaseTimeSeries:
    log: pd.DataFrame
    sourceName: str

# 基础时间序列：约束log包含Time、A列
@dataclass(frozen=True)
class TimeSeries(BaseTimeSeries):
    def __post_init__(self):
        required_cols = {"Time", "A"}
        if not required_cols.issubset(self.log.columns):
            raise ValueError(f"TimeSeries 必须包含列：{required_cols}")

# 扩展时间序列：约束log包含Time、A、B列
@dataclass(frozen=True)
class ExtendedTimeSeries(BaseTimeSeries):
    def __post_init__(self):
        required_cols = {"Time", "A", "B"}
        if not required_cols.issubset(self.log.columns):
            raise ValueError(f"ExtendedTimeSeries 必须包含列：{required_cols}")

# 其余两种时间序列类同理实现

优势 #

• 保持函数式的不可变性，所有数据对象都是冻结的
• 静态类型检查工具（如mypy）能直接识别子类类型，函数可以通过类型注解严格约束输入输出
• 避免组合方式带来的DataFrame拆分/合并开销，以及行不匹配风险
• 通过__post_init__确保每个子类的DataFrame结构符合要求，从源头避免数据错误

方案2：标记联合类型（Tagged Union） #

如果不想维护多层继承结构，可以用单一数据类加类型标记的方式，通过标记区分不同语义的时间序列，同样能实现结构约束。

实现示例 #

from dataclasses import dataclass
import pandas as pd
from typing import Literal, Union

@dataclass(frozen=True)
class TimeSeriesData:
    log: pd.DataFrame
    sourceName: str
    # 用Literal标记所有合法的时间序列类型
    type_tag: Literal["basic", "extended", "type3", "type4"]

    def __post_init__(self):
        # 映射每种类型对应的必填列
        column_constraints = {
            "basic": {"Time", "A"},
            "extended": {"Time", "A", "B"},
            "type3": {"Time", "A", "B", "C"},
            "type4": {"Time", "A", "D"}
        }
        required_cols = column_constraints[self.type_tag]
        if not required_cols.issubset(self.log.columns):
            raise ValueError(f"类型 {self.type_tag} 必须包含列：{required_cols}")

# 定义类型别名，方便函数注解
BasicTimeSeries = TimeSeriesData[type_tag="basic"]
ExtendedTimeSeries = TimeSeriesData[type_tag="extended"]

优势 #

• 无需维护多个子类，结构更简洁
• 同样能通过type_tag和初始化验证保证数据结构正确性
• 适合类型转换频繁的场景，只需修改标记和DataFrame即可生成新对象

配套函数式转换逻辑 #

无论采用哪种方案，都应使用纯函数来实现不同类型时间序列的转换，确保不可变性和无副作用：

def extend_basic_to_extended(basic_ts: TimeSeries, b_series: pd.Series) -> ExtendedTimeSeries:
    # 验证行匹配
    if len(basic_ts.log) != len(b_series):
        raise ValueError("B列长度与基础时间序列不匹配")
    # 生成新的DataFrame（原对象保持不变）
    extended_log = pd.concat([basic_ts.log, b_series.rename("B")], axis=1)
    return ExtendedTimeSeries(log=extended_log, sourceName=basic_ts.sourceName)

方案选择建议 #

• 若需要严格的类型隔离、IDE自动提示和静态类型检查，优先选语义化继承方案
• 若类型较多或转换逻辑灵活，优先选标记联合类型方案

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