from __future__ import annotations from collections import Counter from datetime import date import json from pathlib import Path import re import sqlite3 import numpy as np import pandas as pd import pyarrow.parquet as pq from .config import Settings from .tdx import canonical_id, load_tdx_research_industry, load_tdx_security_names, scan_stock_day_files def _load_legacy_names(path: Path) -> dict[str, str]: if not path.exists(): return {} raw = json.loads(path.read_text(encoding="utf-8")) return {key.replace(".", ""): value for key, value in raw.items()} def _csv_edge(path: Path) -> dict[str, object]: header = "" first = "" last = "" rows = 0 try: with path.open("r", encoding="utf-8", errors="replace") as handle: header = handle.readline().strip() for line in handle: text = line.strip() if not text: continue rows += 1 if not first: first = text last = text except OSError: pass return { "source_code": path.stem, "legacy_csv_rows": rows, "legacy_csv_first_date": first.split(",", 1)[0] if first else None, "legacy_csv_last_date": last.split(",", 1)[0] if last else None, "legacy_csv_header": header, "legacy_csv_size": path.stat().st_size, } def audit_legacy_csv(data_dir: Path) -> pd.DataFrame: records = [_csv_edge(path) for path in sorted(data_dir.glob("*.csv")) if not path.name.startswith("idx_")] if not records: # No legacy CSVs present. Return an EMPTY frame that still carries the full # schema: build_security_master merges this on "source_code", and a column-less # DataFrame raises KeyError: 'source_code'. An empty frame also makes # has_legacy_csv evaluate to False for every stock, which is correct. return pd.DataFrame({ "source_code": pd.Series(dtype="object"), "legacy_csv_rows": pd.Series(dtype="int64"), "legacy_csv_first_date": pd.Series(dtype="object"), "legacy_csv_last_date": pd.Series(dtype="object"), "legacy_csv_header": pd.Series(dtype="object"), "legacy_csv_size": pd.Series(dtype="int64"), }) return pd.DataFrame(records) def _business_days_late(last_date: pd.Series, today: date) -> pd.Series: values = [] end = np.datetime64(today.isoformat(), "D") for value in last_date: if pd.isna(value): values.append(None) continue start = np.datetime64(pd.Timestamp(value).date().isoformat(), "D") values.append(max(int(np.busday_count(start, end)), 0)) return pd.Series(values, dtype="Int64") def _finance_summary(path: Path) -> tuple[pd.DataFrame, dict[str, object]]: if not path.exists(): return pd.DataFrame(columns=["source_code"]), {"exists": False} available_columns = pq.ParquetFile(path).schema.names if "source_code" in available_columns: code_col, period_col = "source_code", "report_period" announced_col, available_col = "announcement_date", "available_trade_date" else: code_col, period_col = "code", "period" announced_col, available_col = "announced", "available" finance = pd.read_parquet(path, columns=[code_col, period_col, announced_col, available_col]) finance[period_col] = pd.to_datetime(finance[period_col], errors="coerce") finance[announced_col] = pd.to_datetime(finance[announced_col], errors="coerce") finance[available_col] = pd.to_datetime(finance[available_col], errors="coerce") grouped = finance.groupby(code_col, as_index=False).agg( finance_records=(period_col, "size"), first_finance_period=(period_col, "min"), latest_finance_period=(period_col, "max"), latest_finance_announcement=(announced_col, "max"), latest_finance_available=(available_col, "max"), ).rename(columns={code_col: "source_code"}) summary = { "exists": True, "rows": int(len(finance)), "stocks": int(finance[code_col].nunique()), "first_period": finance[period_col].min().date().isoformat(), "latest_period": finance[period_col].max().date().isoformat(), "latest_announcement": finance[announced_col].max().date().isoformat(), "columns_in_cached_dataset": int(len(available_columns)), } return grouped, summary def _raw_finance_audit(tdx_root: Path, cache_path: Path) -> dict[str, object]: files = sorted((tdx_root / "vipdoc" / "cw").glob("gpcw*.dat")) usable = [path for path in files if path.stat().st_size > 20] latest_usable = max((path.stem[4:] for path in usable), default=None) newest_mtime = max((path.stat().st_mtime for path in usable), default=0) cache_mtime = cache_path.stat().st_mtime if cache_path.exists() else 0 return { "raw_file_count": len(files), "usable_raw_file_count": len(usable), "latest_usable_report_period": latest_usable, "cache_older_than_raw_source": bool(cache_mtime and newest_mtime > cache_mtime), } def build_security_master(settings: Settings) -> tuple[pd.DataFrame, pd.DataFrame, dict[str, object]]: day = scan_stock_day_files(settings.tdx_root) if not day: raise RuntimeError(f"没有找到通达信股票日线：{settings.tdx_root}") day_df = pd.DataFrame([item.__dict__ for item in day]).drop(columns="path") day_df["first_date"] = pd.to_datetime(day_df["first_date"], errors="coerce") day_df["last_date"] = pd.to_datetime(day_df["last_date"], errors="coerce") legacy_names = _load_legacy_names(settings.legacy_data_dir / "names.json") tdx_names = load_tdx_security_names(settings.tdx_root) industries = load_tdx_research_industry(settings.tdx_root) legacy = audit_legacy_csv(settings.legacy_data_dir) finance_path = settings.financial_pit_path if settings.financial_pit_path.exists() else settings.legacy_finance_parquet finance_by_stock, finance_stats = _finance_summary(finance_path) day_df["security_id"] = day_df["source_code"].map(canonical_id) day_df["exchange"] = day_df["source_code"].str[:2].str.upper() day_df["code"] = day_df["source_code"].str[2:] day_df["tdx_name"] = day_df["source_code"].map(tdx_names) day_df["legacy_name"] = day_df["source_code"].map(legacy_names) day_df["name"] = day_df["tdx_name"].combine_first(day_df["legacy_name"]) day_df["name_source"] = np.select( [day_df["tdx_name"].notna(), day_df["legacy_name"].notna()], ["tdx_security_table_current", "legacy_tencent_names_snapshot"], default="missing", ) industry_df = pd.DataFrame.from_dict(industries, orient="index").rename_axis("source_code").reset_index() master = day_df.merge(industry_df, on="source_code", how="left") master = master.merge(legacy, on="source_code", how="left") master = master.merge(finance_by_stock, on="source_code", how="left") current_cutoff = master["last_date"].max() master["business_days_stale"] = _business_days_late(master["last_date"], date.today()) master["is_name_missing"] = master["name"].isna() | master["name"].eq("") master["is_st_current"] = master["name"].fillna("").str.contains(r"(?:\*?ST|退)", case=False, regex=True) master["has_industry"] = master["industry_l2_code"].notna() master["has_finance"] = master["finance_records"].fillna(0).gt(0) master["has_legacy_csv"] = master["legacy_csv_rows"].fillna(0).gt(0) master["listed_250_days"] = master["record_count"].ge(settings.min_trading_days) master["liquidity_pass"] = master["avg_amount_20"].fillna(0).ge(settings.min_avg_amount_20) master["is_current_to_market_cutoff"] = master["last_date"].eq(current_cutoff) master["price_fresh_enough"] = master["business_days_stale"].fillna(999).le(settings.max_business_day_staleness) master["basic_pool_eligible"] = ( ~master["is_name_missing"] & ~master["is_st_current"] & master["record_size_valid"] & master["latest_ohlc_valid"] & master["is_current_to_market_cutoff"] & master["listed_250_days"] & master["liquidity_pass"] & master["has_industry"] & master["has_finance"] ) master["price_source"] = "tdx_unadjusted_day" master["industry_source"] = "tdx_research_industry_current" master = master.sort_values(["exchange", "code"]).reset_index(drop=True) issue_rows: list[dict[str, object]] = [] checks = { "missing_name": master["is_name_missing"], "current_st_or_delisting_label": master["is_st_current"], "invalid_day_record_size": ~master["record_size_valid"], "invalid_latest_ohlc": ~master["latest_ohlc_valid"], "price_behind_market_cutoff": ~master["is_current_to_market_cutoff"], "less_than_min_trading_days": ~master["listed_250_days"], "below_liquidity_floor": ~master["liquidity_pass"], "missing_industry": ~master["has_industry"], "missing_finance": ~master["has_finance"], } severity = { "missing_name": "error", "current_st_or_delisting_label": "warning", "invalid_day_record_size": "error", "invalid_latest_ohlc": "error", "price_behind_market_cutoff": "warning", "less_than_min_trading_days": "info", "below_liquidity_floor": "info", "missing_industry": "error", "missing_finance": "warning", } for issue_code, mask in checks.items(): for row in master.loc[mask, ["security_id", "source_code"]].itertuples(index=False): issue_rows.append({"security_id": row.security_id, "source_code": row.source_code, "issue_code": issue_code, "severity": severity[issue_code]}) issues = pd.DataFrame(issue_rows, columns=["security_id", "source_code", "issue_code", "severity"]) raw_finance = _raw_finance_audit(settings.tdx_root, finance_path) price_built = settings.daily_price_path.exists() limitations = [ "名称和ST标签只有当前快照", "行业映射只有当前版本，历史回测存在成分穿越风险", "财务缓存不是真正的历史版本快照，后续重述尚未解决", "审计与监管风险目前依据近两年风险公告标题识别，尚非完整监管档案", "股权质押为当前快照，解禁和大股东减持尚未接入", ] if price_built: limitations.insert(0, "前复权来自本地gbbq仿射模型，极早期非正价格区间必须禁用") else: limitations.insert(0, "通达信日线为未复权价格") summary = { "run_date": date.today().isoformat(), "market_data_cutoff": current_cutoff.date().isoformat(), "market_data_business_days_stale": int(np.busday_count(np.datetime64(current_cutoff.date()), np.datetime64(date.today()))), "stock_day_files": int(len(master)), "sh_stocks": int(master["exchange"].eq("SH").sum()), "sz_stocks": int(master["exchange"].eq("SZ").sum()), "named_stocks": int((~master["is_name_missing"]).sum()), "current_st_or_delisting_labels": int(master["is_st_current"].sum()), "industry_mapped_stocks": int(master["has_industry"].sum()), "industry_l1_count": int(master["industry_l1_code"].nunique()), "industry_l2_count": int(master["industry_l2_code"].nunique()), "industry_l3_count": int(master["industry_l3_code"].nunique()), "finance_covered_stocks": int(master["has_finance"].sum()), "legacy_csv_covered_stocks": int(master["has_legacy_csv"].sum()), "basic_pool_eligible": int(master["basic_pool_eligible"].sum()), "issue_counts": {key: int(mask.sum()) for key, mask in checks.items()}, "finance_cache": finance_stats, "finance_raw": raw_finance, "standardized_price_available": price_built, "limitations": limitations, } return master, issues, summary def build_industry_tables(master: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: current = master[[ "security_id", "source_code", "industry_l1_code", "industry_l1_name", "industry_l2_code", "industry_l2_name", "industry_l3_code", "industry_l3_name", "industry_source", ]].copy() rows: list[dict[str, object]] = [] for level in (1, 2, 3): code_col = f"industry_l{level}_code" name_col = f"industry_l{level}_name" for code, name in current[[code_col, name_col]].drop_duplicates().dropna(subset=[code_col]).itertuples(index=False): parent = None if level == 2: parent = code[:3] elif level == 3: parent = code[:5] if len(code) >= 5 else code rows.append({"industry_code": code, "industry_name": name, "level": level, "parent_code": parent, "industry_source": "tdx_research_industry_current"}) dimension = pd.DataFrame(rows).drop_duplicates("industry_code").sort_values(["level", "industry_code"]) return current, dimension def _markdown_report(summary: dict[str, object], master: pd.DataFrame, issues: pd.DataFrame) -> str: top_industries = ( master.groupby("industry_l2_name", dropna=False).size().sort_values(ascending=False).head(15) ) top_lines = "\n".join(f"| {str(name)} | {int(count)} |" for name, count in top_industries.items()) issue_lines = "\n".join( f"| {code} | {int(count)} |" for code, count in Counter(issues["issue_code"]).most_common() ) freshness = "不通过" if summary["market_data_business_days_stale"] > 2 else "通过" finance_stale = "是" if summary["finance_raw"]["cache_older_than_raw_source"] else "否" finance_columns = summary["finance_cache"].get("columns_in_cached_dataset", 0) finance_status = ( f"当前已重建扩展财务字段集，共 {finance_columns} 列；下一步进入派生指标和分行业口径处理。" if finance_columns > 50 else "当前仍是简化财务字段集，下一步必须重新执行全字段抽取和字段质量审计。" ) finance_blocker = ( "3. 财务缓存早于部分原始专业财务文件，需要重建。" if summary["finance_raw"]["cache_older_than_raw_source"] else "3. 财务字段已重建，但原始零值与缺失值仍需按字段语义处理。" ) price_blocker = ( "2. 前复权行情已经生成；早期非正价格区间必须由因子引擎自动屏蔽。" if summary.get("standardized_price_available") else "2. 日线没有复权因子，不能直接用于跨除权日收益和长期趋势计算。" ) return f"""# M1 数据质量报告 生成日期：{summary['run_date']} 行情数据截止：{summary['market_data_cutoff']} 相对运行日滞后工作日：{summary['market_data_business_days_stale']} 行情新鲜度闸门：**{freshness}** ## 1. 证券主表 | 指标 | 数量 | |---|---:| | 日线股票文件 | {summary['stock_day_files']} | | 上海股票 | {summary['sh_stocks']} | | 深圳股票 | {summary['sz_stocks']} | | 有名称 | {summary['named_stocks']} | | 当前 ST/退市标签 | {summary['current_st_or_delisting_labels']} | | 有二级行业映射 | {summary['industry_mapped_stocks']} | | 有历史财务缓存 | {summary['finance_covered_stocks']} | | 通过初始基础池规则 | {summary['basic_pool_eligible']} | 说明：基础池当前按非 ST、至少 250 根日线、最近 20 日平均成交额不低于 5000 万元、行情跟上市场截止日、行业和财务均有覆盖进行筛选。全局行情新鲜度不通过时，基础池只能视为历史快照，不能视为当日结果。 ## 2. 行业覆盖 - 一级行业：{summary['industry_l1_count']} 个。 - 二级行业：{summary['industry_l2_count']} 个。 - 三级行业：{summary['industry_l3_count']} 个。 - 分类口径：通达信研究行业当前映射，不标称为官方申万。 | 二级行业 | 股票数 | |---|---:| {top_lines} ## 3. 财务数据 - 缓存行数：{summary['finance_cache'].get('rows', 0)}。 - 缓存覆盖股票：{summary['finance_cache'].get('stocks', 0)}。 - 报告期：{summary['finance_cache'].get('first_period')} 至 {summary['finance_cache'].get('latest_period')}。 - 最新公告日期：{summary['finance_cache'].get('latest_announcement')}。 - 通达信专业财务原始文件：{summary['finance_raw']['raw_file_count']} 个，其中非空可用 {summary['finance_raw']['usable_raw_file_count']} 个。 - 最新非空报告期：{summary['finance_raw']['latest_usable_report_period']}。 - 财务缓存早于原始文件：**{finance_stale}**。 {finance_status} ## 4. 数据问题统计 | 问题 | 股票数 | |---|---:| {issue_lines} ## 5. 当前阻断项 1. 行情截止日相对运行日超过允许阈值，正式每日任务必须中止并提示更新通达信数据。 {price_blocker} {finance_blocker} 4. 历史 ST 和历史行业成分缺失，历史回测必须显示限制标签。 5. 审计、处罚和资金占用已接入近两年风险公告标题识别，但仍不是完整监管档案；未命中不能解释为无风险。 6. 股权质押已接入当前快照；解禁和大股东减持尚未覆盖。 """ def _persist_database(settings: Settings, tables: dict[str, pd.DataFrame]) -> tuple[str, str, str | None]: duckdb_path = settings.database_dir / "fundamental_mvp.duckdb" try: import duckdb with duckdb.connect(str(duckdb_path)) as connection: for table_name, frame in tables.items(): view_name = f"{table_name}_df" connection.register(view_name, frame) connection.execute(f"CREATE OR REPLACE TABLE {table_name} AS SELECT * FROM {view_name}") if settings.financial_pit_path.exists(): finance_path = str(settings.financial_pit_path).replace("'", "''") connection.execute( f"CREATE OR REPLACE VIEW financial_statement_pit AS " f"SELECT * FROM read_parquet('{finance_path}')" ) if settings.daily_price_path.exists(): price_path = str(settings.daily_price_path).replace("'", "''") connection.execute( f"CREATE OR REPLACE VIEW daily_price AS " f"SELECT * FROM read_parquet('{price_path}')" ) return "duckdb", str(duckdb_path), None except (ImportError, OSError) as exc: sqlite_path = settings.database_dir / "fundamental_mvp.sqlite" with sqlite3.connect(sqlite_path) as connection: for table_name, frame in tables.items(): safe = frame.copy() for column in safe.select_dtypes(include=["datetime64[ns]", "datetimetz"]).columns: safe[column] = safe[column].dt.strftime("%Y-%m-%d") safe.to_sql(table_name, connection, if_exists="replace", index=False) return "sqlite", str(sqlite_path), f"DuckDB 不可用，已降级：{type(exc).__name__}: {exc}" def persist_outputs(settings: Settings, master: pd.DataFrame, issues: pd.DataFrame, summary: dict[str, object]) -> None: settings.processed_dir.mkdir(parents=True, exist_ok=True) settings.database_dir.mkdir(parents=True, exist_ok=True) settings.report_dir.mkdir(parents=True, exist_ok=True) industry_current, industry_dimension = build_industry_tables(master) master.to_parquet(settings.processed_dir / "security_master.parquet", index=False) master.to_csv(settings.processed_dir / "security_master.csv", index=False, encoding="utf-8-sig") industry_current.to_parquet(settings.processed_dir / "industry_current.parquet", index=False) industry_current.to_csv(settings.processed_dir / "industry_current.csv", index=False, encoding="utf-8-sig") industry_dimension.to_csv(settings.processed_dir / "industry_dimension.csv", index=False, encoding="utf-8-sig") issues.to_csv(settings.processed_dir / "data_quality_issues.csv", index=False, encoding="utf-8-sig") tables = { "security_master": master, "industry_current": industry_current, "industry_dimension": industry_dimension, "data_quality_issue": issues, } backend, database_path, fallback_reason = _persist_database(settings, tables) summary["database"] = {"backend": backend, "path": database_path, "fallback_reason": fallback_reason} summary_path = settings.report_dir / "data_quality_summary.json" summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") report = _markdown_report(summary, master, issues) (settings.report_dir / "M1_数据质量报告.md").write_text(report, encoding="utf-8") def run_inventory(settings: Settings) -> dict[str, object]: master, issues, summary = build_security_master(settings) persist_outputs(settings, master, issues, summary) return summary