DeepSeekv3，终于会标注二十四节气了。

登录/注册后可看大图

import akshare as ak
import cnlunar
import datetime
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import pandas as pd
import logging

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']  # 设置默认字体为黑体
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 1. 获取指数数据
def get_index_data(symbol):
    try:
        today = datetime.datetime.now()  # 结束日期为今天
        start_date = datetime.datetime(today.year - 1, 1, 1)  # 起始日期为去年1月1日
        index_df = ak.stock_zh_index_daily(symbol=symbol)  # 获取指数数据

        # 检查数据是否为空
        if index_df.empty:
            raise ValueError(f"{symbol} 数据为空，请检查数据源或网络连接。")

        # 将日期列转换为 datetime 类型，并设置为索引
        index_df['date'] = pd.to_datetime(index_df['date'])
        index_df.set_index('date', inplace=True)

        # 过滤数据，确保在去年1月1日到今天之间
        index_df = index_df[(index_df.index >= start_date) & (index_df.index <= today)]
        return index_df
    except Exception as e:
        logging.error(f"获取 {symbol} 数据时出错: {e}")
        raise

# 2. 获取二十四节气日期
def get_solar_terms():
    try:
        today = datetime.datetime.now()
        start_date = datetime.datetime(today.year - 1, 1, 1)  # 起始日期为去年1月1日

        # 获取当前年份和上一年的节气数据
        solar_terms = {}
        for year in [today.year - 1, today.year]:
            date = datetime.datetime(year, 1, 1)
            lunar = cnlunar.Lunar(date, godType='8char')
            # 更新节气字典
            solar_terms.update(lunar.thisYearSolarTermsDic)

        # 将节气数据整理为 DataFrame
        solar_terms_list = []
        for term, date_tuple in solar_terms.items():
            try:
                # 将 (月, 日) 转换为完整日期
                month, day = date_tuple
                date = datetime.datetime(today.year, month, day)  # 使用当前年份
                # 如果日期超过今天，则使用上一年份
                if date > today:
                    date = datetime.datetime(today.year - 1, month, day)
                # 确保日期在去年1月1日到今天之间
                if start_date <= date <= today:
                    solar_terms_list.append({'Date': date, 'Solar Term': term})  # 确保列名为 'Date' 和 'Solar Term'
            except (TypeError, ValueError) as e:
                logging.warning(f"跳过无效日期: {date_tuple}, 错误: {e}")

        # 转换为 DataFrame
        solar_terms_df = pd.DataFrame(solar_terms_list)
        solar_terms_df.set_index('Date', inplace=True)  # 将 'Date' 列设置为索引
        return solar_terms_df
    except Exception as e:
        logging.error(f"获取二十四节气数据时出错: {e}")
        raise

# 3. 绘制指数走势图并标注节气
def plot_index_with_solar_terms(index_dfs, solar_terms_df, titles):
    try:
        fig, axes = plt.subplots(len(index_dfs), 1, figsize=(14, 10), sharex=True)
        fig.suptitle('上证、深证、沪深300指数走势图（含节气标注）', fontsize=16)

        for i, (ax, index_df, title) in enumerate(zip(axes, index_dfs, titles)):
            # 绘制指数趋势线
            ax.plot(index_df.index, index_df['close'], label=f'{title}收盘价', color='b')

            # 标记是否已添加图例
            added_trading_day = False
            added_non_trading_day = False

            # 用散点图标注节气
            for date, row in solar_terms_df.iterrows():
                # 找到指数数据中最接近的日期
                closest_date = min(index_df.index, key=lambda x: abs(x - date))

                # 判断是否为交易日
                if closest_date == date:
                    # 如果是交易日，用实心圆标注
                    if not added_trading_day:
                        ax.scatter(closest_date, index_df.loc[closest_date, 'close'], color='red', marker='o', s=100, label='节气（交易日）')
                        added_trading_day = True
                    else:
                        ax.scatter(closest_date, index_df.loc[closest_date, 'close'], color='red', marker='o', s=100)
                else:
                    # 如果不是交易日，用空心圆标注
                    if not added_non_trading_day:
                        ax.scatter(closest_date, index_df.loc[closest_date, 'close'], color='red', marker='o', s=100, facecolors='none', label='节气（非交易日）')
                        added_non_trading_day = True
                    else:
                        ax.scatter(closest_date, index_df.loc[closest_date, 'close'], color='red', marker='o', s=100, facecolors='none')

                # 标注节气名称
                ax.text(closest_date, index_df.loc[closest_date, 'close'], row['Solar Term'], fontsize=10, ha='right', va='bottom')

            # 设置子图属性
            ax.set_title(title, fontsize=14)
            ax.set_ylabel('收盘价', fontsize=12)
            ax.grid(True)
            ax.legend()

        # 设置共享的 x 轴属性
        axes[-1].set_xlabel('日期', fontsize=12)
        axes[-1].xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m'))  # 设置日期格式
        axes[-1].xaxis.set_major_locator(mdates.MonthLocator())  # 按月显示刻度
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.show()
    except Exception as e:
        logging.error(f"绘制图表时出错: {e}")
        raise

