【量化专栏 03】捕捉主力资金的起爆点：放量突破策略的代码实现

import akshare as ak
import pandas as pd

def calculate_breakout_signals(symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
    """ 计算股票的放量突破信号 """
    print(f"正在拉取 {symbol} 的行情数据进行突破信号计算...")
    df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq")

    if df is None or df.empty:
        raise ValueError(f"数据获取异常！请核查股票代码：{symbol}")

    df = df.rename(columns={'日期': 'date', '收盘': 'close', '最高': 'high', '成交量': 'volume'})
    df['date'] = pd.to_datetime(df['date'])
    df.set_index('date', inplace=True)

    # ---------------- 核心信号逻辑 ----------------
    # 1. 计算价格突破基准线：过去 20 个交易日的最高价
    # shift(1) 确保从昨天开始往前推，rolling(20).max() 求极大值
    df['past_20d_high'] = df['high'].shift(1).rolling(window=20).max()

    # 2. 计算放量基准线：过去 5 个交易日的平均成交量
    df['past_5d_vol_mean'] = df['volume'].shift(1).rolling(window=5).mean()

    # 3. 生成布尔条件信号
    condition_price = df['close'] > df['past_20d_high']
    condition_volume = df['volume'] > (df['past_5d_vol_mean'] * 2)

    # 4. 信号合并
    df['buy_signal'] = condition_price & condition_volume
    
    return df.dropna()

# === 测试环节 ===
if __name__ == "__main__":
    # 以 浪潮信息 (000977) 为例，寻找其在 2023 年的爆发点
    df_breakout = calculate_breakout_signals("000977", "20230101", "20231231")
    signal_days = df_breakout[df_breakout['buy_signal'] == True]
    print(f"信号触发日：\n{signal_days[['close', 'volume', 'buy_signal']]}")