🚀 A-Share Concept PIT Database (A股概念板块动态截面数据库)

本项目是一个全自动化的 A 股概念板块成分股数据库，专为量化投研设计。于每个交易日收盘后自动抓取增量变动，并以 Release 附件 的形式推送到本仓库，实现跨节点的无缝同步。

📊 数据结构 (Data Schema)

为了极大地压缩体积并提升查询速度，本数据库彻底摒弃了传统的“每日全量横截面”存储方式，采用了 SCD (缓慢变化维度) 区间生效结构。

核心数据文件：all_concepts_pit_scd.csv

字段名	类型	说明	示例
concept_code	str	聚宽概念板块代码	GN036
concept_name	str	概念板块名称	人工智能
stock_code	str	股票代码 (带后缀)	000001.XSHE
in_date	str	纳入日期 (生效起期)	2023-05-12
out_date	str	剔除日期 (生效止期)	2099-12-31 (代表至今有效)

💡 避坑指南： 读取 CSV 时，请务必指定 dtype={'stock_code': str, 'concept_code': str}，防止 Pandas 自动吞掉股票代码开头的 0。

---

🛠️ 快速上手 (Quick Start)

1. 自动拉取与智能缓存 (用于本地或 GPU 服务器)

在你的模型训练脚本（如 DataLoader）前加入以下代码。它自带 本地缓存检测 功能：如果今天已经下载过最新数据，则会直接使用本地文件，避免因反复跑回测而频繁发起网络请求，从而防止被 GitHub API 限制访问。

import os
import requests
import datetime

REPO_OWNER = "tycallen"                   
REPO_NAME = "jquant_data_sync"             
SAVE_PATH = "all_concepts_pit_scd.csv"  

def pull_today_data():
    # --- 本地缓存检测逻辑 ---
    if os.path.exists(SAVE_PATH):
        # 获取本地文件的最后修改日期
        file_mtime = datetime.date.fromtimestamp(os.path.getmtime(SAVE_PATH))
        today = datetime.date.today()
        # 如果文件是今天更新的，直接使用缓存，中止网络请求
        if file_mtime == today:
            print(f"✅ 发现今日缓存 ({SAVE_PATH})，跳过网络请求，极速启动。")
            return
            
    # --- 动态拼接真实下载直链 ---
    today_str = datetime.date.today().strftime('%Y-%m-%d')
    download_url = f"[https://github.com/](https://github.com/){REPO_OWNER}/{REPO_NAME}/releases/download/data-{today_str}/all_concepts_pit_scd.csv"
    
    print(f"🔄 本地无最新缓存，准备拉取今日数据...\n🔗 链接: {download_url}")
    
    try:
        # 直接请求真实文件地址，绕过 GitHub API 速率限制
        response = requests.get(download_url)
        if response.status_code == 200:
            with open(SAVE_PATH, 'wb') as f:
                f.write(response.content)
            print("✅ 数据拉取并更新成功！")
        elif response.status_code == 404:
            print("⚠️ 今日的数据尚未发布（或周末休市无更新）。")
        else:
            print(f"❌ 拉取失败，HTTP 状态码: {response.status_code}")
    except Exception as e:
        print(f"❌ 请求发生异常: {e}")

# 执行拉取/缓存检查
pull_today_data()

2. 时空穿梭查询 (Point-in-Time 还原)当数据加载到内存后，你可以使用以下逻辑，以 $O(N)$ 的极快速度还原历史上任意一天的全市场概念成分股状态，彻底杜绝未来函数：

import pandas as pd

# 1. 加载主表 (强制指定字符串类型)
df = pd.read_csv("all_concepts_pit_scd.csv", dtype={'stock_code': str, 'concept_code': str})

def get_cross_section_concepts(target_date):
    """获取指定日期当天的所有概念成分股"""
    # 核心过滤逻辑：纳入日期 <= 目标日 且 剔除日期 >= 目标日
    pit_df = df[(df['in_date'] <= target_date) & (df['out_date'] >= target_date)]
    return pit_df

# 测试：获取 2023年5月10日 的真实截面数据
target = '2023-05-10'
section_df = get_cross_section_concepts(target)

# 获取特定板块（如"人工智能"）在当天的成分股列表
ai_stocks = section_df[section_df['concept_name'] == '人工智能']['stock_code'].tolist()
print(f"{target} '人工智能'板块共有 {len(ai_stocks)} 只成分股。")