按股票Symbol分组计算股价Pivot Point（前后10值局部高点）实现咨询

实现思路 #

• 先对原始数据做预处理：将日期转为时间格式，按股票代码+日期升序排序，保证单只股票的时间序列顺序正确
• 按Symbol分组后，对每只股票的收盘价用中心滚动窗口做判断：窗口大小为21（当前日+前10个交易日+后10个交易日），如果当日收盘价是窗口内最大值，即为符合要求的局部高点
• 新增要求的两个输出字段：High Points存储符合条件的收盘价，Pivot Point存储布尔标记表示是否为局部高点
• 最终筛选出标记为True的记录即可得到目标输出

完整可运行代码 #

import pandas as pd
import numpy as np

# 读取数据，vol_list为你已有的成交量筛选列表
df_prices = pd.read_csv('/Users/kylemerrick/Desktop/Stock Screener/price_data.csv')
include_pivot_points = df_prices[df_prices['Symbol'].isin(vol_list)].copy()

# 预处理：日期格式转换 + 按股票+日期升序排序，这步是核心前提，否则前后交易日判断完全错误
include_pivot_points['Date'] = pd.to_datetime(include_pivot_points['Date'])
include_pivot_points = include_pivot_points.sort_values(by=['Symbol', 'Date'], ascending=[True, True]).reset_index(drop=True)

n = 10 # 前后各对比10个交易日
# 定义局部高点判断函数
def is_pivot_point(close_series):
    # 中心滚动窗口，大小覆盖当前+前n+后n共2n+1天，要求窗口数据满才判断，避免边界误判
    rolling_max = close_series.rolling(window=2*n+1, center=True, min_periods=2*n+1).max()
    return close_series == rolling_max

# 分组计算标记
include_pivot_points['Pivot Point'] = include_pivot_points.groupby('Symbol')['Close'].apply(is_pivot_point).reset_index(drop=True)
# 生成High Points字段
include_pivot_points['High Points'] = np.where(include_pivot_points['Pivot Point'], include_pivot_points['Close'], np.nan)

# 提取最终结果（仅保留符合条件的记录，如需保留所有行直接用include_pivot_points即可）
final_output = include_pivot_points.loc[include_pivot_points['Pivot Point'], ['Date', 'Symbol', 'Close', 'Volume', 'High Points', 'Pivot Point']]

# 查看结果
print(final_output)

说明 #

• 如果你需要允许收盘价等于前后n天的收盘价也算符合条件，把判断条件里的==改为>=即可
• 边界的前10条和后10条数据因为没有足够的前后对比数据，不会被标记为Pivot Point，符合实际业务逻辑

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