ChatGPT Data Analysis：数据分析

class SandboxExecutor:    def __init__(self):        self.files = {}        self.results = []        self.max_memory_mb = 512        def upload_file(self, filename, content):        self.files[filename] = content        return f'File {filename} uploaded ({len(content)} bytes)'        def nl_to_code(self, intent, data_schema):        intent_map = {            'distribution': 'df.hist()',            'correlation': 'df.corr()',            'summary': 'df.describe()',            'missing': 'df.isnull().sum()'        }        code_lines = ['import pandas as pd']        for key, code in intent_map.items():            if key in intent.lower():                code_lines.append(f'df = pd.read_csv("data.csv")')                code_lines.append(code)        return '\n'.join(code_lines) if len(code_lines) > 1 else 'df.describe()'        def execute(self, code):        return {'status': 'success', 'output': f'Executed: {code[:50]}...'}        def cleanup(self):        self.files.clear()        self.results.clear() sandbox = SandboxExecutor()print(sandbox.upload_file('data.csv', 'col1,col2\n1,2\n3,4'))print(sandbox.nl_to_code('show distribution of features', {}))

class AutoVisualizer:    def __init__(self):        self.chart_rules = {            'numeric_single': 'histogram',            'numeric_numeric': 'scatter',            'categorical_numeric': 'bar',            'time_numeric': 'line',            'categorical_single': 'pie'        }        def select_chart(self, data_types, n_vars):        if n_vars == 1:            return self.chart_rules.get(f'{data_types[0]}_single', 'histogram')        elif n_vars == 2:            return self.chart_rules.get(f'{data_types[0]}_{data_types[1]}', 'scatter')        return 'heatmap'        def adjust_view(self, current_chart, adjustment):        adjustments = {            'group_by': f'{current_chart} grouped by new dimension',            'time_granularity': f'{current_chart} with adjusted time granularity',            'remove_outliers': f'{current_chart} without outliers',            'change_color': f'{current_chart} with updated color scheme'        }        for key, val in adjustments.items():            if key in adjustment.lower():                return val        return current_chart viz = AutoVisualizer()print(viz.select_chart(['numeric'], 1))print(viz.select_chart(['categorical', 'numeric'], 2))print(viz.adjust_view('bar chart', 'group by region'))

