如何确保DataFrame计算时无法访问当前行的后续行数据

我有一个带pd.DatetimeIndex索引、仅含price数值列的比特币数据DataFrame：

price
timestamp   
2022-01-01 00:00:00 46250.00
2022-01-01 00:01:00 46312.76
2022-01-01 00:02:00 46368.73
2022-01-01 00:03:00 46331.08
2022-01-01 00:04:00 46321.34
... ...
2023-02-28 23:55:00 23116.25
2023-02-28 23:56:00 23121.22
2023-02-28 23:57:00 23121.42
2023-02-28 23:58:00 23122.22
2023-02-28 23:59:00 23141.57

我正在搭建一个平台，允许用户编写向量化函数预测比特币未来价格，示例函数如下：

def future_price_will_be_higher(df):
    return df.shift(1) < df

核心需求是禁止用户访问未来数据：比如预测2022-06-01 00:00行的价格时，不能使用该行之后的数据，像df.shift(-1)这类获取未来数据的操作必须被限制。从数学上看，计算df[i]时，绝对不能访问df[j]（j > i）。

现在有几个具体问题：

1. 能否通过代码分析或修改DataFrame的方式，确保向量化的pandas函数无法访问当前行的后续数据？
2. 我曾考虑子类化pd.DataFrame并重写.shift(period)方法，限制period不能为负数，但仅重写shift是否足够？还有哪些函数需要重写？
3. 子类化存在一个问题：对改写后的DataFrame执行任何操作（比如切片df[:]）后，返回的对象类型会变回pd.DataFrame，不再是子类实例：

class OverridenDataFrame(pd.DataFrame):
    def shift(self, periods: int = ..., freq=..., axis=..., fill_value=...) -> pd.DataFrame:
        assert periods > 0, periods
        return super().shift(periods, freq, axis, fill_value)

df = OverridenDataFrame(df)
print(type(df))  # OverridenDataFrame
df = df[:]  # 切片或其他操作返回pd.DataFrame
print(type(df))  # pd.DataFrame

一、代码分析层面：静态检查+运行时监控 #

1. 静态代码扫描 #

用Python的AST（抽象语法树）模块分析用户提交的函数代码，直接拦截违规操作：

• 查找shift调用中是否传入负的periods参数；
• 检查pct_change的periods参数是否为负；
• 识别是否存在反转索引（比如df.iloc[::-1]）、降序排序索引的操作；
• 排查滚动窗口（rolling）是否设置了包含未来数据的窗口规则（比如closed='right'且窗口覆盖后续行）。

这种方式能提前拦截大部分明显的违规代码，成本低且高效。

2. 运行时监控 #

对用户函数的执行过程做数据访问校验：可以封装一个代理对象，每次访问数据行时记录索引范围，确保计算当前行i时，所有访问的索引都≤i。不过这种方式对向量化操作的监控难度较高，更适合结合逐行验证的场景——比如对比用户向量化输出和逐行用历史子集计算的结果是否一致。

二、数据封装层面：改进子类化方案 #

仅重写shift远远不够，用户还能通过多种方式获取未来数据，同时要解决子类类型丢失的问题。

1. 需要重写的关键方法 #

除了shift，还需拦截这些可能访问未来数据的方法：

• pct_change：禁止传入负的periods参数；
• sort_index：禁止降序排序，避免未来数据被移到前面；
• rolling：限制窗口的closed参数为left或both，确保窗口只包含历史数据；
• _constructor：这是pandas控制返回对象类型的核心方法，重写它能让切片、过滤等操作后仍返回子类实例。

2. 修复子类类型丢失的完整子类实现 #

class OverridenDataFrame(pd.DataFrame):
    def _constructor(self, data=None, index=None, columns=None, dtype=None, copy=False):
        # 确保所有操作返回子类实例
        return OverridenDataFrame(data=data, index=index, columns=columns, dtype=dtype, copy=copy)
    
    def shift(self, periods: int = 1, freq=None, axis=0, fill_value=None):
        if periods < 0:
            raise ValueError("禁止使用负periods的shift操作，无法访问未来数据")
        return super().shift(periods=periods, freq=freq, axis=axis, fill_value=fill_value)
    
    def pct_change(self, periods=1, fill_method='pad', limit=None, freq=None, **kwargs):
        if periods < 0:
            raise ValueError("禁止使用负periods的pct_change操作，无法访问未来数据")
        return super().pct_change(periods=periods, fill_method=fill_method, limit=limit, freq=freq, **kwargs)
    
    def sort_index(self, axis=0, level=None, ascending=True, inplace=False, kind='quicksort', na_position='last', sort_remaining=True, ignore_index=False, key=None):
        if not ascending:
            raise ValueError("禁止降序排序索引，避免未来数据被前置")
        return super().sort_index(axis=axis, level=level, ascending=ascending, inplace=inplace, kind=kind, na_position=na_position, sort_remaining=sort_remaining, ignore_index=ignore_index, key=key)

重写_constructor后，执行df[:]、df[df['price'] > 40000]这类操作时，返回的仍是OverridenDataFrame实例。

三、极端场景方案：逐行历史子集限制 #

如果要彻底杜绝未来数据访问，可以遍历每个时间点，只给用户传入当前行及之前的历史数据子集：

def run_with_history_only(df, user_func):
    results = []
    for i in range(len(df)):
        # 只传递当前行及之前的历史数据
        history_df = df.iloc[:i+1]
        pred = user_func(history_df).iloc[-1]
        results.append(pred)
    return pd.Series(results, index=df.index)

这种方式完全避免了未来数据访问，但会损失向量化的性能优势，适合数据量不大、对合规性要求极高的场景。

四、综合方案建议 #

1. 优先用静态代码分析+改进子类化：静态扫描拦截明显违规代码，子类化限制关键方法，同时通过_constructor保持子类类型；
2. 补充运行时验证：对比用户向量化输出和逐行历史计算的结果，确保逻辑合规；
3. 明确规则文档：告知用户平台的约束，提供合规示例，引导用户编写正确的代码。

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