DeepSeekv3，终于会标注二十四节气了。 - 鼎盛网

上深沪深300二十四节气.jpg (278.93 KB, 下载次数: 8)

2025-1-10 12:10 上传

作者: isaac 时间: 2025-1-10 12:11
标题: DeepSeekv3给的代码
本帖最后由 isaac 于 2025-1-10 12:13 编辑

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}")

作者: 由来有趣 时间: 2025-1-10 13:10
标题: 作为从业人员，二十年前就搞了好多这些东西了，然并卵

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

作者: 孤寂北狼 时间: 2025-1-10 15:01

作者: ddr266 时间: 2025-1-10 15:02

作者: isaac 时间: 2025-1-10 18:47
标题: 由于热力图无法复现，刚才的回复估计是错误的，为避免误导，所以删除了。
本帖最后由 isaac 于 2025-1-10 19:10 编辑

放大了p值.jpg (73.5 KB, 下载次数: 1)

下载附件

2025-1-10 19:09 上传

放大p值到0.057后，就可以再次出相关的热力图了，但这已经小于95%了。

作者: isaac 时间: 2025-1-10 23:05
标题: DeepSeekv3给出的代码，验证了节气效应后，有一个好消息和一个坏消息。
本帖最后由 isaac 于 2025-1-10 23:08 编辑

上证节气效应显著性验证.jpg (295.37 KB, 下载次数: 1)

下载附件

2025-1-10 23:06 上传

深证节气效应显著性验证.jpg (301.17 KB, 下载次数: 1)

下载附件

2025-1-10 23:06 上传

沪深300节气效应显著性验证.jpg (294.03 KB, 下载次数: 1)

下载附件

2025-1-10 23:05 上传

好消息是，将数据扩展到上证指数和深证指数这两个历史数据较多的后，证实了节气效应的存在。坏消息是在检验了效应量 (Eta-squared)后，数值非常小，是个小效应，对上证和深证相关的收益影响非常小，对沪深300有一点点影响。DeepSeekv3给的区间是节气附近前后5天。

作者: isaac 时间: 2025-1-10 23:09
标题: DeepSeekv3给的检验显著性的代码
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}")

欢迎光临鼎盛网 (http://www.top81cn.cn/)