Python之如何做交易日历（下）

weixin-npy8264926

567人浏览 · 2026-04-24 17:08:55

weixin-npy8264926 · 2026-04-24 17:08:55 发布

在 A 股量化分析与短线情绪研判中，交易日历是最基础也最核心的底层数据，而涨停连板率、破板率则是判断市场短线情绪的核心指标。市面上常用的交易日历方案存在诸多痛点：

pandas_market_calendars 对 A 股节假日、休市规则适配存在缺陷，更新不及时容易出现日期偏差

tushare 等财经数据接口需要注册账号、获取 token，且部分接口有积分门槛

手动维护交易日历费时费力，且无法实时同步交易所最新的休市安排

本文将基于免费开源的akshare财经数据接口，实现 A 股交易日历的精准获取与校验，同时搭配Streamlit搭建可视化界面，完成涨停连板率、破板率等核心情绪指标的自动计算，开箱即用，无任何积分与注册门槛。

tdxquant+miniQMT强强结合

一、环境准备

本文代码基于 Python3.8 + 版本开发，需提前安装以下依赖库，执行 pip 命令即可一键安装：

pip install streamlit akshare pandas numpy pywencai

各依赖库核心作用说明：

akshare：核心财经数据接口，免费获取 A 股交易日历、行情数据，与交易所规则实时同步
streamlit：零前端代码快速搭建 Web 可视化交互界面
pandas/numpy：标准化数据处理与数值计算
pywencai：问财数据接口，用于快速获取涨停、破板等个股明细数据

二、核心功能 1：A 股交易日历的实现与校验

交易日历的核心能力包含两点：交易日合法性校验、目标日期的前一交易日自动匹配，这是所有日度量化分析的基础。

2.1 交易日历基础数据获取

akshare 提供了新浪财经的 A 股官方交易日历接口tool_trade_date_hist_sina()，可直接获取 A 股历史全量交易日，自动剔除周末、法定节假日、临时休市日期，无需手动维护。基础代码示例：

import akshare as ak
import pandas as pd
from datetime import datetime

# 获取A股全历史交易日历
trade_date_range = ak.tool_trade_date_hist_sina()
# 日期格式统一转换为date类型，方便后续日期对比校验
trade_date_range['trade_date'] = pd.to_datetime(trade_date_range['trade_date']).dt.date

# 查看数据前5行，验证数据格式
print(trade_date_range.head())

2.2 交易日校验与前序交易日匹配函数封装

将交易日判断、前一交易日获取逻辑封装为通用函数，可直接在量化脚本中复用，避免非交易日数据为空导致的程序报错。

def check_trade_date(selected_date):
    """
    校验所选日期是否为A股交易日，并返回对应的前一交易日
    :param selected_date: 待校验的日期，datetime.date类型
    :return: is_trade_date(bool)：是否为交易日；previous_trade_date(date)：前一交易日
    """
    # 实时获取最新A股交易日历
    trade_date_range = ak.tool_trade_date_hist_sina()
    trade_date_range['trade_date'] = pd.to_datetime(trade_date_range['trade_date']).dt.date
    
    # 校验是否为交易日
    is_trade_date = selected_date in trade_date_range['trade_date'].values
    if not is_trade_date:
        return is_trade_date, None
    
    # 筛选目标日期之前的所有交易日，取最大值即为最近的前一交易日
    previous_dates = trade_date_range[trade_date_range['trade_date'] < selected_date]
    if previous_dates.empty:
        raise ValueError("未找到目标日期之前的有效交易日数据")
    previous_trade_date = previous_dates['trade_date'].max()
    
    return is_trade_date, previous_trade_date

# 函数测试
if __name__ == "__main__":
    test_date = datetime.now().date()
    is_trade, pre_date = check_trade_date(test_date)
    print(f"所选日期是否为交易日：{is_trade}")
    print(f"对应前一交易日：{pre_date}")

三、核心功能 2：涨停情绪指标计算

短线交易体系中，连板率、破板率是判断市场情绪强弱的核心指标，本文先封装基础数据获取函数，再完成指标的标准化计算。

3.1 基础数据获取函数封装

基于 pywencai 接口封装涨停、破板、昨日涨停个股数据的获取函数，返回数据标准化，无需额外字段清洗。

import pywencai
import pandas as pd

