[Python技术] 统计个股股性包括涨停次数、首次涨停溢价率、 高开低走次数

import akshare as ak
import pandas as pd

# 获取个股历史数据（示例代码：宁德时代300750）
df = ak.stock_zh_a_hist(symbol="300750", period="daily", start_date="20240701", end_date="20250701")
# 保留核心列：日期、开盘价、收盘价、涨跌幅
df = df[["日期", "开盘", "收盘", "涨跌幅"]]
df["日期"] = pd.to_datetime(df["日期"])
df.set_index("日期", inplace=True)

# 计算涨停次数（假设非ST股）
df["is_limit_up"] = df["涨跌幅"] >= 9.9  # 考虑浮动误差
limit_up_count = df["is_limit_up"].sum()
print(f"近1年涨停次数：{limit_up_count}")

# 计算前日收盘价（shift平移数据）
df["prev_close"] = df["收盘"].shift(1)
# 判断高开：开盘价 > 前日收盘价 * 1.01
df["is_high_open"] = df["开盘"] > df["prev_close"] * 1.01
# 判断低走：收盘价 < 开盘价
df["is_low_close"] = df["收盘"] < df["开盘"]
# 统计高开低走
df["high_open_low_close"] = df["is_high_open"] & df["is_low_close"]
high_open_low_close_count = df["high_open_low_close"].sum()
print(f"高开低走次数：{high_open_low_close_count}")

import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def calculate_limit_price(row, stock_code):
    """
    计算涨停价（考虑不同板块的涨跌幅限制）
    参数:
    row : DataFrame行
    stock_code : 股票代码
    返回:
    float : 涨停价
    """
    # 根据股票代码确定涨跌幅限制
    if str(stock_code).startswith(('68', '30')):  # 科创板和创业板
        limit_rate = 0.20
    elif str(stock_code).startswith(('8')):  # 北交所
        limit_rate = 0.30
    elif 'ST' in row.get('name', '') or '*ST' in row.get('name', ''):  # ST股票
        limit_rate = 0.05
    else:  # 主板股票
        limit_rate = 0.10
    # 计算涨停价（四舍五入到分）
    limit_price = round(row['prev_close'] * (1 + limit_rate), 2)
    return limit_price
def analyze_stock_behavior(symbol, start_date, end_date):
    """
    分析股票行为：涨停次数、高开低走次数、首板次日未突破情况
    参数:
    symbol : 股票代码
    start_date : 开始日期 (YYYYMMDD)
    end_date : 结束日期 (YYYYMMDD)
    返回:
    tuple : (涨停次数, 高开低走次数, 首板次日未突破次数, 包含分析结果的DataFrame)
    """
    # 获取股票基本信息（包含股票名称）
    #stock_info = ak.stock_info_a_code_name()
    #stock_name = stock_info[stock_info['code'] == symbol]['name'].values[0]
    # 获取历史行情数据
    df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
                            start_date=start_date, end_date=end_date,
                            adjust="qfq")  # 使用前复权数据
    # 数据预处理
    df = df[["日期", "开盘", "最高", "最低", "收盘", "涨跌幅"]]
    #print(df)
    df["日期"] = pd.to_datetime(df["日期"])
    df = df.sort_values("日期").reset_index(drop=True)
    # 计算前日收盘价
    df["prev_close"] = df["收盘"].shift(1)
    # 计算涨停价
    df["涨停价"] = df.apply(lambda row: calculate_limit_price(
        {'prev_close': row['prev_close']}, symbol), axis=1)
    # 涨停判断（考虑四舍五入误差）
    df["is_limit_up"] = (df["最高"] >= df["涨停价"] - 0.01) | (df["涨跌幅"] >= 9.5)
    # 高开低走判断
    df["is_high_open"] = df["开盘"] > df["prev_close"] * 1.03  # 高开3%以上
    df["is_low_close"] = df["收盘"] < df["开盘"] * 0.97   # 收盘 低于开盘
    df["high_open_low_close"] = df["is_high_open"] & df["is_low_close"]
    # 首板涨停次日未突破判断
    df["is_first_limit"] = False
    df["next_day_not_break"] = False
    for i in range(1, len(df) - 1):
        # 判断首板涨停：当日涨停且前日未涨停
        if df.loc[i, "is_limit_up"] and not df.loc[i - 1, "is_limit_up"]:
            df.loc[i, "is_first_limit"] = True
            # 判断次日最高价是否未突破涨停价
            if df.loc[i + 1, "最高"] < df.loc[i, "涨停价"]:
                df.loc[i + 1, "next_day_not_break"] = True
    # 统计结果
    limit_up_count = df["is_limit_up"].sum()
    high_open_low_close_count = df["high_open_low_close"].sum()
    next_day_not_break_count = df["next_day_not_break"].sum()
    return limit_up_count, high_open_low_close_count, next_day_not_break_count, df
# 示例使用
if __name__ == "__main__":
    # 设置分析参数
    symbol = "002549"
    end_date = datetime.now().strftime("%Y%m%d")
    start_date = (datetime.now() - timedelta(days=365)).strftime("%Y%m%d")
    # 执行分析
    limit_up, high_open_low, next_day_not_break, result_df = analyze_stock_behavior(
        symbol, start_date, end_date
    )
    # 打印结果
    print(f"分析时段: {start_date} 至 {end_date}")
    print(f"涨停次数: {limit_up}")
    print(f"高开低走次数: {high_open_low}")
    print(f"首板涨停次日未突破次数: {next_day_not_break}")
    # 查看首板次日未突破的案例
    if next_day_not_break > 0:
        print("\n首板涨停次日未突破案例:")
        cases = result_df[result_df["next_day_not_break"]]
        print(cases[["日期", "开盘", "最高", "最低", "收盘", "涨停价"]])