企业官网建设流程全解析

1. 项目概述：这不是一个“问题”，而是一次数据世界的入门安检

“What’s in the data?”——这句话乍看像一句随口的疑问，甚至有点傻气：数据里不就装着数据吗？但在我带过三十多个跨行业数据分析项目、亲手清洗过超过200TB原始日志和业务表之后，我越来越确信：这五个单词，是所有数据工作真正开始前必须签下的第一份免责声明，也是唯一一份不能跳过的入职体检报告。它不是在问“字段名是什么”，而是在问“这些字段名背后，有没有人真正校验过它们是否说真话”。你手里的CSV、Excel、数据库快照、API返回的JSON，从来不是客观事实的复刻，而是经过采集逻辑、系统限制、人为录入、网络抖动、权限过滤、时区偏移、编码转换等至少七层“现实滤镜”之后的残影。我见过电商团队用“下单时间”分析用户活跃度，结果发现83%的订单时间戳被ERP系统强制写成了服务器本地时间（UTC+8），而实际下单发生在东南亚仓库的凌晨三点；也见过医疗AI项目因患者年龄字段里混入了“>90”这样的文本值，导致整个回归模型在验证集上R²暴跌0.4。所以，“What’s in the data?”本质上是在启动一个数据可信度压力测试：它逼你放下建模冲动，先蹲下来，用放大镜看每一行、每一列、每一个空值、每一个异常分布。它适合三类人：刚拿到第一份真实业务数据的新手分析师，需要快速建立数据直觉；正在交付关键报表却屡被业务方质疑“数字不准”的中阶工程师；以及准备把数据接入自动化决策流的产品负责人——因为一旦跳过这一步，后面所有算法优化、可视化美化、A/B测试结论，都只是在一座沙堡上雕花。它不教你怎么写SQL或调参，它只教你一件事：如何用最朴素的工具，完成一次有尊严的数据初筛。

2. 核心思路拆解：为什么必须放弃“直接建模”，选择“先探再动”

2.1 从“数据即资产”到“数据即证物”的认知切换

很多团队一拿到数据，第一反应是“赶紧跑个模型看看效果”。这种路径在Kaggle竞赛里能赢，但在真实业务中大概率会输。原因很简单：Kaggle数据是经过主办方精心蒸馏、去噪、标注的“实验室样本”，而你的生产数据是未经处理的“犯罪现场证物”。我参与过一家连锁药店的会员复购预测项目，初始数据集包含127个字段，团队直接用XGBoost训练，AUC达到0.86，大家一片欢呼。上线两周后，运营反馈“推荐的药品完全不对路”。我们回溯才发现，“最近一次购药日期”字段在32%的记录中是空值，而填充逻辑是用“注册日期”代替——可注册日期平均比首次购药晚11.3天，且大量新注册用户根本没购药行为。模型学到的不是“复购模式”，而是“注册后第N天大概率会买药”的伪规律。这就是典型的“数据未探，模型先跑”陷阱。“What’s in the data?”的核心思路，就是强行把“数据探索”（EDA）从建模流程中的一个可选环节，升级为不可绕过的前置闸门。它要求你像法医一样对待每一份数据：先做尸检（结构扫描），再查病历（质量审计），最后才考虑治疗方案（建模）。这个切换的价值，在于把问题暴露在成本最低的阶段——改一行SQL比重构整个特征工程管道便宜100倍。

2.2 三层穿透式探查框架：结构层 → 质量层 → 语义层

我们不满足于“看一眼head()”或“df.describe()”，而是构建了一个三层穿透框架，确保每个维度都被触达：

• 结构层（Structure Layer）：回答“数据长什么样？”
  关注物理形态：文件格式（CSV/Parquet/JSON）、编码（UTF-8 vs GBK）、分隔符（逗号/制表符/竖线）、行列数、字段数量与类型（数值/字符串/日期/布尔）、嵌套深度（JSON中的多层对象）。这里的关键陷阱是“类型误判”——Pandas自动推断的object类型可能混杂了数字、日期和纯文本，而int64字段里可能藏着NaN（实际是float64）。我曾在一个政府公开数据集中发现，身份证号字段被识别为int64，导致末尾的0全部丢失（如110101199003072540变成11010119900307254），这是典型的结构层失察。

