[Python技术] 清明节快到了，到底是持股过节，还是持币过节？ 用数据说话，统计近10年清明节涨跌幅情况

import streamlit as st
import akshare as ak
import pandas as pd
from datetime import datetime, date

# 页面配置
st.set_page_config(
    page_title="清明节股市效应分析",
    page_icon="🌿",
    layout="wide"
)
st.title("🌿 清明节股市效应分析")


# 核心功能模块
def qingming_date(year):
    """严格采用用户提供的天文计算方法"""
    if year == 2232:  # 特殊年份处理
        return date(year, 4, 4)

    if not 1700 <= year <= 3100:
        raise ValueError("仅支持1700-3100年份")

    # 系数数组（来自用户提供的原始数据）
    coefficient = [
        5.15, 5.37, 5.59, 4.82, 5.02, 5.26, 5.48,
        4.70, 4.92, 5.135, 5.36, 4.60, 4.81, 5.04, 5.26
    ]

    mod = year % 100  # 年份后两位
    idx = (year // 100) - 17  # 计算系数索引
    day = int(mod * 0.2422 + coefficient[idx] - mod // 4)

    # 处理可能的日期溢出
    if day > 23:  # 保证日期在4月4-6日之间
        day = day - 30
    return date(year, 4, day)


@st.cache_data
def get_trade_dates():
    """获取历史交易日历（强化格式兼容性）"""
    df = ak.tool_trade_date_hist_sina()

    # 统一处理日期格式
    trade_dates = []
    for d in df["trade_date"].astype(str):
        # 移除可能存在的分隔符
        clean_d = d.replace("-", "").replace("/", "").strip()

        try:
            # 尝试两种格式解析
            dt = datetime.strptime(clean_d, "%Y%m%d")
        except ValueError:
            try:
                dt = datetime.strptime(clean_d, "%Y%m%d")
            except:
                st.error(f"无效日期格式: {d}")
                continue
        trade_dates.append(dt.date())

    return sorted(trade_dates)


def find_nearest_trading_day(target_date, direction='before'):
    """优化后的交易日查找算法"""
    trade_dates = get_trade_dates()

    # 边界检查
    if not trade_dates:
        return None
    if target_date < trade_dates[0]:
        return None if direction == 'before'else trade_dates[0]
    if target_date > trade_dates[-1]:
        return trade_dates[-1] if direction == 'before'else None

    # 二分查找
    left, right = 0, len(trade_dates) - 1
    while left <= right:
        mid = (left + right) // 2
        if trade_dates[mid] < target_date:
            left = mid + 1
        else:
            right = mid - 1

    return trade_dates[right] if direction == 'before'else trade_dates[left]


@st.cache_data
def get_index_data(symbol="sh000001"):
    """获取指数数据（强化格式处理）"""
    df = ak.stock_zh_index_daily(symbol=symbol)

    # 统一日期处理逻辑
    df['date'] = pd.to_datetime(
        df['date'].astype(str).str.replace("-", ""),
        format="%Y%m%d",
        errors="coerce"
    ).dt.date

    # 过滤无效日期
    return df.dropna(subset=['date']).set_index('date').sort_index()


# 主计算逻辑
def calculate_results():
    results = []
    for year in range(2015, 2025):
        try:
            qingming = qingming_date(year)

            # 获取交易日
            prev_day = find_nearest_trading_day(qingming, 'before')
            next_day = find_nearest_trading_day(qingming, 'after')

            if not (prev_day and next_day):
                raise ValueError("交易日查找失败")

            # 获取价格数据
            df = get_index_data()

            # 类型安全访问
            try:
                prev_close = df.loc[prev_day, 'close']
                next_close = df.loc[next_day, 'close']
            except KeyError as e:
                raise ValueError(f"交易日数据缺失: {e}")

            # 计算涨跌幅
            change = (next_close - prev_close) / prev_close * 100
            results.append({
                "年份": year,
                "清明节": qingming.strftime("%Y-%m-%d"),
                "前交易日": prev_day.strftime("%Y-%m-%d"),
                "后交易日": next_day.strftime("%Y-%m-%d"),
                "涨跌幅(%)": round(change, 2)
            })
        except Exception as e:
            st.error(f"{year}年数据处理异常: {str(e)}")
    return pd.DataFrame(results)


# 可视化模块
def display_results(df):
    col1, col2 = st.columns([3, 2])

    with col1:
        st.subheader("📊 历史数据明细")
        st.dataframe(
            df.style.format({"涨跌幅(%)": "{:.2f}%"}).applymap(
                lambda x: 'color: #ff4444'if x >= 0 else'color: #2ecc71',
                subset=["涨跌幅(%)"]
            ),
            height=500,
            use_container_width=True
        )

    with col2:
        st.subheader("📈 趋势分析")
        st.line_chart(df.set_index("年份")["涨跌幅(%)"])

        # 统计卡片
        col2.metric("平均涨跌幅", f"{df['涨跌幅(%)'].mean():.2f}%")
        col2.metric("上涨概率", f"{(df['涨跌幅(%)'] > 0).mean():.1%}")


def app():
    with st.spinner("🔍 正在计算数据..."):
        result_df = calculate_results()

    if not result_df.empty:
        display_results(result_df)

        # 数据导出
        csv = result_df.to_csv(index=False).encode('utf-8')
        st.download_button(
            label="下载CSV数据",
            data=csv,
            file_name="qingming_effect.csv",
            mime="text/csv"
        )
    else:
        st.error("数据获取失败，请检查网络连接")

# 主程序
if __name__ == "__main__":
    app()