def get_limit_up_data(trade_date):
    """
    获取指定日期的涨停个股数据（剔除ST、科创板、北交所）
    :param trade_date: 目标交易日，datetime.date类型
    :return: 涨停个股DataFrame，包含股票代码、股票名称
    """
    trade_date_str = trade_date.strftime("%Y%m%d")
    # 问财查询语句，可根据自身需求修改筛选条件
    query = f"{trade_date_str} 涨停 非ST 非科创板 非北交所"
    try:
        df = pywencai.get(query=query, loop=True)
        if df is None or df.empty:
            return pd.DataFrame()
        # 统一字段名，方便后续数据处理
        df = df.rename(columns={"代码": "股票代码", "名称": "股票名称"})
        return df[["股票代码", "股票名称"]]
    except Exception as e:
        print(f"涨停数据获取失败：{str(e)}")
        return pd.DataFrame()

def get_poban_data(trade_date):
    """
    获取指定日期的曾涨停（破板）个股数据（剔除ST、科创板、北交所）
    :param trade_date: 目标交易日，datetime.date类型
    :return: 破板个股DataFrame，包含股票代码、股票名称
    """
    trade_date_str = trade_date.strftime("%Y%m%d")
    query = f"{trade_date_str} 曾涨停 非ST 非科创板 非北交所"
    try:
        df = pywencai.get(query=query, loop=True)
        if df is None or df.empty:
            return pd.DataFrame()
        df = df.rename(columns={"代码": "股票代码", "名称": "股票名称"})
        return df[["股票代码", "股票名称"]]
    except Exception as e:
        print(f"破板数据获取失败：{str(e)}")
        return pd.DataFrame()

def get_yesterday_zhangting_data(previous_date):
    """
    获取前一交易日的涨停个股数据
    :param previous_date: 前一交易日，datetime.date类型
    :return: 昨日涨停个股DataFrame
    """
    return get_limit_up_data(previous_date)

3.2 核心情绪指标计算公式与代码实现

连板率：反映昨日涨停个股的溢价能力，是短线情绪的核心正向指标计算公式：连板率 = (昨日涨停今日继续涨停的个股数量 ÷ 昨日涨停个股总数) × 100%
破板率：反映涨停板的封板强度，是短线情绪的反向指标计算公式：破板率 = (当日曾涨停但最终未封板的个股数量 ÷ (当日封死涨停个股数量 + 当日曾涨停个股数量)) × 100%

指标计算函数封装：

def calculate_emotion_index(current_date, previous_date):
    """
    计算A股短线情绪核心指标
    :param current_date: 当前分析日，datetime.date类型
    :param previous_date: 前一交易日，datetime.date类型
    :return: 字典格式，包含连板率、破板率、对应明细数据
    """
    # 获取基础明细数据
    today_zt_df = get_limit_up_data(current_date)
    yesterday_zt_df = get_yesterday_zhangting_data(previous_date)
    today_poban_df = get_poban_data(current_date)
    
    # 提取个股代码列表
    today_zt_stocks = today_zt_df['股票代码'].tolist() if not today_zt_df.empty else []
    yesterday_zt_stocks = yesterday_zt_df['股票代码'].tolist() if not yesterday_zt_df.empty else []
    poban_stocks = today_poban_df['股票代码'].tolist() if not today_poban_df.empty else []
    
    # 连板率计算
    lianban_count = len(set(yesterday_zt_stocks) & set(today_zt_stocks))
    yesterday_zt_total = len(yesterday_zt_stocks)
    lianban_rate = (lianban_count / yesterday_zt_total * 100) if yesterday_zt_total > 0 else 0
    
    # 破板率计算
    poban_total = len(poban_stocks)
    zt_total = len(today_zt_stocks)
    poban_rate = (poban_total / (poban_total + zt_total) * 100) if (poban_total + zt_total) > 0 else 0
    
    return {
        "lianban_rate": round(lianban_rate, 2),
        "poban_rate": round(poban_rate, 2),
        "today_zt_df": today_zt_df,
        "yesterday_zt_df": yesterday_zt_df,
        "today_poban_df": today_poban_df
    }

四、Streamlit 可视化界面完整实现

将上述交易日历、数据获取、指标计算能力整合，搭建可交互的 Web 可视化界面，支持日期选择、交易日自动校验、情绪指标实时计算与明细数据展示，完整可运行代码如下：

# 导入所需依赖库
import streamlit as st
import pywencai
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import akshare as ak