• 质量层（Quality Layer）：回答“数据有多可靠？”
  这是核心战场，聚焦四大硬指标：缺失率（逐字段统计，区分NULL、空字符串、占位符如'N/A'）、重复率（全行去重+关键字段组合去重，如订单ID+时间戳）、异常值（数值型用IQR/3σ，分类型用频次阈值，如出现概率<0.01%的类别需人工核验）、一致性（跨表关联字段的值域是否对齐，如用户表的city_id在城市字典表中是否存在）。特别提醒：缺失不是问题，缺失的模式才是问题。比如，某金融APP的“月均收入”字段在25-35岁用户中缺失率仅2%，而在55岁以上用户中高达68%——这暗示该字段采集逻辑与年龄强相关，而非随机丢失。

• 语义层（Semantic Layer）：回答“数据到底在说什么？”
  这是最易被忽略、却决定业务成败的一层。它追问：字段名是否准确反映含义？（如user_status是“激活/冻结”，还是“新注册/已认证/已注销”？）业务规则是否内嵌在数据中？（如order_amount是否含运费？discount是绝对值还是百分比？）时间字段是否有时区信息？（created_at是UTC还是本地时间？是否带时区标识？）我服务过一家出海SaaS公司，其last_login_time字段在数据库文档中写明“UTC时间”，但实际写入的是服务器本地时间（AWS us-east-1区域），导致全球用户登录热力图严重失真。语义层的探查，必须结合业务文档、API契约、前端埋点代码三者交叉验证，单靠数据本身永远无法自证。

2.3 工具选型逻辑：为什么坚持用Python + Pandas + Great Expectations

市面上有无数数据探查工具：Tableau Prep、Trifacta、OpenRefine，甚至Excel的Power Query。但我们坚持用Python生态，原因很务实：

• 可控性优先：GUI工具的“一键智能清洗”背后是黑箱算法。当它把'1,234.56'自动转成1234.56时，你无法确认它是否同时把'1.234,56'（德语格式）错误地转成了1.23456。而用Pandas，你可以精确控制pd.read_csv(..., thousands=',', decimal='.')，每一步转换都可审计、可复现、可写进CI/CD流水线。

• 与生产环境零缝隙：分析脚本最终要融入ETL管道或模型服务。用Python写的探查逻辑，稍作封装就能变成Airflow任务或Prefect Flow的质检节点。而用Tableau Prep做的清洗，导出后又要重新用SQL或Python重写一遍，徒增维护成本。

• Great Expectations（GE）是质变关键：它把“探查”升维成“声明式质量契约”。你不再写df['age'].isnull().sum()，而是定义expect_column_values_to_be_between('age', min_value=0, max_value=120)。这个Expectation可以：

  • 在开发环境生成数据质量报告（HTML）
  • 在测试环境作为单元测试运行（pytest兼容）
  • 在生产环境实时监控（对接Prometheus/Grafana）
  • 当质量不达标时自动阻断下游任务（如阻止模型训练）

这解决了传统探查最大的痛点：一次性快照，无法持续守护。GE让“What’s in the data?”从一句口头禅，变成一条可执行、可追踪、可告警的硬性规则。

3. 核心细节解析与实操要点：从“看到”到“看懂”的关键动作

3.1 结构层探查：别让编码和分隔符成为第一道墙

很多人栽在第一步：连文件都读不全。这不是技术问题，是习惯问题。我强制自己执行一个“三分钟初始化协议”：

1. 用file命令看本质（Linux/Mac）：

   file -i your_data.csv # 输出示例：your_data.csv: text/plain; charset=iso-8859-1 # 注意：charset不是UTF-8！强行用utf-8读会报错或乱码

2. 用head -n 5看前五行：
   不只看内容，重点看分隔符是否一致、是否有隐藏字符（如\r\n混用）、首行是否为标题（BOM头\ufeff会导致列名多出前缀）。

3. 用csvkit的in2csv做格式诊断（推荐安装）：

   in2csv --help # 查看支持格式 in2csv your_data.xls | head -n 3 # 直接转Excel为CSV流，避免Excel软件依赖

实操心得：

• CSV分隔符陷阱：国内银行导出的对账单常用分号;，而电商后台常用制表符\t。用pd.read_csv(..., sep=';')比sep=None（自动检测）更可靠，因为自动检测在首行有缺失时会失效。
• 编码救急三板斧：
  1. 先试utf-8-sig（兼容BOM）
  2. 再试gbk（中文Windows默认）
  3. 最后用chardet库探测：

     import chardet with open('data.csv', 'rb') as f: raw = f.read(10000) # 只读前1万字节 encoding = chardet.detect(raw)['encoding'] df = pd.read_csv('data.csv', encoding=encoding)

