from __future__ import annotations from dataclasses import dataclass from datetime import date import json from pathlib import Path import struct import numpy as np import pandas as pd from .config import Settings from .tdx import read_day_dates HEADER = struct.Struct(" pd.Timestamp | None: if not np.isfinite(value) or value <= 0: return None raw = int(value) if raw < 1_000_000: raw += 20_000_000 parsed = pd.to_datetime(str(raw), format="%Y%m%d", errors="coerce") return None if pd.isna(parsed) else parsed def _next_trade_dates(announcements: pd.Series, calendar: np.ndarray) -> pd.Series: ann = announcements.to_numpy(dtype="datetime64[D]") positions = np.searchsorted(calendar, ann, side="right") values = np.full(len(positions), np.datetime64("NaT"), dtype="datetime64[D]") valid = positions < len(calendar) values[valid] = calendar[positions[valid]] return pd.Series(pd.to_datetime(values), index=announcements.index) def extract_financial_pit(settings: Settings, force: bool = False) -> FinanceBuildResult: source_dir = settings.tdx_root / "vipdoc" / "cw" files = [path for path in sorted(source_dir.glob("gpcw*.dat")) if path.stem[4:] >= "20181231" and path.stat().st_size > HEADER.size] if not files: raise RuntimeError(f"没有找到可用专业财务文件：{source_dir}") target = settings.financial_pit_path newest_source = max(path.stat().st_mtime for path in files) metadata_path = settings.report_dir / "financial_metadata.json" if target.exists() and target.stat().st_mtime >= newest_source and metadata_path.exists() and not force: cached_metadata = json.loads(metadata_path.read_text(encoding="utf-8")) if cached_metadata.get("schema_version") == FINANCIAL_SCHEMA_VERSION: cached = pd.read_parquet(target, columns=["source_code"]) return FinanceBuildResult(False, len(cached), cached["source_code"].nunique(), target) rows: list[dict[str, object]] = [] file_audit: list[dict[str, object]] = [] max_field = max(max(FINANCIAL_FIELDS), 314) for path in files: raw = path.read_bytes() file_type, report_period_raw, record_count, _, record_size, _ = HEADER.unpack_from(raw) field_count = record_size // 4 if record_size % 4 == 0 else 0 accepted = 0 invalid_announcement = 0 invalid_offset = 0 for index in range(record_count): code_raw, _, offset = INDEX_RECORD.unpack_from(raw, HEADER.size + index * INDEX_RECORD.size) code = code_raw.decode("ascii", errors="ignore") if not code.startswith(("0", "3", "6")): continue if field_count < max_field or offset + record_size > len(raw): invalid_offset += 1 continue values = np.frombuffer(raw, dtype=" dict[str, float | int | None]: actual = pd.to_numeric(frame[reported], errors="coerce") mask = calculated.notna() & actual.notna() & np.isfinite(calculated) & np.isfinite(actual) error = (calculated[mask] - actual[mask]).abs() return { "samples": int(mask.sum()), "median_absolute_error": float(error.median()) if len(error) else None, "p95_absolute_error": float(error.quantile(0.95)) if len(error) else None, } def _validate_financial_relations(frame: pd.DataFrame) -> dict[str, object]: assets = frame["total_assets"].replace(0, np.nan) current_liabilities = frame["current_liabilities"].replace(0, np.nan) revenue = frame["operating_revenue_ytd"].replace(0, np.nan) debt_ratio = frame["total_liabilities"] / assets * 100 current_ratio = frame["current_assets"] / current_liabilities gross_margin = (frame["operating_revenue_ytd"] - frame["operating_cost_ytd"]) / revenue * 100 return { "debt_ratio_pct_points": _relation_error(frame, debt_ratio, "debt_ratio_reported"), "current_ratio": _relation_error(frame, current_ratio, "current_ratio_reported"), "gross_margin_pct_points": _relation_error(frame, gross_margin, "gross_margin_reported"), } def _write_financial_report(settings: Settings, metadata: dict[str, object], quality: pd.DataFrame) -> None: high_zero = quality.nlargest(15, "zero_rate_among_non_null") zero_lines = "\n".join( f"| {row.field} | {row.zero_rate_among_non_null:.1%} | {int(row.non_null_count)} |" for row in high_zero.itertuples(index=False) ) validation_lines = "\n".join( f"| {name} | {item['samples']} | {item['median_absolute_error']:.6g} | {item['p95_absolute_error']:.6g} |" for name, item in metadata["validation"].items() ) report = f"""# M1 财务字段质量报告 生成日期：{metadata['build_date']} 记录数：{metadata['rows']} 股票数：{metadata['stocks']} 报告期：{metadata['first_report_period']} 至 {metadata['latest_report_period']} 最新公告：{metadata['latest_announcement_date']} 最新可用交易日：{metadata['latest_available_trade_date']} 抽取财务字段：{metadata['selected_financial_fields']} 个 ## 1. 重要时间口径 - 财务数据从公告日之后的首个交易日开始可用。 - `report_period`、`announcement_date` 和 `available_trade_date` 分开保存。 - 本地专业财务文件不是历史版本快照，无法还原后续重述前的原始值；`is_vintage_snapshot` 固定为否。 ## 2. 关系式抽样验证 | 关系 | 样本数 | 绝对误差中位数 | 绝对误差P95 | |---|---:|---:|---:| {validation_lines} 关系式误差用于检查字段映射和单位，不表示报表本身存在问题。部分金融企业口径不同，后续将分行业复核。 ## 3. 原始零值率最高字段 | 字段 | 零值率 | 原始非空数 | |---|---:|---:| {zero_lines} 通达信专业财务文件中的数值槽位通常都有值，但部分未披露项目也可能编码为 0。因此“原始非空率 100%”不等于字段真实完整；后续计算指标时需要按字段语义区分“真实零值”和“疑似缺失”。全零字段暂不参与评分。完整统计见 `data/processed/financial_field_quality.csv`。 """ settings.report_dir.mkdir(parents=True, exist_ok=True) (settings.report_dir / "M1_财务字段质量报告.md").write_text(report, encoding="utf-8")