# ---------------------- 数据获取函数封装 ----------------------
def get_limit_up_data(trade_date):
    """获取指定日期的涨停个股数据（剔除ST、科创板、北交所）"""
    trade_date_str = trade_date.strftime("%Y%m%d")
    query = f"{trade_date_str} 涨停 非ST 非科创板 非北交所"
    try:
        df = pywencai.get(query=query, loop=True)
        if df is None or df.empty:
            return pd.DataFrame()
        df = df.rename(columns={"代码": "股票代码", "名称": "股票名称"})
        return df[["股票代码", "股票名称"]]
    except Exception as e:
        st.error(f"涨停数据获取失败：{str(e)}")
        return pd.DataFrame()

def get_poban_data(trade_date):
    """获取指定日期的曾涨停（破板）个股数据（剔除ST、科创板、北交所）"""
    trade_date_str = trade_date.strftime("%Y%m%d")
    query = f"{trade_date_str} 曾涨停 非ST 非科创板 非北交所"
    try:
        df = pywencai.get(query=query, loop=True)
        if df is None or df.empty:
            return pd.DataFrame()
        df = df.rename(columns={"代码": "股票代码", "名称": "股票名称"})
        return df[["股票代码", "股票名称"]]
    except Exception as e:
        st.error(f"破板数据获取失败：{str(e)}")
        return pd.DataFrame()

def get_yesterday_zhangting_data(previous_date):
    """获取前一交易日的涨停个股数据"""
    return get_limit_up_data(previous_date)

# ---------------------- 核心应用主函数 ----------------------
def app():
    # 页面标题
    st.title("A股涨停概念情绪分析工具")
    st.divider()

    # 日期选择交互组件
    max_date = datetime.now().date()
    selected_date = st.date_input("选择分析日期", max_value=max_date, value=max_date)
    
    # 交易日历校验
    st.info("正在校验A股交易日历数据...")
    trade_date_range = ak.tool_trade_date_hist_sina()
    trade_date_range['trade_date'] = pd.to_datetime(trade_date_range['trade_date']).dt.date

    # 非交易日拦截
    if selected_date not in trade_date_range['trade_date'].values:
        st.error("所选日期不是A股交易日，请选择其他日期！")
        return
    
    # 获取前一交易日
    previous_dates = trade_date_range[trade_date_range['trade_date'] < selected_date]
    if previous_dates.empty:
        st.error("未找到目标日期之前的有效交易日数据")
        return
    previous_date = previous_dates['trade_date'].max()
    
    # 展示分析日期区间
    st.success(f"分析日期：{selected_date} | 前一交易日：{previous_date}")
    st.divider()

    # 数据获取与指标计算
    with st.spinner("正在获取涨停数据并计算情绪指标..."):
        # 明细数据获取
        today_zt_df = get_limit_up_data(selected_date)
        yesterday_zt_df = get_yesterday_zhangting_data(previous_date)
        today_poban_df = get_poban_data(selected_date)

        # 核心指标计算
        yesterday_zt_stocks = yesterday_zt_df['股票代码'].tolist() if not yesterday_zt_df.empty else []
        today_zt_stocks = today_zt_df['股票代码'].tolist() if not today_zt_df.empty else []
        # 连板率计算
        lianban_molecule = len(set(yesterday_zt_stocks) & set(today_zt_stocks))
        lianban_denominator = len(yesterday_zt_stocks)
        lianban_rate = round((lianban_molecule / lianban_denominator * 100) if lianban_denominator > 0 else 0, 2)
        # 破板率计算
        poban_stocks = today_poban_df['股票代码'].tolist() if not today_poban_df.empty else []
        poban_molecule = len(poban_stocks)
        poban_denominator = len(poban_stocks) + len(today_zt_stocks)
        poban_rate = round((poban_molecule / poban_denominator * 100) if poban_denominator > 0 else 0, 2)

    # 情绪指标可视化展示
    st.subheader("📊 市场短线情绪核心指标")
    col1, col2 = st.columns(2)
    with col1:
        st.metric(label="连板率", value=f"{lianban_rate} %")
    with col2:
        st.metric(label="破板率", value=f"{poban_rate} %")
    st.divider()

    # 明细数据分tab展示
    st.subheader("📋 明细数据详情")
    tab1, tab2, tab3 = st.tabs(["今日涨停个股", "昨日涨停个股", "今日破板个股"])
    with tab1:
        st.dataframe(today_zt_df, use_container_width=True)
    with tab2:
        st.dataframe(yesterday_zt_df, use_container_width=True)
    with tab3:
        st.dataframe(today_poban_df, use_container_width=True)

# ---------------------- 程序入口 ----------------------
if __name__ == "__main__":
    # 页面全局配置
    st.set_page_config(layout="wide", page_title="A股情绪分析工具", page_icon="📈")
    # 启动应用
    app()