# 主程序
if __name__ == "__main__":
    try:
        # 获取数据
        sh_df = get_index_data("sh000001")  # 上证指数
        sz_df = get_index_data("sz399001")  # 深证成指
        hs300_df = get_index_data("sh000300")  # 沪深300指数
        solar_terms_df = get_solar_terms()

        # 打印数据以检查
        logging.info("上证指数数据：")
        logging.info(sh_df.head())
        logging.info("深证成指数据：")
        logging.info(sz_df.head())
        logging.info("沪深300指数数据：")
        logging.info(hs300_df.head())
        logging.info("\n二十四节气数据：")
        logging.info(solar_terms_df)

        # 绘制走势图并标注节气
        plot_index_with_solar_terms(
            [sh_df, sz_df, hs300_df],  # 指数数据列表
            solar_terms_df,  # 节气数据
            ["上证指数", "深证成指", "沪深300指数"]  # 图表标题列表
        )
    except Exception as e:
        logging.error(f"程序运行出错: {e}")

isaac 发表于 2025-1-10 12:11

登录/注册后可看大图

登录/注册后可看大图

import akshare as ak
import cnlunar
import datetime
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy import stats
import scikit_posthocs as sp
import seaborn as sns
import time

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']  # 设置默认字体为黑体
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题

# 1. 获取指数历史数据
def get_index_data(index_symbol, max_retries=3):
    for i in range(max_retries):
        try:
            # 获取指数数据
            index_df = ak.stock_zh_index_daily(symbol=index_symbol)
            if not index_df.empty:
                # 将日期列转换为 datetime 类型，并设置为索引
                index_df['date'] = pd.to_datetime(index_df['date'])
                index_df.set_index('date', inplace=True)
                # 计算每日收益率
                index_df['return'] = index_df['close'].pct_change()
                return index_df
        except Exception as e:
            print(f"获取数据失败，重试 {i + 1}/{max_retries}: {e}")
            time.sleep(2)  # 等待2秒后重试
    raise ValueError(f"获取指数 {index_symbol} 数据失败，请检查网络或数据源。")

# 2. 获取指数交易日期范围内的二十四节气日期
def get_solar_terms(index_df):
    # 获取指数的最早和最晚日期
    start_date = index_df.index.min()
    end_date = index_df.index.max()
    solar_terms_list = []

    for year in range(start_date.year, end_date.year + 1):
        date = datetime.datetime(year, 1, 1)
        lunar = cnlunar.Lunar(date, godType='8char')
        for term, date_tuple in lunar.thisYearSolarTermsDic.items():
            month, day = date_tuple
            date = datetime.datetime(year, month, day)
            # 确保节气日期在指数的交易日期范围内
            if start_date <= date <= end_date:
                # 找到最近的交易日
                closest_date = pd.bdate_range(end=date, periods=1)[0]
                solar_terms_list.append({'Date': closest_date, 'Solar Term': term})

    # 转换为 DataFrame
    solar_terms_df = pd.DataFrame(solar_terms_list)
    solar_terms_df.set_index('Date', inplace=True)
    solar_terms_df = solar_terms_df.sort_index()
    return solar_terms_df

# 3. 划分各节气区间
def split_solar_periods(index_df, solar_terms_df, window=5):
    # 标记各节气区间
    index_df['solar_term'] = None
    for date, row in solar_terms_df.iterrows():
        start_date = date - pd.Timedelta(days=window)
        end_date = date + pd.Timedelta(days=window)
        index_df.loc[start_date:end_date, 'solar_term'] = row['Solar Term']

    # 过滤出节气区间的数据
    solar_periods = index_df.dropna(subset=['solar_term'])
    return solar_periods

# 4. 显著性检验和事后检验
def perform_significance_test(solar_periods):
    # 检查每组数据量是否足够
    groups = [group['return'].dropna() for _, group in solar_periods.groupby('solar_term')]
    if any(len(group) < 3 for group in groups):
        print("某些节气区间的数据量不足，无法进行显著性检验")
        return None, None, None

    # 使用 Kruskal-Wallis 检验
    stat, p = stats.kruskal(*groups)
    print(f"Kruskal-Wallis 检验: 统计量={stat:.4f}, p 值={p:.4f}")

    # 计算效应量 (Eta-squared)
    total_samples = sum(len(group) for group in groups)
    eta_squared = stat / (total_samples - 1)  # Eta-squared 公式
    print(f"效应量 (Eta-squared): {eta_squared:.4f}")

    # 判断显著性
    if p < 0.05:
        print("\n各节气区间之间存在显著差异")
        # 进行 Dunn 事后检验
        dunn_result = sp.posthoc_dunn(solar_periods.dropna(), val_col='return', group_col='solar_term')
        print("\nDunn 事后检验结果：")
        print(dunn_result)
        return dunn_result, p, eta_squared  # 返回 Dunn 结果、p 值和效应量
    else:
        print("\n各节气区间之间不存在显著差异")
        return None, p, eta_squared  # 返回 p 值和效应量

