forked from cleveragents/cleveragents-core
aditya
e5a142ef76
The generate prompt only received the analyze_requirements summary,
which often compressed the file-by-file DoD into a brief paragraph.
The LLM then produced 1 file instead of the 5-6 specified.
Fix: added {original_prompt} variable to the generate prompt template
so the LLM sees both the architecture analysis AND the original DoD
with explicit file listings. Also strengthened the instruction to
"output ALL files mentioned in the Definition of Done."
Verified: sim10 (CSV Analyzer) now produces 6 files matching DoD
(loader.py, analyzer.py, reporter.py, cli.py, requirements.txt).
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""Statistical analysis, outlier detection, and correlation computation."""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from scipy import stats
|
|
from typing import Dict, List, Any, Tuple, Optional
|
|
import warnings
|
|
|
|
warnings.filterwarnings('ignore', category=RuntimeWarning)
|
|
|
|
|
|
class DataAnalyzer:
|
|
"""Performs comprehensive statistical analysis on DataFrame."""
|
|
|
|
def __init__(self, df: pd.DataFrame):
|
|
self.df = df
|
|
self.numeric_columns = self._get_numeric_columns()
|
|
self.categorical_columns = self._get_categorical_columns()
|
|
self.datetime_columns = self._get_datetime_columns()
|
|
|
|
def _get_numeric_columns(self) -> List[str]:
|
|
"""Get list of numeric columns."""
|
|
return [col for col in self.df.columns if pd.api.types.is_numeric_dtype(self.df[col])]
|
|
|
|
def _get_categorical_columns(self) -> List[str]:
|
|
"""Get list of categorical/string columns."""
|
|
return [col for col in self.df.columns
|
|
if pd.api.types.is_string_dtype(self.df[col]) or
|
|
pd.api.types.is_categorical_dtype(self.df[col]) or
|
|
self.df[col].dtype == 'object']
|
|
|
|
def _get_datetime_columns(self) -> List[str]:
|
|
"""Get list of datetime columns."""
|
|
return [col for col in self.df.columns if pd.api.types.is_datetime64_any_dtype(self.df[col])]
|
|
|
|
def compute_basic_info(self) -> Dict[str, Any]:
|
|
"""Compute basic information about the dataset."""
|
|
return {
|
|
'total_rows': len(self.df),
|
|
'total_columns': len(self.df.columns),
|
|
'numeric_columns': len(self.numeric_columns),
|
|
'categorical_columns': len(self.categorical_columns),
|
|
'datetime_columns': len(self.datetime_columns),
|
|
'memory_usage_mb': round(self.df.memory_usage(deep=True).sum() / (1024 * 1024), 2),
|
|
'missing_values_total': self.df.isnull().sum().sum(),
|
|
'duplicate_rows': self.df.duplicated().sum()
|
|
}
|
|
|
|
def compute_summary_statistics(self) -> Dict[str, Dict[str, Any]]:
|
|
"""Compute comprehensive summary statistics for all columns."""
|
|
stats_dict = {}
|
|
|
|
# Numeric columns
|
|
for col in self.numeric_columns:
|
|
series = self.df[col].dropna()
|
|
if len(series) == 0:
|
|
stats_dict[col] = {'type': 'numeric', 'error': 'No non-null values'}
|
|
continue
|
|
|
|
try:
|
|
# Calculate mode safely
|
|
mode_result = stats.mode(series, keepdims=False, nan_policy='omit')
|
|
mode_value = mode_result.mode if hasattr(mode_result, 'mode') else mode_result[0]
|
|
|
|
# Handle case where mode might be an array
|
|
if isinstance(mode_value, np.ndarray):
|
|
mode_value = mode_value[0] if len(mode_value) > 0 else np.nan
|
|
|
|
quartiles = series.quantile([0.25, 0.5, 0.75])
|
|
|
|
stats_dict[col] = {
|
|
'type': 'numeric',
|
|
'count': len(series),
|
|
'missing': self.df[col].isnull().sum(),
|
|
'mean': round(series.mean(), 4),
|
|
'median': round(series.median(), 4),
|
|
'mode': round(float(mode_value), 4) if not pd.isna(mode_value) else None,
|
|
'std_dev': round(series.std(), 4),
|
|
'variance': round(series.var(), 4),
|
|
'min': round(series.min(), 4),
|
|
'max': round(series.max(), 4),
|
|
'q1': round(quartiles[0.25], 4),
|
|
'q3': round(quartiles[0.75], 4),
|
|
'iqr': round(quartiles[0.75] - quartiles[0.25], 4),
|
|
'skewness': round(series.skew(), 4),
|
|
'kurtosis': round(series.kurtosis(), 4)
|
|
}
|
|
except Exception as e:
|
|
stats_dict[col] = {'type': 'numeric', 'error': str(e)}
|
|
|
|
# Categorical columns
|
|
for col in self.categorical_columns:
|
|
series = self.df[col].dropna()
|
|
if len(series) == 0:
|
|
stats_dict[col] = {'type': 'categorical', 'error': 'No non-null values'}
|
|
continue
|
|
|
|
try:
|
|
value_counts = series.value_counts()
|
|
stats_dict[col] = {
|
|
'type': 'categorical',
|
|
'count': len(series),
|
|
'missing': self.df[col].isnull().sum(),
|
|
'unique_values': series.nunique(),
|
|
'most_frequent': value_counts.index[0] if len(value_counts) > 0 else None,
|
|
'most_frequent_count': value_counts.iloc[0] if len(value_counts) > 0 else 0,
|
|
'least_frequent': value_counts.index[-1] if len(value_counts) > 0 else None,
|
|
'least_frequent_count': value_counts.iloc[-1] if len(value_counts) > 0 else 0,
|
|
'top_5_values': dict(value_counts.head().items())
|
|
}
|
|
except Exception as e:
|
|
stats_dict[col] = {'type': 'categorical', 'error': str(e)}
|
|
|
|
# Datetime columns
|
|
for col in self.datetime_columns:
|
|
series = self.df[col].dropna()
|
|
if len(series) == 0:
|
|
stats_dict[col] = {'type': 'datetime', 'error': 'No non-null values'}
|
|
continue
|
|
|
|
try:
|
|
stats_dict[col] = {
|
|
'type': 'datetime',
|
|
'count': len(series),
|
|
'missing': self.df[col].isnull().sum(),
|
|
'min_date': series.min().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'max_date': series.max().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'date_range_days': (series.max() - series.min()).days,
|
|
'unique_dates': series.nunique()
|
|
}
|
|
except Exception as e:
|
|
stats_dict[col] = {'type': 'datetime', 'error': str(e)}
|
|
|
|
return stats_dict
|
|
|
|
def detect_outliers_iqr(self) -> Dict[str, Dict[str, Any]]:
|
|
"""Detect outliers using Interquartile Range (IQR) method."""
|
|
outliers_dict = {}
|
|
|
|
for col in self.numeric_columns:
|
|
series = self.df[col].dropna()
|
|
if len(series) < 4: # Need at least 4 values for quartiles
|
|
outliers_dict[col] = {'error': 'Insufficient data for outlier detection'}
|
|
continue
|
|
|
|
try:
|
|
q1 = series.quantile(0.25)
|
|
q3 = series.quantile(0.75)
|
|
iqr = q3 - q1
|
|
|
|
lower_bound = q1 - 1.5 * iqr
|
|
upper_bound = q3 + 1.5 * iqr
|
|
|
|
outliers = series[(series < lower_bound) | (series > upper_bound)]
|
|
outlier_indices = self.df[col][(self.df[col] < lower_bound) | (self.df[col] > upper_bound)].index.tolist()
|
|
|
|
outliers_dict[col] = {
|
|
'lower_bound': round(lower_bound, 4),
|
|
'upper_bound': round(upper_bound, 4),
|
|
'outlier_count': len(outliers),
|
|
'outlier_percentage': round((len(outliers) / len(series)) * 100, 2),
|
|
'outlier_values': [round(x, 4) for x in sorted(outliers.values)],
|
|
'outlier_indices': outlier_indices[:20] # Limit to first 20 indices
|
|
}
|
|
except Exception as e:
|
|
outliers_dict[col] = {'error': str(e)}
|
|
|
|
return outliers_dict
|
|
|
|
def compute_correlation_matrix(self) -> Optional[Dict[str, Any]]:
|
|
"""Compute correlation matrix for numeric columns."""
|
|
if len(self.numeric_columns) < 2:
|
|
return {'error': 'Need at least 2 numeric columns for correlation analysis'}
|
|
|
|
try:
|
|
numeric_df = self.df[self.numeric_columns].select_dtypes(include=[np.number])
|
|
|
|
# Remove columns with no variance
|
|
numeric_df = numeric_df.loc[:, numeric_df.var() != 0]
|
|
|
|
if len(numeric_df.columns) < 2:
|
|
return {'error': 'Need at least 2 numeric columns with variance for correlation'}
|
|
|
|
# Compute correlations
|
|
pearson_corr = numeric_df.corr(method='pearson')
|
|
spearman_corr = numeric_df.corr(method='spearman')
|
|
|
|
# Find strong correlations (> 0.7 or < -0.7)
|
|
strong_correlations = []
|
|
for i in range(len(pearson_corr.columns)):
|
|
for j in range(i + 1, len(pearson_corr.columns)):
|
|
col1, col2 = pearson_corr.columns[i], pearson_corr.columns[j]
|
|
pearson_val = pearson_corr.iloc[i, j]
|
|
spearman_val = spearman_corr.iloc[i, j]
|
|
|
|
if abs(pearson_val) > 0.7 and not pd.isna(pearson_val):
|
|
strong_correlations.append({
|
|
'column1': col1,
|
|
'column2': col2,
|
|
'pearson': round(pearson_val, 4),
|
|
'spearman': round(spearman_val, 4)
|
|
})
|
|
|
|
return {
|
|
'pearson_correlation': pearson_corr.round(4).to_dict(),
|
|
'spearman_correlation': spearman_corr.round(4).to_dict(),
|
|
'strong_correlations': strong_correlations,
|
|
'columns_analyzed': list(numeric_df.columns)
|
|
}
|
|
|
|
except Exception as e:
|
|
return {'error': f'Error computing correlations: {str(e)}'}
|
|
|
|
def analyze_missing_patterns(self) -> Dict[str, Any]:
|
|
"""Analyze missing data patterns."""
|
|
missing_data = {}
|
|
|
|
# Missing values per column
|
|
missing_per_column = self.df.isnull().sum()
|
|
missing_percentages = (missing_per_column / len(self.df) * 100).round(2)
|
|
|
|
# Columns with missing data
|
|
columns_with_missing = missing_per_column[missing_per_column > 0].to_dict()
|
|
|
|
# Missing data patterns
|
|
missing_combinations = self.df.isnull().value_counts().head(10)
|
|
|
|
return {
|
|
'total_missing_values': missing_per_column.sum(),
|
|
'missing_percentage_overall': round((missing_per_column.sum() / (len(self.df) * len(self.df.columns))) * 100, 2),
|
|
'columns_with_missing': dict(zip(columns_with_missing.keys(),
|
|
[{'count': int(count), 'percentage': float(missing_percentages[col])}
|
|
for col, count in columns_with_missing.items()])),
|
|
'complete_rows': len(self.df.dropna()),
|
|
'complete_rows_percentage': round((len(self.df.dropna()) / len(self.df)) * 100, 2)
|
|
}
|
|
|
|
def run_full_analysis(self) -> Dict[str, Any]:
|
|
"""Run complete analysis and return all results."""
|
|
return {
|
|
'basic_info': self.compute_basic_info(),
|
|
'summary_statistics': self.compute_summary_statistics(),
|
|
'outliers': self.detect_outliers_iqr(),
|
|
'correlations': self.compute_correlation_matrix(),
|
|
'missing_data': self.analyze_missing_patterns()
|
|
} |