""" Universal data profiling engine: auto-detect schema, statistical features, and semantic inference from arbitrary Excel data. Enhanced with content-based value analysis, distribution shape analysis, derived metric detection, and multi-dimensional quality scoring. """ import pandas as pd import numpy as np from datetime import datetime, date from collections import Counter import re import math from report_config import ColumnProfile, ColumnRole # ===================================================================== # DATE PATTERNS — expanded for broader format coverage # ===================================================================== DATE_PATTERNS = [ re.compile(r'^\d{4}年\d{1,2}月\d{1,2}日$'), re.compile(r'^\d{4}-\d{2}-\d{2}$'), re.compile(r'^\d{4}/\d{1,2}/\d{1,2}$'), re.compile(r'^\d{4}\.\d{1,2}\.\d{1,2}$'), re.compile(r'^\d{4}年\d{1,2}月$'), re.compile(r'^\d{4}-\d{2}$'), re.compile(r'^\d{2}-\d{2}$'), re.compile(r'^\d{2}/\d{2}$'), re.compile(r'^\d{8}$'), # YYYYMMDD ] TIME_KEYWORDS = [ '日期', '时间', 'date', 'time', '年', '月', '日', '周', '期', '季度', 'period', 'month', 'year', 'quarter', 'week', 'day', 'timestamp', '月份', '年份', '周期', '时段', 'time', 'datetime', 'date', 'created', 'updated', 'modified', '发生', '录入', '创建', ] NUMERIC_KEYWORDS = [ '金额', '数量', '台数', '件数', '元', '价格', '收入', '支出', '利润', '成本', '费用', '销量', '销售额', '总数', '合计', 'amount', 'price', 'qty', 'quantity', 'revenue', 'cost', 'sales', 'volume', 'value', 'total', 'sum', 'count', '单数', '笔数', '人数', '天数', '比率', '占比', '比例', '率', '预算', 'budget', '花费', 'spend', 'fee', '金额', '单价', 'unit', '得分', 'score', 'rating', '评分', ] CATEGORY_KEYWORDS = [ '国家', '区域', '地区', '城市', '省份', '状态', '类型', '类别', '分类', '部门', '组', '等级', '级别', '品牌', '渠道', 'country', 'region', 'city', 'status', 'type', 'category', 'department', 'group', 'level', 'brand', 'channel', '负责人', 'owner', 'manager', '产品', 'product', '阶段', '供应商', 'supplier', '客户', 'customer', '行业', 'industry', '性别', 'gender', '职位', 'title', '角色', 'role', '标签', 'tag', '科目', 'account', '方向', 'direction', '方式', 'method', '意向', 'intent', 'intention', 'priority', '优先级', ] ID_KEYWORDS = [ 'id', '编号', '序号', '代码', 'code', 'no', '编码', '合同号', '订单号', '工单号', '流水号', '单号', '标识', 'key', 'uuid', 'guid', 'sn', '序列号', '身份证', 'phone', '手机', '邮箱', 'email', '电话', 'tel', 'mobile', ] TEXT_KEYWORDS = [ '备注', '描述', '说明', '详情', '内容', '意见', '建议', '进度更新', 'note', 'description', 'detail', 'remark', 'comment', 'memo', '地址', 'address', '介绍', '摘要', 'summary', '附注', '反馈', 'feedback', '理由', 'reason', '原因', 'cause', ] RATE_KEYWORDS = [ '率', 'ratio', 'rate', '占比', '比例', 'percentage', 'pct', 'percent', 'conversion', '转化率', '完成率', '增长率', ] # Patterns for value-based content detection PHONE_PATTERN = re.compile(r'^[\+]?[\d\-\(\)\s]{6,20}$') EMAIL_PATTERN = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$') URL_PATTERN = re.compile(r'^https?://', re.IGNORECASE) YEAR_PATTERN = re.compile(r'^\d{4}$') def _parse_date(val): if pd.isna(val): return None if isinstance(val, (datetime, date)): return val if isinstance(val, (int, float)) and not math.isnan(val): try: return pd.Timestamp(val).to_pydatetime() except (ValueError, OverflowError): pass s = str(val).strip() for pattern in DATE_PATTERNS: if pattern.match(s): for fmt in ('%Y年%m月%d日', '%Y-%m-%d', '%Y/%m/%d', '%Y.%m.%d', '%Y年%m月', '%Y-%m', '%m-%d', '%m/%d', '%Y%m%d'): try: return datetime.strptime(s, fmt) except ValueError: continue return None # ===================================================================== # VALUE-BASED CONTENT ANALYSIS # ===================================================================== def _analyze_value_patterns(series: pd.Series, sample_count: int = 100) -> dict: """Analyze actual data values to detect patterns and content types.""" non_null = series.dropna().astype(str).head(sample_count) if len(non_null) == 0: return {} patterns = {} # Check if values look like percentages # Only flag as percentage if: ends with % OR is a decimal fraction (0.0-1.0) pct_like = sum(1 for v in non_null if v.endswith('%') or (v.replace('.', '', 1).lstrip('-').isdigit() and 0 < float(v) <= 1 and not v.isdigit())) patterns['pct_ratio'] = pct_like / len(non_null) # Check for yes/no or true/false patterns yn_vals = {'是', '否', 'yes', 'no', 'true', 'false', 'y', 'n', 't', 'f', '有', '无', '0', '1'} yn_like = sum(1 for v in non_null if v.lower() in yn_vals) patterns['binary_ratio'] = yn_like / len(non_null) # Check for ordinal/categorical short text short_text = sum(1 for v in non_null if len(v) <= 20) patterns['short_text_ratio'] = short_text / len(non_null) # Check for phone-like patterns phone_like = sum(1 for v in non_null if PHONE_PATTERN.match(v)) patterns['phone_ratio'] = phone_like / len(non_null) # Check for email-like patterns email_like = sum(1 for v in non_null if EMAIL_PATTERN.match(v)) patterns['email_ratio'] = email_like / len(non_null) # Check for URL-like patterns url_like = sum(1 for v in non_null if URL_PATTERN.match(v)) patterns['url_ratio'] = url_like / len(non_null) # Check for pure digit strings (possible IDs) digit_only = sum(1 for v in non_null if v.isdigit() and len(v) >= 6) patterns['digit_id_ratio'] = digit_only / len(non_null) # Check for year-like values year_like = sum(1 for v in non_null if YEAR_PATTERN.match(v)) patterns['year_ratio'] = year_like / len(non_null) # Detect ordinal levels ordinal_sets = [ {'高', '中', '低', 'A', 'B', 'C', '甲', '乙', '丙'}, {'一级', '二级', '三级', '四级', 'level 1', 'level 2', 'level 3'}, {'critical', 'major', 'minor', 'high', 'medium', 'low'}, ] for oset in ordinal_sets: ord_like = sum(1 for v in non_null if v in oset) if ord_like / len(non_null) > 0.3: patterns['ordinal'] = True break else: patterns['ordinal'] = False # Average text length patterns['avg_text_len'] = round(non_null.str.len().mean(), 1) # Unique ratio unique_ratio = series.nunique() / max(len(non_null), 1) patterns['unique_ratio'] = round(unique_ratio, 4) return patterns def _infer_role_from_values(value_patterns: dict, col_name: str, dtype_str: str, unique_count: int, total_rows: int) -> str: """Infer column role based on value content analysis results.""" up = value_patterns # High ratio of email patterns if up.get('email_ratio', 0) > 0.5: return 'id' # High ratio of phone patterns if up.get('phone_ratio', 0) > 0.5: return 'id' # High ratio of URL patterns if up.get('url_ratio', 0) > 0.5: return 'text' # Mostly binary values (yes/no) if up.get('binary_ratio', 0) > 0.6: return 'category' # Mostly percentage values if up.get('pct_ratio', 0) > 0.5: return 'numeric' # High ratio of digit-only long strings (likely IDs) if up.get('digit_id_ratio', 0) > 0.5 and unique_count > total_rows * 0.5: return 'id' # Ordinal level detected if up.get('ordinal', False): return 'category' return None # No clear signal from values # ===================================================================== # SEMANTIC KEYWORD-BASED ROLE INFERENCE # ===================================================================== def _infer_column_role(col_name: str, dtype_str: str, sample_values: list, null_rate: float, unique_count: int, total_rows: int, value_patterns: dict = None) -> ColumnRole: col_lower = col_name.lower().strip() # 1) Value-based inference first (stronger signal) if value_patterns: value_role = _infer_role_from_values(value_patterns, col_name, dtype_str, unique_count, total_rows) if value_role: return ColumnRole(value_role) # 2) Keyword-based inference (expanded) if any(kw in col_lower for kw in ID_KEYWORDS): return ColumnRole.ID if any(kw in col_lower for kw in TIME_KEYWORDS): return ColumnRole.TIME if any(kw in col_lower for kw in NUMERIC_KEYWORDS): return ColumnRole.NUMERIC if any(kw in col_lower for kw in CATEGORY_KEYWORDS): return ColumnRole.CATEGORY if any(kw in col_lower for kw in TEXT_KEYWORDS): return ColumnRole.TEXT # 3) dtype-based fallback if 'int' in dtype_str or 'float' in dtype_str: if unique_count <= 15 and total_rows > 20: return ColumnRole.CATEGORY return ColumnRole.NUMERIC if 'bool' in dtype_str: return ColumnRole.CATEGORY if 'datetime' in dtype_str: return ColumnRole.TIME # 4) Cardinality-based inference if total_rows > 0: cardinality_ratio = unique_count / total_rows if cardinality_ratio > 0.8 and unique_count > 20: return ColumnRole.TEXT if cardinality_ratio < 0.3 and unique_count <= 30: return ColumnRole.CATEGORY return ColumnRole.TEXT def _infer_metric_label(col_name: str, role: ColumnRole, value_patterns: dict = None) -> str: col_lower = col_name.lower().strip() # If values are percentage-like, mark as '比率' if value_patterns and value_patterns.get('pct_ratio', 0) > 0.5: for kw in ['率', '转化', '占比', '比例']: if kw in col_lower: return col_name return col_name + '(占比)' label_map = { '金额': '金额', '销售额': '销售额', '收入': '收入', '利润': '利润', '数量': '数量', '台数': '台数', '件数': '件数', '订单数': '订单数', '成本': '成本', '费用': '费用', '销量': '销量', '占比': '占比', '天数': '天数', '人数': '人数', '比率': '比率', '转化率': '转化率', '增长率': '增长率', '完成率': '完成率', '单价': '单价', '价格': '价格', '得分': '得分', '评分': '评分', } for kw, label in label_map.items(): if kw in col_lower: return label # Check for rate-related keywords if any(kw in col_lower for kw in RATE_KEYWORDS): return '比率' if role == ColumnRole.NUMERIC: return col_name elif role == ColumnRole.TIME: return '日期' elif role == ColumnRole.CATEGORY: return col_name return col_name def _infer_unit(col_name: str, value_patterns: dict = None) -> str: col_lower = col_name.lower().strip() # If values are percentage-like if value_patterns and value_patterns.get('pct_ratio', 0) > 0.5: return '%' unit_map = { '金额': '元', '销售额': '元', '收入': '元', '利润': '元', '成本': '元', '费用': '元', '台数': '台', '件数': '件', '数量': '', '人数': '人', '天数': '天', '占比': '%', '比率': '%', '比例': '%', '率': '%', '转化率': '%', '增长率': '%', '完成率': '%', '单价': '元', '价格': '元', '得分': '分', '评分': '分', } for kw, unit in unit_map.items(): if kw in col_lower: return unit return '' # ===================================================================== # DISTRIBUTION SHAPE ANALYSIS # ===================================================================== def _calc_distribution_shape(series: pd.Series) -> dict: """Compute skewness, kurtosis and distribution type for numeric series.""" try: s = series.dropna() if len(s) < 4: return {} skew = round(float(s.skew()), 3) kurt = round(float(s.kurtosis()), 3) # Determine distribution type abs_skew = abs(skew) if abs_skew < 0.5: skew_type = '近似对称' elif abs_skew < 1.0: skew_type = '轻度偏态' else: skew_type = '显著偏态' if skew > 0.5: skew_dir = '右偏（长尾在右侧，大部分值偏小）' elif skew < -0.5: skew_dir = '左偏（长尾在左侧，大部分值偏大）' else: skew_dir = '基本对称' # Concentration analysis cv = float(s.std()) / float(s.mean()) if float(s.mean()) != 0 else 0 if cv < 0.3: concentration = '高度集中' elif cv < 0.7: concentration = '中度集中' elif cv < 1.2: concentration = '适度分散' else: concentration = '高度分散' return { 'skewness': skew, 'kurtosis': kurt, 'skew_type': skew_type, 'skew_direction': skew_dir, 'cv': round(cv, 3), 'concentration': concentration, } except Exception: return {} # ===================================================================== # DERIVED METRIC DETECTION # ===================================================================== def _detect_derived_relations(df: pd.DataFrame, numeric_cols: list) -> list[dict]: """ Detect potential derived relationships among numeric columns. E.g., A - B = C, A + B = C, A / B = C (approx.) """ relations = [] num_names = [c['column_name'] for c in numeric_cols] if len(num_names) < 3: return relations sample = df[num_names].dropna().head(500) for i, a_name in enumerate(num_names): for j, b_name in enumerate(num_names): if j <= i: continue a = sample[a_name] b = sample[b_name] # Check subtraction: a - b ≈ c or b - a ≈ c for diff_name in num_names: if diff_name in (a_name, b_name): continue d = sample[diff_name] diff_ab = (a - b - d).abs().mean() diff_ba = (b - a - d).abs().mean() threshold = max(d.mean(), 1) * 0.1 if diff_ab < threshold: relations.append({ 'type': 'subtraction', 'expression': f'{a_name} - {b_name} ≈ {diff_name}', 'accuracy': round(float(1 - diff_ab / max(float(d.mean()), 1)), 3), 'formula': f'{diff_name} = {a_name} - {b_name}', }) break elif diff_ba < threshold: relations.append({ 'type': 'subtraction', 'expression': f'{b_name} - {a_name} ≈ {diff_name}', 'accuracy': round(float(1 - diff_ba / max(float(d.mean()), 1)), 3), 'formula': f'{diff_name} = {b_name} - {a_name}', }) break # Check addition: a + b ≈ c for sum_name in num_names: if sum_name in (a_name, b_name): continue s = sample[sum_name] sum_ab = (a + b - s).abs().mean() threshold = max(s.mean(), 1) * 0.1 if sum_ab < threshold: relations.append({ 'type': 'addition', 'expression': f'{a_name} + {b_name} ≈ {sum_name}', 'accuracy': round(float(1 - sum_ab / max(float(s.mean()), 1)), 3), 'formula': f'{sum_name} = {a_name} + {b_name}', }) break # Also check for ratio relations if len(num_names) >= 2: for i, a_name in enumerate(num_names): for j, b_name in enumerate(num_names): if j <= i: continue a = sample[a_name] b = sample[b_name] ratio = (a / b.replace(0, np.nan)).dropna() if len(ratio) > 0: ratio_std = float(ratio.std()) ratio_mean = float(ratio.mean()) if ratio_mean > 0 and ratio_std / ratio_mean < 0.1: # Consistent ratio found relations.append({ 'type': 'ratio', 'expression': f'{a_name} / {b_name} ≈ {ratio_mean:.3f}', 'accuracy': round(float(1 - ratio_std / ratio_mean), 3), 'formula': f'{a_name} = {b_name} × {ratio_mean:.2f}', }) return relations # ===================================================================== # MAIN PROFILING FUNCTION # ===================================================================== def profile_dataframe(df: pd.DataFrame) -> dict: total_rows = len(df) columns = [] for col in df.columns: series = df[col] dtype_str = str(series.dtype) null_count = int(series.isna().sum()) null_rate = round(null_count / total_rows, 4) if total_rows else 0.0 non_null = series.dropna() unique_count = int(non_null.nunique()) sample_values = non_null.head(5).tolist() sample_values = [str(v) for v in sample_values] # Enhanced: Value pattern analysis value_patterns = _analyze_value_patterns(series) # Enhanced: Distribution shape analysis for numeric columns distribution_shape = None numeric_stats = None if pd.api.types.is_numeric_dtype(series) and not pd.api.types.is_bool_dtype(series): try: numeric_stats = { 'mean': round(float(series.mean()), 2) if not pd.isna(series.mean()) else 0, 'median': round(float(series.median()), 2) if not pd.isna(series.median()) else 0, 'min': round(float(series.min()), 2) if not pd.isna(series.min()) else 0, 'max': round(float(series.max()), 2) if not pd.isna(series.max()) else 0, 'std': round(float(series.std()), 2) if not pd.isna(series.std()) else 0, 'sum': round(float(series.sum()), 2) if not pd.isna(series.sum()) else 0, 'p25': round(float(series.quantile(0.25)), 2) if not pd.isna(series.quantile(0.25)) else 0, 'p75': round(float(series.quantile(0.75)), 2) if not pd.isna(series.quantile(0.75)) else 0, } distribution_shape = _calc_distribution_shape(series) except Exception: numeric_stats = None # Enhanced role inference with value patterns role = _infer_column_role(col, dtype_str, sample_values, null_rate, unique_count, total_rows, value_patterns) label = _infer_metric_label(col, role, value_patterns) unit = _infer_unit(col, value_patterns) # Enhanced: detect if column is a high-cardinality ID is_high_cardinality_id = (role == ColumnRole.TEXT and unique_count / max(total_rows, 1) > 0.8 and unique_count > 20) if is_high_cardinality_id: role = ColumnRole.ID columns.append(ColumnProfile( column_name=col, dtype=dtype_str, role=role, null_count=null_count, null_rate=null_rate, unique_count=unique_count, sample_values=sample_values, numeric_stats=numeric_stats, inferred_label=label, )) # Append extra metadata not in ColumnProfile columns[-1]._unit = unit columns[-1]._distribution_shape = distribution_shape columns[-1]._value_patterns = value_patterns time_cols = [c for c in columns if c.role == ColumnRole.TIME] numeric_cols = [c for c in columns if c.role == ColumnRole.NUMERIC] category_cols = [c for c in columns if c.role == ColumnRole.CATEGORY] text_cols = [c for c in columns if c.role == ColumnRole.TEXT] id_cols = [c for c in columns if c.role == ColumnRole.ID] # Date range inference date_range = (None, None) time_granularity = 'unknown' if time_cols: series = df[time_cols[0].column_name].dropna() parsed = series.apply(_parse_date).dropna() if len(parsed) > 0: date_range = (parsed.min(), parsed.max()) if len(parsed) >= 2: diff = (parsed.max() - parsed.min()).days if diff <= 1: time_granularity = 'daily' elif diff <= 7: time_granularity = 'weekly' elif diff <= 31: time_granularity = 'monthly' elif diff <= 92: time_granularity = 'quarterly' else: time_granularity = 'yearly' # Enhanced: Detect derived relations among numeric columns derived_relations = _detect_derived_relations(df, [c.__dict__ for c in numeric_cols]) # Enhanced: Multi-dimensional quality scoring quality_score, quality_details = _calc_quality_score( df, columns, numeric_cols, date_range ) # Outlier detection (Enhanced: with CV-based filtering) outlier_columns = [] for c in numeric_cols: ns = c.numeric_stats if ns and ns.get('std', 0) > 0 and ns.get('mean', 0) > 0: cv = ns['std'] / ns['mean'] if cv > 3: outlier_columns.append(c.column_name) return { 'total_rows': total_rows, 'total_columns': len(columns), 'columns': [c.__dict__ for c in columns], 'time_columns': [c.__dict__ for c in time_cols], 'numeric_columns': [c.__dict__ for c in numeric_cols], 'category_columns': [c.__dict__ for c in category_cols], 'text_columns': [c.__dict__ for c in text_cols], 'id_columns': [c.__dict__ for c in id_cols], 'date_range': ( date_range[0].strftime('%Y-%m-%d') if date_range[0] else None, date_range[1].strftime('%Y-%m-%d') if date_range[1] else None, ), 'time_granularity': time_granularity, 'data_quality': { 'score': quality_score, 'details': quality_details, 'high_null_columns': [c.column_name for c in columns if c.null_rate > 0.3], 'outlier_columns': outlier_columns, }, 'derived_relations': derived_relations, 'column_stats': [{ 'column_name': col.column_name, 'role': col.role.value, 'dtype': col.dtype, 'null_rate': col.null_rate, 'unique_count': col.unique_count, 'distribution_shape': getattr(col, '_distribution_shape', None), 'inferred_label': col.inferred_label, 'unit': getattr(col, '_unit', ''), 'numeric_stats': col.numeric_stats, } for col in columns], } # ===================================================================== # ENHANCED QUALITY SCORING # ===================================================================== def _calc_quality_score(df: pd.DataFrame, columns: list, numeric_cols: list, date_range: tuple) -> tuple: """Multi-dimensional quality scoring (0-100).""" score = 100 details = {} # 1) Completeness (30%) — null rates avg_null_rate = np.mean([c.null_rate for c in columns]) if columns else 0 completeness_penalty = min(30, avg_null_rate * 100 * 2) completeness = max(0, 30 - completeness_penalty) details['completeness'] = round(completeness, 1) # 2) Uniqueness (20%) — presence of ID columns or unique identifiers id_ratio = len([c for c in columns if c.role == ColumnRole.ID]) / max(len(columns), 1) uniqueness = min(20, 10 + id_ratio * 10) details['uniqueness'] = round(uniqueness, 1) # 3) Numeric health (25%) — outliers, zeros, negative values numeric_health = 25 for c in numeric_cols: series = df[c.column_name].dropna() if len(series) == 0: continue # Check for negative values in non-negative expected columns if c.inferred_label in ('台数', '数量', '金额', '人数'): neg_ratio = (series < 0).sum() / len(series) if neg_ratio > 0.05: numeric_health -= 5 # Check for excessive zeros zero_ratio = (series == 0).sum() / len(series) if zero_ratio > 0.5: numeric_health -= 3 details['numeric_health'] = max(0, numeric_health) # 4) Temporal consistency (15%) — if time columns exist, check date ordering temporal = 15 if date_range[0] and date_range[1]: if date_range[0] <= date_range[1]: temporal = 15 else: temporal = 5 details['temporal_consistency'] = temporal # 5) Completeness of categorical data (10%) cat_health = 10 for c in columns: if c.role == ColumnRole.CATEGORY and c.null_rate > 0.2: cat_health -= 2 details['categorical_health'] = max(0, cat_health) score = completeness + uniqueness + numeric_health + temporal + cat_health score = max(0, min(100, round(score))) return score, details # ===================================================================== # HELPER FUNCTIONS (enhanced) # ===================================================================== def profile_category_distribution(df: pd.DataFrame, col_name: str, top_n: int = 15) -> dict: if col_name not in df.columns: return {} counts = df[col_name].value_counts().head(top_n).to_dict() total = df[col_name].notna().sum() # Calculate concentration (Herfindahl index) pcts = [v / total for v in counts.values()] if total else [] hhi = sum(p * p for p in pcts) if pcts else 0 return { 'total_categories': df[col_name].nunique(), 'top_items': {str(k): {'count': int(v), 'pct': round(v / total * 100, 1) if total else 0} for k, v in counts.items()}, 'concentration_hhi': round(hhi, 4), 'concentration_label': '高度集中' if hhi > 0.5 else '中度集中' if hhi > 0.2 else '分散', } def profile_numeric_series(df: pd.DataFrame, col_name: str) -> dict: if col_name not in df.columns: return {} series = df[col_name].dropna() if len(series) == 0: return {} shape = _calc_distribution_shape(series) result = { 'count': len(series), 'sum': round(float(series.sum()), 2), 'mean': round(float(series.mean()), 2), 'median': round(float(series.median()), 2), 'min': round(float(series.min()), 2), 'max': round(float(series.max()), 2), 'std': round(float(series.std()), 2), } if shape: result.update(shape) return result def detect_data_issues(df: pd.DataFrame) -> list[dict]: issues = [] for col in df.columns: null_rate = df[col].isna().mean() if null_rate > 0.5: issues.append({ 'column': col, 'type': 'high_missing', 'severity': 'major', 'message': f'列"{col}"缺失率{null_rate:.1%}，建议排除或补全', }) elif null_rate > 0.1: issues.append({ 'column': col, 'type': 'moderate_missing', 'severity': 'minor', 'message': f'列"{col}"缺失率{null_rate:.1%}', }) if pd.api.types.is_numeric_dtype(df[col]): series = df[col].dropna() if len(series) > 0: q1, q3 = series.quantile(0.25), series.quantile(0.75) iqr = q3 - q1 lower, upper = q1 - 3 * iqr, q3 + 3 * iqr outlier_count = ((series < lower) | (series > upper)).sum() if outlier_count > len(series) * 0.1: issues.append({ 'column': col, 'type': 'outliers', 'severity': 'major', 'message': f'列"{col}"存在{outlier_count}个异常值（{outlier_count/len(series):.1%}）', }) # Check for negative values neg_count = (series < 0).sum() if neg_count > 0: issues.append({ 'column': col, 'type': 'negative_values', 'severity': 'minor', 'message': f'列"{col}"存在{neg_count}个负值', }) # Check for constant columns if df[col].nunique() <= 1 and null_rate < 1.0: issues.append({ 'column': col, 'type': 'constant_column', 'severity': 'minor', 'message': f'列"{col}"为常量列（仅1个唯一值），对分析无贡献', }) return issues def generate_summary_text(profile: dict) -> str: lines = [] lines.append(f"共 {profile['total_rows']:,} 行 × {profile['total_columns']} 列") num_cols = profile.get('numeric_columns', []) cat_cols = profile.get('category_columns', []) time_cols = profile.get('time_columns', []) lines.append(f"数值列: {len(num_cols)} 个 | 分类列: {len(cat_cols)} 个 | 时间列: {len(time_cols)} 个") dr = profile.get('date_range', (None, None)) if dr[0]: lines.append(f"时间范围: {dr[0]} ~ {dr[1]}") lines.append(f"时间粒度: {profile.get('time_granularity', 'unknown')}") q = profile.get('data_quality', {}) lines.append(f"数据质量评分: {q.get('score', 0)}/100") if q.get('details'): det = q['details'] lines.append(f" 完整性: {det.get('completeness', 0)}/30 | " f"数值健康: {det.get('numeric_health', 0)}/25 | " f"时间一致性: {det.get('temporal_consistency', 0)}/15") if q.get('high_null_columns'): lines.append(f"高缺失列: {', '.join(q['high_null_columns'])}") # Enhanced: derived relations derived = profile.get('derived_relations', []) if derived: lines.append(f"检测到 {len(derived)} 个数值关系:") for rel in derived[:5]: lines.append(f" {rel['formula']} (置信度: {rel['accuracy']:.0%})") # Distribution shape summary for numeric columns shape_cols = [] for nc in num_cols[:3]: shape = nc.get('distribution_shape') if shape: shape_cols.append(f"{nc.get('inferred_label', nc['column_name'])}[{shape.get('concentration', 'N/A')}]") if shape_cols: lines.append(f"分布特征: {' | '.join(shape_cols)}") return '\n'.join(lines) if __name__ == '__main__': import sys if len(sys.argv) > 1: fp = sys.argv[1] try: df = pd.read_excel(fp) except Exception: df = pd.read_excel(fp, sheet_name=0) profile = profile_dataframe(df) print(generate_summary_text(profile)) issues = detect_data_issues(df) if issues: print(f"\n数据问题 ({len(issues)}):") for iss in issues: print(f" [{iss['severity']}] {iss['message']}")