• 嵌套JSON的扁平化：遇到{"user": {"id": 123, "profile": {"age": 25}}}，别用手写json_normalize。用pandas.json_normalize(df['raw_json_column'], sep='_')，sep参数能避免字段名冲突（如user_id和order_user_id）。

3.2 质量层探查：用“四象限法”定位致命缺陷

我把质量缺陷按影响程度（高/低）和发现难度（易/难）分为四象限，优先处理“高影响-易发现”类：

针对“高影响-易发现”缺陷，我的标准化检查清单：

• 缺失率TOP5字段速查：

  # 不只看isnull()，还要抓空字符串和占位符 def calc_null_rate(series): total = len(series) nulls = series.isnull().sum() empties = (series == '').sum() na_strings = series.isin(['N/A', 'NULL', 'null', 'nan']).sum() return (nulls + empties + na_strings) / total if total > 0 else 0 missing_report = pd.DataFrame({ 'field': df.columns, 'null_rate': [calc_null_rate(df[col]) for col in df.columns] }).sort_values('null_rate', ascending=False).head(5)

• 重复行深度诊断：
  df.duplicated().sum()只能告诉你有多少重复，但不知道为什么重复。必须做：

  1. df[df.duplicated(keep=False)]找出所有重复块
  2. 对关键业务字段（如order_id,timestamp）分组，看是“同一订单被写入两次”，还是“不同订单共享了相同ID”
  3. 检查_etl_timestamp（如果存在）：重复行的时间戳是否完全一致？若是，则是上游写入bug；若不一致，则可能是重试机制导致。

• 数值型异常值的双保险检测：
  单用IQR（四分位距）会漏掉“高原型”分布（如大量0值+少量大额值）。我坚持用IQR + 分位数双阈值：

  def detect_outliers(series, iqr_factor=1.5, top_percentile=99.5): Q1 = series.quantile(0.25) Q3 = series.quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - iqr_factor * IQR upper_bound = Q3 + iqr_factor * IQR # 同时用99.5%分位数兜底 percentile_bound = series.quantile(top_percentile / 100) final_upper = max(upper_bound, percentile_bound) return series[(series < lower_bound) | (series > final_upper)] # 示例：检查交易金额 outliers = detect_outliers(df['amount']) print(f"Amount outliers: {len(outliers)} out of {len(df)} ({len(outliers)/len(df)*100:.2f}%)")

提示：对金额类字段，异常值常是“单位错误”。比如amount字段本应是“元”，但部分记录是“分”（10000元被录成1000000分）。此时看异常值是否集中在某个整数倍（如×100），比单纯删掉更有价值。

3.3 语义层探查：与业务方“对齐语言”的实战技巧

语义层的问题，80%源于“术语不统一”。技术团队说的“用户活跃”，业务方可能指“当日登录”，也可能指“当日产生支付”。我的破局方法是：用数据反向翻译业务规则。

• 字段名歧义破解术：
  遇到模糊字段名（如status,flag,type），立即做三件事：

  1. 穷举取值：df['status'].value_counts(dropna=False)，看有哪些值、各占多少比例
  2. 关联上下文：对每个取值，抽样10条记录，人工查看其created_at,updated_at,user_id等周边字段，找规律
  3. 反向验证：假设status=1代表“已支付”，则筛选status==1的记录，检查payment_time是否非空且早于updated_at；若大量payment_time为空，则假设错误。

• 时间字段时区校验法：
  这是跨境业务的生死线。我的标准操作：

  1. 看字段名是否含时区标识（created_at_utc,local_time）
  2. 若无，查数据库表结构：SHOW CREATE TABLE orders（MySQL）或\d orders（PostgreSQL），看字段类型是否为TIMESTAMP WITH TIME ZONE
  3. 终极验证：找一个已知时区的事件（如“北京时间2023-10-01 10:00:00举办的线上活动”），查数据库中对应event_time字段的存储值。若存的是2023-10-01 02:00:00，则为UTC；若存的是2023-10-01 10:00:00，则为本地时间。

