# -*- coding: utf-8 -*- """ 全市场个股目录（只读诊断脚本，不属于项目流水线） 目的：生成一份覆盖"全部个股"的完整目录 + 分类表，不改变任何既有筛选逻辑。 既有的 basic_pool_eligible / fundamental_eligible / candidate_tier 原样保留， 本脚本只是把它们并列展示，并标注每只股票"被哪一条卡掉"。 输出（写到 data/processed/ 下，文件名带 full_market_ 前缀）： full_market_catalog.csv 带 BOM，Excel 可直接打开 full_market_catalog.parquet 用法： python deploy/build_full_market_catalog.py """ from __future__ import annotations import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "vendor")) sys.path.insert(0, str(ROOT / "src")) import pandas as pd # noqa: E402 PROCESSED = ROOT / "data" / "processed" def _yes_no(series: pd.Series) -> pd.Series: return series.fillna(False).astype(bool) def main() -> int: master_path = PROCESSED / "security_master.parquet" scores_path = PROCESSED / "fundamental_scores.parquet" cand_path = PROCESSED / "stock_screen_candidates.parquet" for p in (master_path, scores_path, cand_path): if not p.exists(): print(f"[ERROR] 缺少输入：{p}", file=sys.stderr) return 1 master = pd.read_parquet(master_path) scores = pd.read_parquet(scores_path) cand = pd.read_parquet(cand_path) # ---- 从候选表取分级与评分（它已含全部 5218 只）-------------------- keep = [ "source_code", "candidate_tier", "candidate_tier_order", "candidate_eligible", "fundamental_eligible", "is_st_current", "total_score", "raw_total_score", "risk_penalty", "hard_risk_count", "warning_count", "score_coverage", "financial_health_status", "fundamental_grade", "is_financial_sector", "hotspot_score", "hotspot_rank", "is_hot", "combined_score", "industry_l1_code", "industry_l1_name", "industry_l2_code", "industry_l2_name", "market_cap", "pe_ttm", "pb", "latest_close", "name", ] keep = [c for c in keep if c in cand.columns] c = cand[keep].copy() # ---- 并上基础池的判定条件（含未进入候选表的兜底）-------------------- mcols = [ "source_code", "avg_amount_20", "record_count", "first_date", "last_date", "business_days_stale", "is_name_missing", "record_size_valid", "latest_ohlc_valid", "is_current_to_market_cutoff", "listed_250_days", "liquidity_pass", "has_industry", "has_finance", "basic_pool_eligible", ] mcols = [c2 for c2 in mcols if c2 in master.columns] m = master[mcols].copy() df = m.merge(c, on="source_code", how="outer", indicator=True) # 行业名称优先用 master 的（更全） for col in ("industry_l1_name", "industry_l2_name"): a = f"{col}_m" if col in master.columns: df = df.merge(master[["source_code", col]].rename(columns={col: a}), on="source_code", how="left") df[col] = df[a].combine_first(df.get(col)) df = df.drop(columns=[a]) # ---- 标注排除原因（逐条，可多条并存）-------------------------------- reasons = [] def add(mask: pd.Series, label: str) -> None: mask = _yes_no(mask) if mask.any(): reasons.append(pd.Series(label, index=df.index[mask])) not_basic = ~_yes_no(df.get("basic_pool_eligible")) add(not_basic & _yes_no(df.get("is_name_missing")), "基础池-名称缺失") add(not_basic & _yes_no(df.get("is_st_current")), "基础池-ST或退市") add(not_basic & ~_yes_no(df.get("record_size_valid")), "基础池-日线记录尺寸异常") add(not_basic & ~_yes_no(df.get("latest_ohlc_valid")), "基础池-最新OHLC异常") add(not_basic & ~_yes_no(df.get("is_current_to_market_cutoff")), "基础池-价格未跟上行情截面") add(not_basic & ~_yes_no(df.get("listed_250_days")), "基础池-上市不足250交易日") add(not_basic & ~_yes_no(df.get("liquidity_pass")), "基础池-20日均额低于门槛") add(not_basic & ~_yes_no(df.get("has_industry")), "基础池-缺行业映射") add(not_basic & ~_yes_no(df.get("has_finance")), "基础池-缺财务数据") no_fund = _yes_no(df.get("basic_pool_eligible")) & ~_yes_no(df.get("fundamental_eligible")) add(no_fund & _yes_no(df.get("is_financial_sector")), "基本面-金融行业走专用模型") add(no_fund & df.get("hard_risk_count", pd.Series(0, index=df.index)).fillna(0).gt(0), "基本面-触发财务硬伤") add(no_fund & df.get("score_coverage", pd.Series(1, index=df.index)).fillna(0).lt(0.6), "基本面-评分覆盖率不足") add(no_fund & df.get("total_score", pd.Series(0, index=df.index)).lt(65).fillna(False), "基本面-总分低于65") if reasons: joined = pd.concat(reasons).groupby(level=0).apply(lambda s: "；".join(sorted(set(s)))) df["exclude_reasons"] = joined.reindex(df.index).fillna("") else: df["exclude_reasons"] = "" df.loc[_yes_no(df.get("candidate_eligible")), "exclude_reasons"] = "" df["in_full_market"] = True df["in_basic_pool"] = _yes_no(df.get("basic_pool_eligible")) df["in_fundamental_eligible"] = _yes_no(df.get("fundamental_eligible")) df["in_candidate_pool"] = _yes_no(df.get("candidate_eligible")) # ---- 确定性分级：所有个股必有且仅有一个层级 -------------------------- df["catalog_tier"] = "其他（数据不完整）" df.loc[df["in_basic_pool"], "catalog_tier"] = "仅入基础池" df.loc[df["in_fundamental_eligible"], "catalog_tier"] = "基本面合格" df.loc[df["candidate_tier"].isin(["观察", "基本面精选", "热点交集", "优先研究"]), "catalog_tier"] = df.loc[df["candidate_tier"].isin(["观察", "基本面精选", "热点交集", "优先研究"]), "candidate_tier"] order = {"优先研究": 0, "热点交集": 1, "基本面精选": 2, "观察": 3, "基本面合格": 4, "仅入基础池": 5, "其他（数据不完整）": 6} df["catalog_order"] = df["catalog_tier"].map(order).fillna(9).astype(int) out_cols = [ "source_code", "name", "catalog_tier", "exclude_reasons", "in_full_market", "in_basic_pool", "in_fundamental_eligible", "in_candidate_pool", "candidate_tier", "is_st_current", "is_financial_sector", "industry_l1_name", "industry_l2_name", "latest_close", "avg_amount_20", "record_count", "first_date", "last_date", "business_days_stale", "market_cap", "pe_ttm", "pb", "total_score", "raw_total_score", "risk_penalty", "hard_risk_count", "warning_count", "score_coverage", "fundamental_grade", "financial_health_status", "hotspot_score", "hotspot_rank", "is_hot", "combined_score", ] out_cols = [x for x in out_cols if x in df.columns] out = df[out_cols + ["catalog_order"]].sort_values( ["catalog_order", "total_score", "source_code"], ascending=[True, False, True] ).drop(columns=["catalog_order"]).reset_index(drop=True) csv_path = PROCESSED / "full_market_catalog.csv" pq_path = PROCESSED / "full_market_catalog.parquet" out.to_csv(csv_path, index=False, encoding="utf-8-sig") out.to_parquet(pq_path, index=False) # ---- 校验与摘要 ------------------------------------------------------ print(f"全市场目录：{len(out)} 行，{len(out.columns)} 列") print(f" 写入 {csv_path}") print(f" 写入 {pq_path}") print() print("层级分布：") for tier, n in out["catalog_tier"].value_counts().reindex( ["优先研究", "热点交集", "基本面精选", "观察", "基本面合格", "仅入基础池", "其他（数据不完整）"] ).fillna(0).astype(int).items(): print(f" {tier:12s} {n:5d}") print() covered = out["in_full_market"].sum() print(f"覆盖校验：全部个股 = {len(out)}，其中收录 = {int(covered)}") print(f" 基础池 = {int(out['in_basic_pool'].sum())}") print(f" 基本面合格 = {int(out['in_fundamental_eligible'].sum())}") print(f" 候选池 = {int(out['in_candidate_pool'].sum())}") print() print("被排除原因 Top10：") ex = out.loc[out["exclude_reasons"].ne("") & ~out["in_candidate_pool"], "exclude_reasons"] exploded = ex.str.split("；").explode().value_counts().head(10) for k, v in exploded.items(): print(f" {k:24s} {v:5d}") return 0 if __name__ == "__main__": raise SystemExit(main())