嘿，这个需求在量化交易或者实时数据监控里太常见了，完全不用不好意思问！咱结合你用的datetime64索引的pandas数据，一步步来实现：

核心思路 #

本质是要追踪连续满足特定逻辑的时间窗口长度，当这个窗口超过15秒时执行指定代码。我们可以用pandas的时间序列工具来高效实现，不用自己写复杂的循环。

具体步骤&代码示例 #

假设你的股票数据DataFrame叫stock_df，索引是datetime64[ns]，有一列close_price代表收盘价（你可以替换成自己的字段）。

1. 计算相邻数据点的时间间隔 #

首先，我们需要知道每两个连续数据点之间的时间差，用来累加连续满足条件的时长：

import pandas as pd

# 计算每行与上一行的时间差，结果是Timedelta类型
stock_df['time_diff'] = stock_df.index.diff()
# 第一行没有上一行，所以time_diff是NaT，我们可以填充0秒
stock_df['time_diff'] = stock_df['time_diff'].fillna(pd.Timedelta(seconds=0))

2. 标记满足特定逻辑的行 #

先把你的“特定逻辑”转换成布尔值列，比如假设你的逻辑是“收盘价低于100元”，就这么写：

# 替换成你自己的特定逻辑！比如成交量大于某值、涨跌幅超过阈值等
stock_df['condition_met'] = stock_df['close_price'] < 100

3. 分组连续满足条件的时间段 #

我们需要把连续满足条件的行归为一组，方便计算每组的持续时长：

# 用~取反后累加，这样每次条件不满足时，group_id就会加1，连续满足的行同属一个group
stock_df['group_id'] = (~stock_df['condition_met']).cumsum()

4. 计算每组的持续时长并筛选超过15秒的组 #

接下来对每个group计算总时长，然后找出符合要求的组：

# 按group_id分组，计算每个组的总持续时间和是否满足条件
group_stats = stock_df.groupby('group_id').agg(
    total_duration=('time_diff', 'sum'),
    is_valid=('condition_met', 'all')  # 确保整个组都满足条件
)

# 筛选出持续时间超过15秒的有效组
target_groups = group_stats[(group_stats['is_valid']) & (group_stats['total_duration'] > pd.Timedelta(seconds=15))]

5. 执行指定代码 #

现在我们可以针对这些目标组执行你的“指定代码”，比如打印该时间段的行情，或者触发交易信号：

# 遍历每个目标组，执行你的代码
for group_id in target_groups.index:
    # 获取该组的所有数据
    group_data = stock_df[stock_df['group_id'] == group_id]
    print(f"触发条件！时间段：{group_data.index[0]} 至 {group_data.index[-1]}，持续时长：{target_groups.loc[group_id, 'total_duration']}")
    
    # 这里替换成你的"指定代码"
    # my_specific_code(group_data)

实时数据场景的补充 #

如果是处理实时流数据（比如从API实时拉取的行情），上面的离线方法可能不太适用，你可以维护几个变量来记录状态：

• 当前是否满足条件
• 满足条件的起始时间
  每次收到新数据时：

1. 检查是否满足特定逻辑
2. 如果满足，且之前未满足：记录起始时间
3. 如果满足，且之前已满足：计算当前时间与起始时间的差，超过15秒就触发代码
4. 如果不满足：重置起始时间和状态

搜索方向建议 #

如果你想自己拓展，可以搜这些关键词：

• pandas time series consecutive condition duration
• datetime64 calculate time difference python
• real-time stock data monitoring python

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