• 业务规则内嵌挖掘：
  比如discount_rate字段，文档写“折扣率（0-1）”，但数据中出现1.5。这时不要急着清洗，先问：

  • 这些1.5是否都出现在product_category=='vip_bundle'的记录中？
  • 查看vip_bundle的销售政策文档，发现“VIP礼包享受150%返现”，原来1.5是返现倍数，不是折扣率。
    结论：字段名错误，应重命名为cashback_multiple。这比盲目归零或截断更有业务价值。

4. 实操过程与核心环节实现：一份可直接运行的探查脚本

4.1 构建你的第一个“数据体检报告”（Python版）

以下是一个精简但完整的探查脚本，我把它命名为data_health_check.py。它不追求炫技，只保证：跑得快、看得懂、改得动。

# -*- coding: utf-8 -*- """ 数据健康检查脚本 v1.0 输入：CSV/Excel/Parquet文件路径 输出：HTML报告 + 控制台摘要 """ import pandas as pd import numpy as np from datetime import datetime import warnings warnings.filterwarnings('ignore') def load_data(filepath): """智能加载数据，自动处理常见格式和编码""" if filepath.endswith('.csv'): # 尝试多种编码 encodings = ['utf-8-sig', 'gbk', 'latin-1'] for enc in encodings: try: return pd.read_csv(filepath, encoding=enc, low_memory=False) except UnicodeDecodeError: continue raise ValueError(f"无法用{encodings}解码 {filepath}") elif filepath.endswith(('.xls', '.xlsx')): return pd.read_excel(filepath, engine='openpyxl') elif filepath.endswith('.parquet'): return pd.read_parquet(filepath) else: raise ValueError(f"不支持的文件格式: {filepath}") def generate_health_report(df, filename): """生成核心健康报告""" report = {} # 基础信息 report['file_info'] = { 'filename': filename, 'shape': df.shape, 'memory_usage_mb': round(df.memory_usage(deep=True).sum() / 1024**2, 2), 'dtypes_summary': df.dtypes.value_counts().to_dict() } # 结构层：字段概览 report['schema'] = [] for col in df.columns: dtype = str(df[col].dtype) non_null_count = df[col].count() unique_count = df[col].nunique() sample_values = df[col].dropna().head(3).tolist() report['schema'].append({ 'column': col, 'dtype': dtype, 'non_null_count': non_null_count, 'null_count': len(df) - non_null_count, 'null_rate': round((len(df) - non_null_count) / len(df) * 100, 2), 'unique_count': unique_count, 'unique_rate': round(unique_count / len(df) * 100, 2) if len(df) > 0 else 0, 'sample_values': sample_values }) # 质量层：TOP5缺失字段 missing_df = pd.DataFrame(report['schema']).sort_values('null_rate', ascending=False) report['top_missing'] = missing_df.head(5).to_dict('records') # 质量层：数值型字段异常值 numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist() report['outliers'] = [] for col in numeric_cols: if len(df[col].dropna()) < 10: # 样本太少跳过 continue Q1 = df[col].quantile(0.25) Q3 = df[col].quantile(0.75) IQR = Q3 - Q1 lower = Q1 - 1.5 * IQR upper = Q3 + 1.5 * IQR outlier_mask = (df[col] < lower) | (df[col] > upper) outlier_count = outlier_mask.sum() if outlier_count > 0: report['outliers'].append({ 'column': col, 'outlier_count': outlier_count, 'outlier_rate': round(outlier_count / len(df) * 100, 2), 'lower_bound': round(lower, 2), 'upper_bound': round(upper, 2), 'sample_outliers': df[outlier_mask][col].head(3).tolist() }) # 语义层：高频字符串字段的值分布（前10） string_cols = df.select_dtypes(include=['object']).columns.tolist() report['top_string_values'] = [] for col in string_cols: if df[col].nunique() < 50: # 值太多不展示 value_counts = df[col].value_counts(dropna=False).head(10) report['top_string_values'].append({ 'column': col, 'value_counts': value_counts.to_dict() }) return report def print_console_summary(report): """打印简洁的控制台摘要""" print("\n" + "="*60) print("📊 数据健康检查报告摘要") print("="*60) info = report['file_info'] print(f"📁 文件: {info['filename']}") print(f"📦 行数×列数: {info['shape'][0]} × {info['shape'][1]}") print(f"💾 内存占用: {info['memory_usage_mb']} MB") print(f"\n🔍 结构层速览:") print(f" • 字段总数: {len(report['schema'])}") print(f" • 数值型字段: {len([x for x in report['schema'] if 'int' in x['dtype'] or 'float' in x['dtype']])}") print(f" • 字符串字段: {len([x for x in report['schema'] if 'object' in x['dtype']])}") print(f"\n⚠️ 质量层预警:") if report['top_missing']: top_miss = report['top_missing'][0] print(f" • 缺失最严重字段: '{top_miss['column']}' ({top_miss['null_rate']}% 缺失)") else: print(" • 无高缺失字段") if report['outliers']: top_out = report['outliers'][0] print(f" • 异常值最多字段: '{top_out['column']}' ({top_out['outlier_rate']}% 异常)") else: print(" • 无显著异常值") print(f"\n💡 语义层线索:") if report['top_string_values']: first_str = report['top_string_values'][0] print(f" • '{first_str['column']}' 前3值: {list(first_str['value_counts'].keys())[:3]}") print("="*60) def main(): import sys if len(sys.argv) != 2: print("用法: python data_health_check.py <文件路径>") sys.exit(1) filepath = sys.argv[1] print(f"正在检查: {filepath}") try: df = load_data(filepath) report = generate_health_report(df, filepath) print_console_summary(report) # 保存为HTML（简易版） html_report = f""" <html> <head><title>数据健康报告 - {filepath}</title></head> <body> <h1>📊 数据健康报告</h1> <h2>基础信息</h2> <p><strong>文件名:</strong> {report['file_info']['filename']}</p> <p><strong>形状:</strong> {report['file_info']['shape'][0]} 行 × {report['file_info']['shape'][1]} 列</p> <h2>TOP5缺失字段</h2> <table border="1"> <tr><th>字段</th><th>缺失率</th><th>缺失数</th></tr> {''.join([f'<tr><td>{x["column"]}</td><td>{x["null_rate"]}%</td><td>{x["null_count"]}</td></tr>' for x in report['top_missing']])} </table> <h2>异常值字段</h2> <p>{'无' if not report['outliers'] else '详见下表'}</p> {'<table border="1"><tr><th>字段</th><th>异常率</th><th>上下界</th></tr>' + ''.join([f'<tr><td>{x["column"]}</td><td>{x["outlier_rate"]}%</td><td>[{x["lower_bound"]}, {x["upper_bound"]}]</td></tr>' for x in report['outliers']]) + '</table>' if report['outliers'] else ''} </body> </html> """ html_path = filepath.rsplit('.', 1)[0] + '_health_report.html' with open(html_path, 'w', encoding='utf-8') as f: f.write(html_report) print(f"\n✅ HTML报告已保存至: {html_path}") except Exception as e: print(f"❌ 检查失败: {e}") sys.exit(1) if __name__ == "__main__": main()

