Агенты для статистических сводок

import pandas as pd

df = pd.read_csv('sales_data.csv')

# Basic numeric summary
print(df.describe())

# Also describe categorical columns
print(df.describe(include='object'))

# Custom percentiles
print(df.describe(percentiles=[0.05, 0.25, 0.5, 0.75, 0.95]))

# Convert to clean dict for agent use
def get_numeric_summary(df):
    desc = df.describe().round(2)
    return {
        col: desc[col].to_dict()
        for col in desc.columns
    }

import pandas as pd
import numpy as np

def compute_correlation_matrix(df):
    numeric = df.select_dtypes(include='number')
    corr = numeric.corr().round(3)
    return corr.to_dict()

def find_strong_correlations(df, threshold=0.7):
    numeric = df.select_dtypes(include='number')
    corr = numeric.corr()
    strong = []
    cols = corr.columns
    for i in range(len(cols)):
        for j in range(i + 1, len(cols)):  # upper triangle only
            val = corr.iloc[i, j]
            if abs(val) >= threshold:
                strong.append({
                    'col_a': cols[i],
                    'col_b': cols[j],
                    'correlation': round(val, 3),
                    'direction': 'positive' if val > 0 else 'negative'
                })
    return sorted(strong, key=lambda x: abs(x['correlation']), reverse=True)

import pandas as pd

def detect_outliers_iqr(df, columns=None, multiplier=1.5):
    numeric = df.select_dtypes(include='number')
    if columns:
        numeric = numeric[columns]

    outlier_report = {}
    for col in numeric.columns:
        q1 = numeric[col].quantile(0.25)
        q3 = numeric[col].quantile(0.75)
        iqr = q3 - q1
        lower = q1 - multiplier * iqr
        upper = q3 + multiplier * iqr

        outliers = numeric[(numeric[col] < lower) | (numeric[col] > upper)][col]
        outlier_report[col] = {
            'q1': round(q1, 3),
            'q3': round(q3, 3),
            'iqr': round(iqr, 3),
            'lower_bound': round(lower, 3),
            'upper_bound': round(upper, 3),
            'outlier_count': len(outliers),
            'outlier_pct': round(len(outliers) / len(df) * 100, 1),
            'outlier_values': outliers.tolist()[:10]  # first 10
        }
    return outlier_report

import pandas as pd
from scipy import stats

def identify_distribution(series):
    '''Classify distribution type using skewness, kurtosis, and normality test.'''
    clean = series.dropna()
    if len(clean) < 8:
        return {'type': 'insufficient_data'}

    skewness = clean.skew()
    kurtosis = clean.kurtosis()

    # Shapiro-Wilk normality test (reliable up to ~5000 samples)
    sample = clean.sample(min(500, len(clean)), random_state=42)
    _, p_value = stats.shapiro(sample)

    distribution = 'normal' if p_value > 0.05 else 'non-normal'

    if abs(skewness) < 0.5 and p_value > 0.05:
        dist_type = 'normal'
    elif skewness > 1.0:
        dist_type = 'right-skewed (long right tail)'
    elif skewness < -1.0:
        dist_type = 'left-skewed (long left tail)'
    elif abs(skewness) < 0.5 and kurtosis < -1:
        dist_type = 'uniform-like'
    else:
        dist_type = 'approximately normal'

    return {
        'type': dist_type,
        'skewness': round(skewness, 3),
        'kurtosis': round(kurtosis, 3),
        'normality_pvalue': round(p_value, 4),
        'is_normal': p_value > 0.05
    }

from scipy import stats
import numpy as np

def test_normality(series, alpha=0.05):
    clean = series.dropna().values
    n = len(clean)
    results = {}

    # Shapiro-Wilk (best for n < 5000)
    if n <= 5000:
        stat, p = stats.shapiro(clean)
        results['shapiro_wilk'] = {
            'statistic': round(stat, 4),
            'p_value': round(p, 4),
            'is_normal': p > alpha
        }

    # D'Agostino-Pearson (n >= 20)
    if n >= 20:
        stat, p = stats.normaltest(clean)
        results['dagostino_pearson'] = {
            'statistic': round(stat, 4),
            'p_value': round(p, 4),
            'is_normal': p > alpha
        }

    # Kolmogorov-Smirnov
    mean, std = np.mean(clean), np.std(clean)
    stat, p = stats.kstest(clean, 'norm', args=(mean, std))
    results['kolmogorov_smirnov'] = {
        'statistic': round(stat, 4),
        'p_value': round(p, 4),
        'is_normal': p > alpha
    }

    return results

