Реализация AI-автоматического профилирования данных
Профилирование данных — первый шаг любого дата-проекта: понять что есть в источнике, насколько данные чистые и как они распределены. AI-профилирование автоматизирует этот процесс и добавляет семантическое понимание: не просто "колонка содержит числа от 0 до 99999", а "вероятно, это возраст клиента, 15% нулей выглядят как заглушки".
Быстрый профилировщик
import pandas as pd
import numpy as np
from anthropic import Anthropic
class AIDataProfiler:
def __init__(self):
self.llm = Anthropic()
def profile(self, df: pd.DataFrame, dataset_name: str = "dataset") -> dict:
"""Полный профиль датасета за один вызов"""
profile = {
'name': dataset_name,
'shape': {'rows': len(df), 'columns': len(df.columns)},
'memory_mb': df.memory_usage(deep=True).sum() / 1024**2,
'columns': {},
'correlations': {},
'issues': []
}
# Профиль каждой колонки
for col in df.columns:
profile['columns'][col] = self._profile_column(df[col])
# Корреляции между числовыми
num_cols = df.select_dtypes(include='number').columns
if len(num_cols) > 1:
corr = df[num_cols].corr()
# Только значимые (|r| > 0.5)
for i in range(len(num_cols)):
for j in range(i+1, len(num_cols)):
r = corr.iloc[i, j]
if abs(r) > 0.5:
profile['correlations'][f"{num_cols[i]} × {num_cols[j]}"] = round(r, 3)
# Системные проблемы
profile['issues'] = self._detect_issues(df, profile)
# AI-интерпретация
profile['ai_summary'] = self._generate_summary(profile)
profile['semantic_types'] = self._detect_semantic_types(df, profile)
return profile
def _profile_column(self, series: pd.Series) -> dict:
null_count = int(series.isnull().sum())
total = len(series)
base = {
'dtype': str(series.dtype),
'null_count': null_count,
'null_pct': round(null_count / total * 100, 2),
'unique_count': int(series.nunique()),
'unique_pct': round(series.nunique() / total * 100, 2)
}
if pd.api.types.is_numeric_dtype(series):
non_null = series.dropna()
base.update({
'min': float(non_null.min()) if len(non_null) > 0 else None,
'max': float(non_null.max()) if len(non_null) > 0 else None,
'mean': float(non_null.mean()) if len(non_null) > 0 else None,
'median': float(non_null.median()) if len(non_null) > 0 else None,
'std': float(non_null.std()) if len(non_null) > 1 else None,
'zeros_pct': round((non_null == 0).sum() / total * 100, 2),
'negatives_pct': round((non_null < 0).sum() / total * 100, 2),
'q1': float(non_null.quantile(0.25)) if len(non_null) > 0 else None,
'q3': float(non_null.quantile(0.75)) if len(non_null) > 0 else None,
})
else:
top5 = series.value_counts().head(5)
base.update({
'top_values': top5.to_dict(),
'top_value_pct': round(top5.iloc[0] / total * 100, 2) if len(top5) > 0 else 0,
'avg_length': round(series.dropna().astype(str).str.len().mean(), 1),
'min_length': int(series.dropna().astype(str).str.len().min()) if null_count < total else None,
'max_length': int(series.dropna().astype(str).str.len().max()) if null_count < total else None,
})
return base
def _detect_issues(self, df: pd.DataFrame, profile: dict) -> list[str]:
issues = []
for col, stats in profile['columns'].items():
# Много пропусков
if stats['null_pct'] > 20:
issues.append(f"HIGH NULLS: {col} has {stats['null_pct']:.1f}% missing values")
# Только одно значение
if stats['unique_count'] == 1:
issues.append(f"CONSTANT: {col} has only one unique value")
# Почти уникальные строки (возможный ID)
if stats['unique_pct'] > 95 and stats['dtype'] == 'object':
issues.append(f"HIGH CARDINALITY: {col} may be an ID column ({stats['unique_pct']:.0f}% unique)")
# Много нулей в числовых
if stats.get('zeros_pct', 0) > 30:
issues.append(f"MANY ZEROS: {col} has {stats['zeros_pct']:.1f}% zero values (possible fill)")
# Дубликаты строк
dup_count = df.duplicated().sum()
if dup_count > 0:
issues.append(f"DUPLICATES: {dup_count} duplicate rows ({dup_count/len(df)*100:.1f}%)")
return issues
def _detect_semantic_types(self, df: pd.DataFrame, profile: dict) -> dict:
"""LLM-определение семантических типов колонок"""
col_descriptions = []
for col, stats in profile['columns'].items():
if stats['dtype'] == 'object':
sample = list(stats.get('top_values', {}).keys())[:3]
col_descriptions.append(f"- {col}: samples={sample}")
else:
col_descriptions.append(
f"- {col}: min={stats.get('min')}, max={stats.get('max')}, "
f"mean={stats.get('mean')}"
)
response = self.llm.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=400,
messages=[{
"role": "user",
"content": f"""Identify semantic types for these columns.
{chr(10).join(col_descriptions)}
Return JSON: {{"column_name": "semantic_type"}}
Types: id, name, email, phone, address, date, timestamp, currency, age, percentage, category, status, text, url, country, city"""
}]
)
try:
import json
return json.loads(response.content[0].text)
except Exception:
return {}
def _generate_summary(self, profile: dict) -> str:
"""AI-резюме профиля датасета"""
issues_str = "\n".join(profile['issues'][:5]) if profile['issues'] else "No critical issues"
cols_summary = {col: {
'dtype': s['dtype'],
'null_pct': s['null_pct'],
'unique_pct': s['unique_pct']
} for col, s in list(profile['columns'].items())[:10]}
response = self.llm.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=300,
messages=[{
"role": "user",
"content": f"""Summarize this dataset profile in 3-5 sentences for a data engineer.
Focus on data quality, potential issues, and readiness for ML.
Shape: {profile['shape']}
Issues: {issues_str}
Column sample: {cols_summary}"""
}]
)
return response.content[0].text
Автоматическое профилирование 100-колоночного датасета занимает 30-60 секунд. Без инструмента — то же самое вручную: 2-3 часа. AI-интерпретация снижает время понимания нового источника данных с нескольких часов до 5-10 минут.