使用方法：

1. 将脚本保存为data_health_check.py
2. 终端执行：python data_health_check.py ./data/orders_2023.csv
3. 控制台立即输出摘要，同时生成同名HTML报告

为什么这个脚本值得你抄作业：

• 零依赖：只用pandas和numpy，无需安装复杂工具链
• 防错设计：自动尝试多种编码，避免UnicodeDecodeError卡死
• 结果导向：不渲染花哨图表，只输出业务方能看懂的数字（缺失率%、异常率%）
• 可扩展：所有函数模块化，想加“时间字段时区检测”，只需在generate_health_report里新增一段逻辑

4.2 用Great Expectations实现自动化质量守门

上面的手动脚本适合单次探查，但生产环境需要持续守护。以下是GE的最小可行集成方案：

# 1. 安装（推荐用conda，避免依赖冲突） conda create -n ge_env python=3.9 conda activate ge_env pip install great-expectations # 2. 初始化GE项目 great_expectations init # 按提示选择数据源（CSV/Database等） # 3. 创建数据资产（以CSV为例） great_expectations datasource new # 选择 "Files on a filesystem (for processing with Pandas or Spark)" # 输入CSV路径

核心配置文件great_expectations/yaml_config.yml示例：

datasources: my_csv_datasource: class_name: Datasource execution_engine: class_name: PandasExecutionEngine data_connectors: configured_data_connector_name: class_name: ConfiguredAssetFilesystemDataConnector base_directory: ./data/ assets: orders: pattern: orders_(.*).csv group_names: - timestamp batch_spec_passthrough: reader_options: encoding: utf-8-sig parse_dates: ['created_at', 'updated_at']

