diff --git a/scripts/data_platform/migrate_constituent.py b/scripts/data_platform/migrate_constituent.py index 2ce65dc..16cffc2 100644 --- a/scripts/data_platform/migrate_constituent.py +++ b/scripts/data_platform/migrate_constituent.py @@ -172,8 +172,14 @@ def migrate(db_path: str) -> None: df_deep["code"] = df_deep["code"].apply(norm_code) print(f"[深证 union] rows={len(df_deep)}") - # 3. 中证 1000/2000 announce_union 全集(优先); 缺则回退 snapshot 当前 - ANNOUNCE_IDX = ["000852", "932000"] + # 3. 中证 announce_union 全集(优先, 自动发现 *_announce_union.parquet); 缺则回退 snapshot + # G1+G2 扩展: 不再硬编码 000852/932000, glob 整个 HIST 目录自动发现新指数 + announce_files = sorted(glob.glob(os.path.join(HIST, "*_announce_union.parquet"))) + ANNOUNCE_IDX = [ + os.path.basename(f).split("_announce_union.parquet")[0] + for f in announce_files + ] + print(f"[中证 announce] 发现 {len(ANNOUNCE_IDX)} 个指数: {ANNOUNCE_IDX}") df_csi = _read_announce_union_aggregated(HIST, ANNOUNCE_IDX) if df_csi.empty: # 回退: 旧 snapshot-only 路径(announce_union 未生成时) diff --git a/scripts/data_platform/parse_csindex_announce.py b/scripts/data_platform/parse_csindex_announce.py index c65ebff..9884c60 100644 --- a/scripts/data_platform/parse_csindex_announce.py +++ b/scripts/data_platform/parse_csindex_announce.py @@ -369,15 +369,23 @@ def parse_xlsx_adjustments(path: Path, target_index_code: str = "000852") -> Tup def parse_pdf_adjustments(path: Path, target_section: str = "中证1000") -> Tuple[List[dict], List[dict]]: """解析 PDF 的指定指数 section - target_section: '中证1000' or '中证2000' + target_section: '中证1000' / '中证2000' / '中证全指' / '中证能源' 等 返 (add_rows, remove_rows) + + header_re 已泛化: 支持任意"中证XXX"/"沪深XXX"/"上证XXX"/"深证XXX"/"中证N位数字" 等指数简称。 + target_section 精确匹配 group(1)。 """ with pdfplumber.open(path) as pdf: full_text = "\n".join((p.extract_text() or "") for p in pdf.pages) - # 定位所有 section header + # 定位所有 section header (泛化: 数字代号 + 中文简称均可) header_re = re.compile( - r"(沪深300|中证500|中证1000|中证2000|中证A\d+|上证\d+|科创50|北证\d+)\s*指数样本调整名单[:：]?" + r"(沪深300|沪深[一-龥]{1,10}|" + r"中证\d{1,4}|中证A\d+|中证[一-龥]{1,10}|" + r"上证\d+|上证[一-龥]{1,10}|" + r"深证\d+|深证[一-龥]{1,10}|" + r"科创50|北证\d+|国证\d+|国证[一-龥]{1,10})" + r"\s*指数样本调整名单[:：]?" ) headers = list(header_re.finditer(full_text)) if not headers: @@ -551,6 +559,101 @@ def parse_launch_xlsx(path: Path, index_code: str, publish_date: str, notice_id: return out +# ======================== 通用指数处理 (G1+G2 扩展) ======================== +# 内置指数 -> PDF section 名映射 (用于 parse_pdf_adjustments 精确定位 section) +# 实证 csindex 公告系统 PDF section 用指数简称 (无"指数"后缀), 如 "中证1000 指数样本调整名单" +# 行业指数系列简称实证: akshare 返回 "800能源/800材料/..." (csindex PDF section 多用"中证800能源"全称) +INDEX_SECTION_MAP: Dict[str, str] = { + # 已有 (CSI 1000/2000 走 main --only 路径, 不走 --indices) + "000852": "中证1000", + "932000": "中证2000", + # G1: 中证全指 (全市场池, 解锁策略 02) + "000985": "中证全指", + # G2: 中证一级行业指数 (行业轮动, 解锁策略 03) - 中证 800 行业指数系列 + "000928": "中证800能源", + "000929": "中证800材料", + "000930": "中证800工业", + "000931": "中证800可选", + "000932": "中证800消费", + "000933": "中证800医药", + "000934": "中证800金融", + "000935": "中证800信息", + "000936": "中证800通信", + "000937": "中证800公用", + # 000938 不是行业指数 (中证民企ESG 50 等), 留作可选 +} + + +def process_generic_index( + idx_code: str, + target_section: str, + out_dir: Path, + cache_dir: Path, + refresh_index: bool = False, + max_notices: int = 0, +) -> pd.DataFrame: + """通用指数处理流程 (复用 CSI 1000 的 4 步逻辑, 幂等可重跑): + + 1. fetch_notices_by_index(idx_code) 拉公告列表 + 2. filter_adjustment_notices 筛 theme=指数调样 + 3. process_notice 解析 PDF/xlsx (target_section 定位 PDF section) + - max_notices > 0 时, 只处理前 N 条 (按 publishDate 升序, 最近的 N 条), + 避免 000985 等公告多的指数跑数小时 (PDF section 多不匹配, 治偏差增益微小) + 4. fetch_akshare_current 加 current 行 (兜底治偏差: 即使无公告也有当前快照) + + 返回 announce_union DataFrame (也写 {idx_code}_announce_union.parquet) + """ + detail_cache = cache_dir / "detail" + file_cache = cache_dir / "files" + + log.info(f"\n========== generic index: {idx_code} (section={target_section}) ==========") + + # 1. indexCode 公告列表 + notices_idx = fetch_notices_by_index( + idx_code, cache_dir / f"notices_{idx_code}.json", force=refresh_index, + ) + adj = filter_adjustment_notices(notices_idx) + log.info(f" indexCode={idx_code} theme=指数调样 notices: {len(adj)}") + + # 2. 逐公告解析 (PDF/xlsx), max_notices 截断 + if max_notices > 0 and len(adj) > max_notices: + # 保留最近 max_notices 条 (adj 已按 publishDate 升序, 取末尾即最新) + adj_trimmed = adj[-max_notices:] + log.info(f" max_notices={max_notices}: 截断 {len(adj)} -> {len(adj_trimmed)} (保留最新)") + adj = adj_trimmed + + records: List[dict] = [] + for x in adj: + nid = x["id"] + try: + recs = process_notice(nid, detail_cache, file_cache, idx_code, target_section) + records.extend(recs) + except Exception as e: + log.error(f" {idx_code} nid={nid} FAILED: {e}") + log.info(f" after notices: {len(records)} records " + f"(add={sum(1 for r in records if r['adjust_type']=='add')}, " + f"remove={sum(1 for r in records if r['adjust_type']=='remove')})") + + # 3. akshare 当前快照 (兜底, 保证至少有 current 行) + try: + records.extend(fetch_akshare_current(idx_code)) + except Exception as e: + log.warning(f" akshare {idx_code} current 失败 (非致命): {e}") + + # 4. 写 parquet + df = pd.DataFrame(records, columns=[ + "updateDate", "index_code", "code", "code_name", "adjust_type", "notice_id", "source"]) + out_path = out_dir / f"{idx_code}_announce_union.parquet" + df.to_parquet(out_path, index=False) + distinct = df["code"].nunique() if len(df) else 0 + n_add = (df["adjust_type"] == "add").sum() if len(df) else 0 + n_rem = (df["adjust_type"] == "remove").sum() if len(df) else 0 + n_cur = (df["adjust_type"] == "current").sum() if len(df) else 0 + log.info(f" {idx_code} result: {out_path.name} total={len(df)} " + f"(add={n_add}, remove={n_rem}, current={n_cur}, distinct={distinct})") + return df + + def main(): parser = argparse.ArgumentParser() parser.add_argument("--out-dir", default=str(OUT_DIR)) @@ -558,6 +661,16 @@ def main(): parser.add_argument("--refresh-list", action="store_true", help="强制重新拉全量公告列表") parser.add_argument("--refresh-index", action="store_true", help="强制重新拉 indexCode 公告列表") parser.add_argument("--only", choices=["1000", "2000", "both"], default="both") + parser.add_argument( + "--indices", + help="通用模式: 逗号分隔的指数代码 (如 000985,000928~000937), " + "用 INDEX_SECTION_MAP 处理 (代替默认 1000/2000 硬编码路径)", + ) + parser.add_argument( + "--max-notices-per-index", type=int, default=0, + help="通用模式: 每指数最多处理的公告数 (0=不限, >0 取最新 N 条). " + "避免 000985 公告多的指数跑数小时", + ) args = parser.parse_args() out_dir = Path(args.out_dir) @@ -566,6 +679,31 @@ def main(): detail_cache = cache_dir / "detail" file_cache = cache_dir / "files" + # ===== 通用模式: --indices ===== + if args.indices: + codes = [c.strip() for c in args.indices.split(",") if c.strip()] + log.info(f"通用模式: indices={codes}, max_notices_per_index={args.max_notices_per_index}") + for code in codes: + section = INDEX_SECTION_MAP.get(code) + if not section: + log.warning(f" {code} 不在 INDEX_SECTION_MAP, 跳过 (请补充映射)") + continue + try: + process_generic_index( + code, section, out_dir, cache_dir, + refresh_index=args.refresh_index, + max_notices=args.max_notices_per_index, + ) + except Exception as e: + log.error(f" {code} 通用处理 FAILED: {e}") + # 总结 + log.info("\n========== DONE (generic) ==========") + log.info(f"OUT_DIR: {out_dir.resolve()}") + for f in sorted(out_dir.glob("*_announce_union.parquet")): + log.info(f" {f.name}: {f.stat().st_size} bytes") + return + + # ===== 默认模式: CSI 1000/2000 (硬编码, 不动) ===== # 1. 拉/缓存全量列表(兜底) + indexCode 精准列表(主源) notices = fetch_all_notices(cache_dir / "all_notices.json", force=args.refresh_list) notices_000852 = fetch_notices_by_index(