# 5. 将显著性关系保存到 Excel 文件
def save_significance_to_excel(dunn_result, index_name, p_value, eta_squared, filename_prefix="significance_results"):
    if dunn_result is not None:
        filename = f"{filename_prefix}_{index_name}.xlsx"
        with pd.ExcelWriter(filename) as writer:
            dunn_result.to_excel(writer, sheet_name="Dunn 事后检验结果")
            # 保存 p 值和效应量
            summary_df = pd.DataFrame({
                "Kruskal-Wallis p 值": [p_value],
                "效应量 (Eta-squared)": [eta_squared]
            })
            summary_df.to_excel(writer, sheet_name="显著性检验总结", index=False)
        print(f"\n显著性关系已保存到文件: {filename}")
    else:
        print("\n无显著性关系可保存")

# 6. 绘制显著性关系的热力图（上三角）并标注显著性验证方法和结果
def plot_significance_heatmap(dunn_result, index_name, p_value, eta_squared):
    if dunn_result is not None:
        # 创建上三角矩阵
        mask = np.triu(np.ones_like(dunn_result, dtype=bool))

        # 绘制热力图
        plt.figure(figsize=(12, 10))
        ax = sns.heatmap(
            dunn_result,
            mask=mask,
            annot=True,
            fmt=".3f",
            cmap="viridis",
            vmin=0,
            vmax=1,
            linewidths=0.5,
            cbar_kws={"shrink": 0.8}
        )

        # 添加显著性验证方法和结果的说明
        plt.text(
            0.5, -0.2,  # 文字位置
            "显著性验证方法：\n"
            "1. 使用 Kruskal-Wallis 检验判断各节气区间是否存在显著差异。\n"
            "2. 若 Kruskal-Wallis 检验的 p 值 < 0.05，则使用 Dunn 事后检验确定具体差异。\n"
            "3. 热力图中展示 Dunn 事后检验的 p 值，颜色越深表示差异越显著。\n"
            f"4. 效应量 (Eta-squared): {eta_squared:.4f}",
            ha='center', va='center', transform=ax.transAxes, fontsize=12
        )

        # 设置标题和标签
        title = (
            f"{index_name} 各节气区间显著性关系热力图（上三角）\n"
            f"Kruskal-Wallis 检验的 p 值为: {p_value:.4f}\n"
            f"效应量 (Eta-squared): {eta_squared:.4f}"
        )
        plt.title(title, fontsize=16, pad=20)
        plt.xlabel("节气区间", fontsize=14)
        plt.ylabel("节气区间", fontsize=14)
        plt.tight_layout()
        plt.show()
    else:
        print("\n无显著性关系可绘制热力图")

# 7. 主程序
def analyze_index(index_symbol, index_name, window=5):
    try:
        # 获取数据
        index_df = get_index_data(index_symbol)
        solar_terms_df = get_solar_terms(index_df)

        # 划分各节气区间
        solar_periods = split_solar_periods(index_df, solar_terms_df, window=window)

        # 打印数据以检查
        print(f"\n{index_name} 各节气区间数据：")
        print(solar_periods.groupby('solar_term')['return'].describe())

        # 显著性检验和事后检验
        dunn_result, p_value, eta_squared = perform_significance_test(solar_periods)

        # 输出 Kruskal-Wallis 检验的 p 值和效应量
        print(f"\n{index_name} Kruskal-Wallis 检验的 p 值为: {p_value:.4f}")
        print(f"{index_name} 效应量 (Eta-squared): {eta_squared:.4f}")

        # 将显著性关系保存到 Excel 文件
        if dunn_result is not None:
            save_significance_to_excel(dunn_result, index_name, p_value, eta_squared)

        # 绘制显著性关系的热力图（上三角）
        if dunn_result is not None:
            plot_significance_heatmap(dunn_result, index_name, p_value, eta_squared)

    except ValueError as ve:
        print(f"数据获取或处理错误: {ve}")
    except Exception as e:
        print(f"程序运行出错: {e}")

# 8. 批量分析多个指数
if __name__ == "__main__":
    # 定义指数列表（指数代码和名称）
    index_list = [
        {"symbol": "sh000001", "name": "上证指数"},
        {"symbol": "sz399001", "name": "深证成指"},
        {"symbol": "sh000300", "name": "沪深300"}
    ]

    # 分析每个指数
    for index in index_list:
        print(f"\n开始分析指数: {index['name']} ({index['symbol']})")
        analyze_index(index["symbol"], index["name"])
        print(f"\n{index['name']} 分析完成！\n{'-' * 50}")