定义你的第一条质量契约（Expectation）：

# 在Jupyter中运行 import great_expectations as ge from great_expectations.core.batch import BatchRequest context = ge.get_context() batch_request = BatchRequest( datasource_name="my_csv_datasource", data_connector_name="configured_data_connector_name", data_asset_name="orders" ) # 创建期望套件 expectation_suite_name = "orders_suite" context.create_expectation_suite( expectation_suite_name=expectation_suite_name, overwrite_existing=True ) validator = context.get_validator( batch_request=batch_request, expectation_suite_name=expectation_suite_name ) # 添加具体期望 validator.expect_table_row_count_to_equal(value=100000) # 总行数应为10万 validator.expect_column_values_to_not_be_null("order_id") # 订单ID不能为空 validator.expect_column_values_to_be_between("amount", min_value=0.01, max_value=100000) # 金额合理范围 validator.expect_column_values_to_match_regex("email", r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$') # 邮箱格式 # 保存期望套件 validator.save_expectation_suite(discard_failed_expectations=False)

运行质检并生成报告：

# 命令行一键运行 great_expectations checkpoint run orders_checkpoint # 报告自动生成在 great_expectations/uncommitted/data_docs/local_site/index.html

注意：GE的威力在于“契约即代码”。你定义的每一条expect_*，都会变成CI/CD中的一个测试用例。当上游数据源变更（如新增字段、修改类型），GE会在Pipeline中立刻失败，并给出清晰错误：“expect_column_values_to_be_betweenfailed for column 'amount' — found value -5000.0”。这比等模型上线后业务方投诉高效十倍。

5. 常见问题与排查技巧实录：那些没人告诉你的坑

5.1 “数据没问题，但结果总不对”——隐性漂移的识别与应对

这是最高频、最隐蔽的故障。现象：昨天探查一切正常，今天跑同样脚本，模型指标却下跌。根源往往是数据漂移（Data Drift），而非代码错误。

典型场景与排查技巧：

• 概念漂移（Concept Drift）：业务规则变了，但数据字段名没变。
  案例：某信贷风控模型用credit_score字段，阈值设为650分。某月起，征信机构更新评分模型，同样用户分数普遍下降50分。模型仍用650阈值，导致大量优质客户被拒。
  排查：定期对比字段分布。用scipy.stats.ks_2samp做KS检验：

  from scipy import stats # 比较上周和本周的credit_score分布 ks_stat, p_value = stats.ks_2samp(last_week_scores, this_week_scores) if p_value < 0.05: # 分布显著不同 print("⚠️ 概念漂移预警！")

• 采样漂移（Sampling Drift）：数据抽取逻辑被悄悄修改。
  案例：ETL任务原用WHERE created_at >= '2023-01-01'，运维误删了单引号，变成WHERE created_at >= 2023-01-01（被解释为数字减法），结果只抽取了2023年1月1日当天的数据。
  排查：对时间字段做min()/max()/count()三连查，对比历史基线。设置告警：若max(created_at)比当前时间早72小时，即触发告警。

• 编码漂移（Encoding Drift）：上游系统升级，改变了字符串编码或JSON序列化方式。
  案例：老版本API返回{"name": "张三"}，新版本返回{"name": "Zhang San"}，但字段名仍是name。模型学到的“中文姓名特征”全部失效。
  排查：对字符串字段计算df['name'].str.len().describe()，若平均长度突降50%，高度可疑。

5.2 “为什么df.describe()看不出问题？”——描述统计的三大盲区

df.describe()是新手最爱，但它有致命局限：

我的替代方案：df.info()+df.nunique()+ 自定义分布快照

def quick_distribution_snapshot(df, top_n=10): """生成字段分布快照，规避describe盲区""" snapshot = {} for col in df.columns: if df[col].dtype in ['object', 'bool']: # 字符串/布尔：看唯一值和高频值 vc = df[col].value_counts(dropna=False).head(top_n) snapshot[col] = { 'type': 'categorical', 'n_unique': df[col].nunique(dropna=False), 'total': len(df), 'top_values': vc.to_dict() } else: # 数值：看分位数和异常值 desc = df[col].describe(percentiles=[.01, .25, .5, .75, .99]) outliers = detect_outliers(df[col], iqr_factor=2.0) # 更宽松 snapshot[col] = { 'type': 'numeric', 'min