class StatisticalAnalyzer:    def __init__(self, data):        self.data = data        def descriptive_stats(self, column):        values = [row[column] for row in self.data if column in row]        n = len(values)        mean = sum(values) / n if n > 0 else 0        sorted_v = sorted(values)        median = sorted_v[n // 2] if n > 0 else 0        variance = sum((x - mean) ** 2 for x in values) / n if n > 0 else 0        return {'n': n, 'mean': round(mean, 2), 'median': median, 'std': round(variance ** 0.5, 2)}        def select_test(self, data_type, n_groups, normal=True):        if data_type == 'continuous' and n_groups == 2:            return 't_test' if normal else 'mann_whitney'        elif data_type == 'continuous' and n_groups > 2:            return 'anova' if normal else 'kruskal_wallis'        elif data_type == 'categorical':            return 'chi_square'        return 'descriptive'        def interpret_result(self, test_name, p_value, alpha=0.05):        sig = '显著' if p_value < alpha else '不显著'        return f'{test_name}: p={p_value:.4f}，结果{sig}（alpha={alpha}）' data = [{'x': 1}, {'x': 2}, {'x': 3}, {'x': 4}, {'x': 5}]analyzer = StatisticalAnalyzer(data)print(analyzer.descriptive_stats('x'))print(analyzer.select_test('continuous', 2, normal=True))print(analyzer.interpret_result('t_test', 0.03))

class DataCleaner:    def __init__(self, data):        self.data = data        self.report = {}        def detect_missing(self, columns):        for col in columns:            missing = sum(1 for row in self.data if col not in row or row[col] is None)            self.report[col] = {'missing': missing, 'rate': round(missing / len(self.data), 2)}        return self.report        def detect_outliers_iqr(self, values):        sorted_v = sorted(values)        n = len(sorted_v)        q1 = sorted_v[n // 4]        q3 = sorted_v[3 * n // 4]        iqr = q3 - q1        lower = q1 - 1.5 * iqr        upper = q3 + 1.5 * iqr        outliers = [v for v in values if v < lower or v > upper]        return {'lower': lower, 'upper': upper, 'outliers': outliers}        def apply_rule(self, rule_text):        rules = {            'fill_median': 'Fill missing with median',            'remove_duplicates': 'Remove duplicate rows',            'normalize_date': 'Normalize date format to YYYY-MM-DD',            'strip_whitespace': 'Remove leading/trailing whitespace'        }        applied = []        for key, desc in rules.items():            if key in rule_text.lower():                applied.append(desc)        return applied cleaner = DataCleaner([{'a': 1, 'b': 2}, {'a': None, 'b': 3}, {'a': 5}])print(cleaner.detect_missing(['a', 'b']))print(cleaner.detect_outliers_iqr([1, 2, 3, 4, 5, 100]))print(cleaner.apply_rule('fill_median and remove_duplicates'))

class AnalysisToolComparison:    def __init__(self):        self.tools = {            'chatgpt': {'setup_time': 'minutes', 'flexibility': 'high', 'reproducibility': 'low', 'skill_required': 'natural language'},            'bi_tool': {'setup_time': 'days', 'flexibility': 'medium', 'reproducibility': 'high', 'skill_required': 'SQL + drag-drop'},            'code_script': {'setup_time': 'hours', 'flexibility': 'high', 'reproducibility': 'high', 'skill_required': 'Python/R'}        }        def recommend(self, scenario):        if scenario == 'exploration':            return 'chatgpt'        elif scenario == 'monitoring':            return 'bi_tool'        elif scenario == 'research':            return 'code_script'        return 'chatgpt'        def lifecycle(self):        return {'explore': 'chatgpt', 'validate': 'code_script', 'productionize': 'bi_tool'} comp = AnalysisToolComparison()print(f'Exploration: {comp.recommend("exploration")}')print(f'Monitoring: {comp.recommend("monitoring")}')print(f'Lifecycle: {comp.lifecycle()}')

class LimitChecker:    def __init__(self):        self.limits = {'memory_mb': 512, 'timeout_sec': 60, 'file_size_mb': 512, 'max_rows': 100000}        def check_feasibility(self, data_size_mb, estimated_time_sec, n_rows):        issues = []        if data_size_mb > self.limits['file_size_mb']:            issues.append(f'File too large: {data_size_mb}MB > {self.limits["file_size_mb"]}MB')        if estimated_time_sec > self.limits['timeout_sec']:            issues.append(f'Timeout risk: {estimated_time_sec}s > {self.limits["timeout_sec"]}s')        if n_rows > self.limits['max_rows']:            issues.append(f'Too many rows: {n_rows} > {self.limits["max_rows"]}')        return {'feasible': len(issues) == 0, 'issues': issues}        def privacy_check(self, data_type):        sensitive = ['pii', 'medical', 'financial', 'password', 'credit_card']        return {'sensitive': data_type.lower() in sensitive, 'action': 'Must anonymize before upload' if data_type.lower() in sensitive else 'OK to upload'} checker = LimitChecker()print(checker.check_feasibility(600, 30, 50000))print(checker.privacy_check('medical'))

class ChatGPTDataAnalysis:    def __init__(self):        self.sandbox = {'memory_mb': 512, 'timeout_sec': 60}        self.uploaded_files = {}        self.conversation = []        def upload(self, filename, data):        self.uploaded_files[filename] = data        return {'status': 'uploaded', 'rows': len(data), 'columns': list(data[0].keys()) if data else []}        def analyze(self, query):        self.conversation.append({'role': 'user', 'content': query})        if 'distribution' in query.lower():            return {'action': 'generate_histogram', 'interpretation': 'Data shows normal distribution with mean=50'}        elif 'correlation' in query.lower():            return {'action': 'compute_correlation', 'interpretation': 'Feature A and B have r=0.65, p<0.05'}        elif 'clean' in query.lower():            return {'action': 'data_cleaning', 'interpretation': 'Found 5% missing values, filled with median'}        return {'action': 'describe', 'interpretation': 'Summary statistics generated'}        def check_limit(self, data_size_mb):        return data_size_mb <= self.sandbox['memory_mb']        def privacy_assess(self, has_pii):        if has_pii:            return {'safe': False, 'action': 'Anonymize before upload'}        return {'safe': True, 'action': 'OK to proceed'}        def compare_with_bi(self, use_case):        if use_case == 'exploration':            return 'ChatGPT: faster for ad-hoc analysis'        elif use_case == 'monitoring':            return 'BI tool: better for recurring dashboards'        return 'Complementary tools' agent = ChatGPTDataAnalysis()data = [{'age': 25, 'income': 50000}, {'age': 30, 'income': 60000}]print(agent.upload('users.csv', data))print(agent.analyze('show distribution of income'))print(agent.check_limit(200))print(agent.privacy_assess(True))print(agent.compare_with_bi('exploration'))