def full_statistical_profile(df):
    numeric = df.select_dtypes(include='number')
    profiles = {}

    for col in numeric.columns:
        series = numeric[col].dropna()
        profile = {
            'count': len(series),
            'null_count': df[col].isnull().sum(),
            'mean': round(series.mean(), 4),
            'median': round(series.median(), 4),
            'std': round(series.std(), 4),
            'min': round(series.min(), 4),
            'max': round(series.max(), 4),
            'range': round(series.max() - series.min(), 4),
            'q1': round(series.quantile(0.25), 4),
            'q3': round(series.quantile(0.75), 4)
        }
        profile['distribution'] = identify_distribution(series)
        outliers = detect_outliers_iqr(df, columns=[col])
        profile['outliers'] = outliers.get(col, {})
        profiles[col] = profile

    return {
        'profiles': profiles,
        'correlations': find_strong_correlations(df),
        'shape': {'rows': len(df), 'columns': len(df.columns)}
    }

import json

INSIGHT_PROMPT = '''You are a data analyst. Generate 3-5 key insights from this statistical summary.

Dataset: {dataset_name}
Statistical profile:
{profile_json}

For each insight:
- Be specific (use actual numbers)
- Focus on business relevance
- Flag any data quality concerns (high null counts, extreme outliers)
- Note unexpected patterns

Format as bullet points.'''

def generate_insights(df, dataset_name='dataset'):
    profile = full_statistical_profile(df)
    profile_text = json.dumps(profile, indent=2, default=str)[:3000]
    return llm_call(INSIGHT_PROMPT.format(
        dataset_name=dataset_name,
        profile_json=profile_text
    ))

# Example output:
# - Revenue column has 3.2% outliers; max value ($48,500) is 12x the median ($3,900)
# - Customer age follows a right-skewed distribution (skewness: 1.4) with most users under 35
# - Strong positive correlation (r=0.87) between session_count and lifetime_value

import numpy as np
from scipy import stats

def detect_trend(values):
    if len(values) < 3:
        return {'trend': 'insufficient_data'}

    x = np.arange(len(values))
    y = np.array(values, dtype=float)
    slope, intercept, r_value, p_value, std_err = stats.linregress(x, y)

    if p_value > 0.05:
        trend = 'flat (no significant trend)'
    elif slope > 0:
        trend = 'increasing'
    else:
        trend = 'decreasing'

    return {
        'trend': trend,
        'slope_per_period': round(slope, 4),
        'r_squared': round(r_value ** 2, 4),
        'p_value': round(p_value, 4),
        'significant': p_value < 0.05
    }

# Example: monthly revenue trend
monthly_revenue = [10000, 11200, 10800, 12500, 13100, 14200, 15000]
print(detect_trend(monthly_revenue))
# {'trend': 'increasing', 'slope_per_period': 834.2, 'r_squared': 0.95, ...}

from scipy import stats
import pandas as pd

def compare_groups(df, group_column, value_column):
    groups = df.groupby(group_column)[value_column].apply(list)
    group_names = list(groups.index)

    comparison = []
    for i, g1 in enumerate(group_names):
        for j, g2 in enumerate(group_names):
            if j <= i:
                continue
            a, b = groups[g1], groups[g2]
            mean_a = pd.Series(a).mean()
            mean_b = pd.Series(b).mean()

            # Mann-Whitney U (non-parametric, no normality assumption)
            stat, p = stats.mannwhitneyu(a, b, alternative='two-sided')
            pct_diff = (mean_a - mean_b) / mean_b * 100 if mean_b != 0 else 0

            comparison.append({
                'group_a': g1, 'group_b': g2,
                'mean_a': round(mean_a, 2),
                'mean_b': round(mean_b, 2),
                'pct_difference': round(pct_diff, 1),
                'p_value': round(p, 4),
                'significant': p < 0.05
            })
    return comparison

import pandas as pd
import numpy as np

def analyze_missing_values(df):
    total_rows = len(df)
    missing_report = {}

    for col in df.columns:
        null_count = df[col].isnull().sum()
        if null_count == 0:
            continue
        null_pct = null_count / total_rows * 100

        # Check if nulls correlate with another column (systematic pattern)
        pattern = 'random'
        if null_pct > 5:
            for other_col in df.select_dtypes(include='object').columns:
                if other_col == col:
                    continue
                null_rates = df.groupby(other_col)[col].apply(
                    lambda x: x.isnull().mean()
                )
                if null_rates.max() - null_rates.min() > 0.3:  # 30% difference
                    pattern = f'correlated_with_{other_col}'
                    break

        missing_report[col] = {
            'null_count': int(null_count),
            'null_pct': round(null_pct, 1),
            'severity': 'high' if null_pct > 20 else 'medium' if null_pct > 5 else 'low',
            'pattern': pattern
        }

    return missing_report