# Data Analysis Agents

Agents specialized for data cleaning, exploration, visualization, statistical analysis, and insight generation.

---

## Data Analysis Agent Capabilities

| Capability | Description | Tools |
|---|---|---|
| **Data Ingestion** | Load from files, databases, APIs | pandas, polars, DuckDB, SQLAlchemy |
| **Cleaning** | Handle missing, outliers, types | pandas, Great Expectations, pandera |
| **Exploration** | Summary stats, distributions, correlations | pandas, ydata-profiling, sweetviz |
| **Visualization** | Charts, dashboards, interactive | plotly, altair, matplotlib, seaborn |
| **Statistical Analysis** | Tests, regression, ANOVA, time series | statsmodels, scipy, pingouin |
| **ML Analysis** | Clustering, classification, forecasting | scikit-learn, XGBoost, Prophet |
| **Reporting** | Notebooks, dashboards, PDF/HTML | Jupyter, Quarto, Streamlit, Evidence |
| **SQL Analysis** | Query, optimize, explain | DuckDB, SQLGlot, pgAdmin |
| **Causal Analysis** | A/B tests, causal inference | DoWhy, CausalML, EconML |
| **Time Series** | Forecasting, decomposition, anomaly | Prophet, NeuralForecast, statsforecast |

---

## Data Analysis Agent Architectures

### 1. Exploratory Data Analysis (EDA) Agent
```python
class EDAAgent:
    def __init__(self):
        self.profiler = DataProfiler()
        self.visualizer = Visualizer()
        self.stat_tester = StatisticalTester()
    
    async def analyze(self, dataset: Dataset, config: EDAConfig = None) -> EDAReport:
        config = config or EDAConfig()
        
        # 1. Basic profiling
        profile = await self.profiler.profile(dataset)
        
        # 2. Data quality assessment
        quality = await self.assess_quality(dataset)
        
        # 3. Statistical summaries
        summaries = await self.compute_summaries(dataset)
        
        # 4. Correlation analysis
        correlations = await self.analyze_correlations(dataset)
        
        # 5. Distribution analysis
        distributions = await self.analyze_distributions(dataset)
        
        # 6. Outlier detection
        outliers = await self.detect_outliers(dataset)
        
        # 7. Missing data patterns
        missing = await self.analyze_missing(dataset)
        
        # 8. Generate visualizations
        viz = await self.generate_visualizations(dataset, config)
        
        return EDAReport(
            profile=profile,
            quality=quality,
            summaries=summaries,
            correlations=correlations,
            distributions=distributions,
            outliers=outliers,
            missing=missing,
            visualizations=viz,
            recommendations=self.generate_recommendations(dataset)
        )
    
    async def profile(self, dataset: Dataset) -> DataProfile:
        return DataProfile(
            rows=len(dataset),
            columns=len(dataset.columns),
            dtypes=dataset.dtypes.to_dict(),
            memory_usage=dataset.memory_usage(deep=True).sum(),
            sample=dataset.head(5).to_dict()
        )
```

### 2. Statistical Analysis Agent
```python
class StatisticalAnalysisAgent:
    def __init__(self):
        self.test_selector = TestSelector()
        self.power_analyzer = PowerAnalyzer()
        self.effect_size = EffectSizeCalculator()
    
    async def analyze(self, dataset: Dataset, question: ResearchQuestion) -> StatisticalReport:
        # 1. Select appropriate test
        test = self.test_selector.select(question, dataset)
        
        # 2. Check assumptions
        assumptions = await self.check_assumptions(dataset, test)
        
        # 3. Run test
        result = await self.run_test(dataset, test)
        
        # 4. Effect size
        effect = await self.effect_size.calculate(result, test)
        
        # 5. Power analysis
        power = await self.power_analyzer.post_hoc(result, test)
        
        # 6. Interpretation
        interpretation = await self.interpret(result, test, assumptions)
        
        return StatisticalReport(
            test=test,
            result=result,
            assumptions=assumptions,
            effect_size=effect,
            power=power,
            interpretation=interpretation,
            visualization=self.create_visualization(result, test)
        )
    
    def select_test(self, question: ResearchQuestion, dataset: Dataset) -> TestType:
        # Decision tree for test selection
        if question.type == "compare_means":
            if question.groups == 2:
                return TestType.T_TEST if question.paired else TestType.T_TEST_INDEP
            else:
                return TestType.ANOVA
        elif question.type == "compare_proportions":
            return TestType.CHI_SQUARE
        elif question.type == "correlation":
            return TestType.PEARSON if question.linear else TestType.SPEARMAN
        elif question.type == "regression":
            return TestType.LINEAR_REGRESSION
        ...
```

### 3. Causal Analysis Agent
```python
class CausalAnalysisAgent:
    def __init__(self):
        self.do_why = DoWhyTools()
        self.causalml = CausalMLTools()
        self.econml = EconMLTools()
    
    async def estimate_causal_effect(self, 
                                     data: Dataset, 
                                     treatment: str, 
                                     outcome: str,
                                     confounders: list[str]) -> CausalResult:
        # 1. Causal graph
        graph = self.build_causal_graph(treatment, outcome, confounders)
        
        # 2. Identify estimand
        estimand = self.do_why.identify_effect(graph, treatment, outcome)
        
        # 2. Estimate (multiple methods)
        estimates = await asyncio.gather(
            self.do_why.estimate(estimand, method="backdoor"),
            self.causalml.estimate(data, treatment, outcome, confounders),
            self.econml.estimate(data, treatment, outcome, confounders)
        )
        
        # 3. Sensitivity analysis
        sensitivity = self.do_why.sensitivity_analysis(estimand)
        
        # 4. Robustness checks
        robustness = await self.robustness_checks(data, treatment, outcome)
        
        return CausalResult(
            estimates=estimates,
            consensus=self.aggregate_estimates(estimates),
            sensitivity=sensitivity,
            robustness=robustness,
            assumptions=self.list_assumptions(graph)
        )
    
    async def ab_test_analysis(self, 
                               control: Dataset, 
                               treatment: Dataset,
                               metric: str) -> ABTestResult:
        # Bayesian A/B test
        result = await self.bayesian_ab_test(control, treatment, metric)
        
        # Frequentist
        freq_result = self.frequentist_test(control, treatment, metric)
        
        # Sequential testing
        sequential = self.sequential_test(control, treatment, metric)
        
        return ABTestResult(
            bayesian=result,
            frequentist=freq_result,
            sequential=sequential,
            recommendation=self.make_recommendation(result, freq_result)
        )
```

### 4. Time Series Analysis Agent
```python
class TimeSeriesAgent:
    def __init__(self):
        self.forecasters = {
            "prophet": ProphetForecaster(),
            "neural": NeuralForecaster(),
            "statistical": StatisticalForecaster(),
            "ensemble": EnsembleForecaster()
        }
        self.decomposer = TimeSeriesDecomposer()
        self.anomaly_detector = AnomalyDetector()
    
    async def forecast(self, 
                       series: TimeSeries, 
                       horizon: int,
                       config: ForecastConfig = None) -> ForecastResult:
        config = config or ForecastConfig()
        
        # 1. Decompose
        decomposition = await self.decomposer.decompose(series)
        
        # 2. Check stationarity
        stationary = await self.check_stationarity(series)
        
        # 3. Model selection
        if config.auto_select:
            best_model = await self.select_model(series)
        else:
            best_model = config.model
        
        # 4. Train & forecast
        forecast = await self.forecasters[best_model].forecast(series, horizon)
        
        # 4. Uncertainty quantification
        prediction_intervals = self.compute_intervals(forecast, config.confidence)
        
        # 5. Backtesting
        backtest = await self.backtest(series, best_model, config)
        
        return ForecastResult(
            model=best_model,
            forecast=forecast,
            intervals=prediction_intervals,
            backtest=backtest,
            decomposition=decomposition
        )
    
    async def detect_anomalies(self, series: TimeSeries) -> AnomalyReport:
        # Statistical
        stat_anomalies = self.anomaly_detector.statistical(series)
        
        # ML-based
        ml_anomalies = await self.anomaly_detector.ml_based(series)
        
        # Seasonal decomposition residuals
        seasonal_anomalies = self.anomaly_detector.seasonal_residuals(series)
        
        return AnomalyReport(
            statistical=stat_anomalies,
            ml=ml_anomalies,
            seasonal=seasonal_anomalies,
            consensus=self.consensus_anomalies([
                stat_anomalies, ml_anomalies, seasonal_anomalies
            ])
        )
```

---

## Specialized Data Domains

### 1. Business Intelligence Agent
```python
class BIAnalysisAgent:
    def __init__(self):
        self.sql_engine = DuckDBEngine()
        self.kpi_library = KPILibrary()
    
    async def generate_dashboard(self, spec: DashboardSpec) -> Dashboard:
        # 1. Define metrics
        metrics = [self.kpi_library.get(m) for m in spec.metrics]
        
        # 2. Build queries
        queries = [self.build_query(m, spec.filters, spec.time_range) for m in metrics]
        
        # 3. Execute
        results = await asyncio.gather(*[
            self.sql_engine.query(q) for q in queries
        ])
        
        # 4. Build visualizations
        charts = [self.create_chart(m, r) for m, r in zip(metrics, results)]
        
        # 4. Layout
        layout = self.create_layout(charts, spec.layout)
        
        return Dashboard(
            metrics=metrics,
            charts=charts,
            layout=layout,
            filters=spec.filters,
            refresh_interval=spec.refresh_interval
        )
    
    async def ad_hoc_query(self, question: str) -> QueryResult:
        # NL to SQL
        sql = await self.nl_to_sql(question)
        
        # Validate
        validated = self.validate_sql(sql)
        
        # Execute
        result = await self.sql_engine.query(validated)
        
        # Explain
        explanation = await self.explain_result(result, question)
        
        return QueryResult(sql=sql, data=result, explanation=explanation)
```

### 2. Customer Analytics Agent
```python
class CustomerAnalyticsAgent:
    def __init__(self):
        self.segmenter = CustomerSegmenter()
        self.clv = CLVCalculator()
        self.churn = ChurnPredictor()
        self.rfm = RFMAnalyzer()
    
    async def analyze_cohort(self, cohort: CohortDefinition) -> CohortAnalysis:
        # 1. Define cohort
        customers = await self.get_cohort(cohort)
        
        # 2. RFM analysis
        rfm = await self.rfm.analyze(customers)
        
        # 3. Segmentation
        segments = await self.segmenter.segment(customers)
        
        # 4. CLV
        clv = await self.clv.calculate(customers)
        
        # 5. Churn risk
        churn = await self.churn.predict(customers)
        
        # 6. LTV:CAC
        ltv_cac = await self.ltv_cac_ratio(customers)
        
        return CohortAnalysis(
            rfm=rfm,
            segments=segments,
            clv=clv,
            churn=churn,
            ltv_cac=ltv_cac
        )
```

### 3. Financial Analysis Agent
```python
class FinancialAnalysisAgent:
    def __init__(self):
        self.ratios = FinancialRatios()
        self.valuation = ValuationModels()
        self.forecast = FinancialForecaster()
    
    async def analyze_company(self, ticker: str, period: Period) -> FinancialReport:
        # 1. Fetch data
        financials = await self.fetch_financials(ticker, period)
        
        # 2. Ratio analysis
        ratios = self.ratios.compute_all(financials)
        
        # 3. Trend analysis
        trends = self.analyze_trends(financials)
        
        # 4. Peer comparison
        peers = await self.fetch_peers(ticker)
        comparison = self.compare_peers(ratios, peers)
        
        # 4. Valuation
        valuation = self.valuation.dcf(financials)
        
        # 5. Risk assessment
        risk = self.assess_risk(financials, ratios)
        
        return FinancialReport(
            financials=financials,
            ratios=ratios,
            trends=trends,
            peer_comparison=comparison,
            valuation=valuation,
            risk=risk
        )
```

---

## Data Science Workflows

### 1. ML Pipeline Agent
```python
class MLPipelineAgent:
    def __init__(self):
        self.automl = AutoML()
        self.feature_engine = FeatureEngine()
        self.model_registry = ModelRegistry()
    
    async def build_pipeline(self, spec: MLPipelineSpec) -> Pipeline:
        # 1. Feature engineering
        features = await self.feature_engine.engineer(
            spec.data, spec.target, spec.task_type
        )
        
        # 2. Model selection
        if spec.auto_select:
            model = await self.automl.select(features, spec.target, spec.task_type)
        else:
            model = spec.model
        
        # 3. Hyperparameter tuning
        best_params = await self.tune_hyperparameters(model, features, spec.target)
        
        # 4. Train final model
        pipeline = Pipeline([
            ("preprocessor", spec.preprocessor),
            ("model", model.set_params(**best_params))
        ])
        pipeline.fit(features, spec.target)
        
        # 5. Evaluate
        metrics = self.evaluate(pipeline, spec.test_data)
        
        # 6. Register
        version = await self.model_registry.register(pipeline, metrics)
        
        return Pipeline(
            pipeline=pipeline,
            metrics=metrics,
            version=version,
            feature_importance=self.get_importance(pipeline)
        )
```

### 2. A/B Testing Agent
```python
class ABTestingAgent:
    def __init__(self):
        self.experiment_runner = ExperimentRunner()
        self.sequential = SequentialTesting()
        self.bayesian = BayesianABTest()
    
    async def design_experiment(self, spec: ABTestSpec) -> ExperimentDesign:
        # Power analysis
        sample_size = self.power_analysis(
            baseline=spec.baseline,
            mde=spec.mde,
            alpha=spec.alpha,
            power=spec.power
        )
        
        # Randomization
        randomization = self.design_randomization(spec)
        
        # Guardrails
        guardrails = self.design_guardrails(spec)
        
        return ExperimentDesign(
            sample_size=sample_size,
            randomization=randomization,
            guardrails=guardrails,
            duration=self.estimate_duration(sample_size, spec.traffic)
        )
    
    async def analyze_results(self, experiment: Experiment) -> ABTestResult:
        # Frequentist
        freq = self.frequentist_analysis(experiment)
        
        # Bayesian
        bayes = await self.bayesian.analyze(experiment)
        
        # Sequential
        seq = self.sequential.analyze(experiment)
        
        # Subgroup analysis
        subgroups = self.subgroup_analysis(experiment)
        
        # Guardrail check
        guardrail_check = self.check_guardrails(experiment)
        
        return ABTestResult(
            frequentist=freq,
            bayesian=bayes,
            sequential=seq,
            subgroups=subgroups,
            guardrails=guardrail_check,
            recommendation=self.recommend(freq, bayes, seq)
        )
```

---

## Visualization & Reporting

### 1. Automated Visualization
```python
class VisualizationAgent:
    def __init__(self):
        self.chart_recommender = ChartRecommender()
        self.color_palette = ColorPalette()
    
    async def create_visualization(self, data: Dataset, goal: VisualizationGoal) -> Chart:
        # 1. Recommend chart type
        chart_type = self.chart_recommender.recommend(data, goal)
        
        # 2. Generate config
        config = self.generate_config(data, chart_type, goal)
        
        # 3. Apply styling
        styled = self.apply_styling(config, goal.audience)
        
        # 4. Add interactivity
        if goal.interactive:
            config = self.add_interactivity(config)
        
        return Chart(type=chart_type, config=config, data=data)
    
    def recommend_chart(self, data: Dataset, goal: VisualizationGoal) -> ChartType:
        n_vars = len(data.numeric_columns)
        n_cats = len(data.categorical_columns)
        n_rows = len(data)
        
        if goal.type == "trend" and n_vars >= 1:
            return ChartType.LINE
        elif goal.type == "comparison" and n_cats >= 1:
            return ChartType.BAR if n_rows < 50 else ChartType.HORIZONTAL_BAR
        elif goal.type == "distribution" and n_vars >= 1:
            return ChartType.HISTOGRAM if n_rows < 10000 else ChartType.DENSITY
        elif goal.type == "relationship" and n_vars >= 2:
            return ChartType.SCATTER
        elif goal.type == "composition":
            return ChartType.PIE if n_cats < 10 else ChartType.TREEMAP
        elif goal.type == "correlation":
            return ChartType.HEATMAP
        return ChartType.AUTO
```

### 2. Report Generation Agent
```python
class ReportAgent:
    def __init__(self):
        self.template_engine = JinjaEnvironment()
        self.exporters = {
            "html": HTMLExporter(),
            "pdf": PDFExporter(),
            "notebook": NotebookExporter(),
            "slides": SlidesExporter()
        }
    
    async def generate_report(self, 
                              analysis: AnalysisReport, 
                              template: ReportTemplate,
                              format: str = "html") -> Report:
        # 1. Prepare context
        context = self.prepare_context(analysis)
        
        # 2. Render template
        rendered = await self.template_engine.render(template, context)
        
        # 3. Export
        exported = await self.exporters[format].export(rendered)
        
        return Report(
            content=exported,
            format=format,
            metadata=ReportMetadata(
                generated_at=datetime.utcnow(),
                author="DataAnalysisAgent",
                version="1.0"
            )
        )
```

---

## Data Quality & Governance

### Data Quality Agent
```python
class DataQualityAgent:
    def __init__(self):
        self.expectations = GreatExpectations()
        self.pandera = PanderaValidator()
        self.profiler = DataProfiler()
    
    async def assess(self, dataset: Dataset, suite: ExpectationSuite = None) -> QualityReport:
        # 1. Profile
        profile = await self.profiler.profile(dataset)
        
        # 2. Validate expectations
        if suite:
            validation = await self.expectations.validate(dataset, suite)
        else:
            # Auto-generate expectations
            suite = await self.auto_generate_expectations(dataset)
            validation = await self.expectations.validate(dataset, suite)
        
        # 3. Schema validation
        schema_result = await self.pandera.validate(dataset)
        
        # 4. Compute scores
        scores = self.compute_scores(profile, validation, schema_result)
        
        return QualityReport(
            profile=profile,
            validation=validation,
            schema=schema_result,
            scores=scores,
            issues=self.identify_issues(validation, schema_result),
            recommendations=self.generate_recommendations(scores)
        )
```

---

## Evaluation Benchmarks

| Benchmark | Task | Metric |
|---|---|---|
| **Text-to-SQL** | Spider, BIRD | Execution Accuracy, Execution Match |
| **Text-to-Visualization** | nvBench, VisText | Chart Type Accuracy, Spec Match |
| **Data Analysis** | DASBench, MLAgentBench | Insight Quality, Code Correctness |
| **SQL Generation** | Spider, KaggleDBQA | Execution Accuracy |
| **Forecasting** | M4, M5, Tourism | MAE, RMSE, MAPE, sMAPE |
| **Causal Inference** | CausalBench, TWIN | ATE Error, Coverage |
| **AutoML** | AMLB, OpenML | Regret, Training Time |

---

## Best Practices

### Data Handling
1. **Never modify raw data** - work on copies
2. **Version data** - use DVC, LakeFS, or Delta Lake
3. **Document transformations** - lineage tracking
4. **Validate at each step** - schema, stats, constraints

### Analysis Rigor
1. **Pre-register analysis** - avoid p-hacking
2. **Check assumptions** - normality, independence, homoscedasticity
3. **Report effect sizes** - not just p-values
4. **Report confidence intervals** - not just point estimates
5. **Pre-specify subgroup analyses** - avoid fishing

### Reproducibility
1. **Pin dependencies** - lockfiles, containers
2. **Seed randomness** - numpy, torch, random
3. **Track experiments** - MLflow, Weights & Biases, ClearML
4. **Notebook hygiene** - executable, ordered, documented

### Communication
1. **Know your audience** - executive vs technical
2. **Lead with insights** - not methodology
3. **Visualize uncertainty** - intervals, not points
4. **Actionable recommendations** - so what? now what?

---

## Related Notes

- [[Tools/Code-Execution.md]] — Sandbox execution
- [[ToolUse/API-Integration.md]] — Database APIs
- [[Tools/Static-Analysis.md]] — Code quality for analysis
- [[Tools/Build-Tools.md]] — Reproducible environments
- [[ModelArchitectures/Reasoning-Models.md]] — Reasoning for analysis
- [[ModelArchitectures/Small-Language-Models.md]] — SLMs for analysis
- [[ContextEngineering/RAG-Fundamentals.md]] — RAG for data docs
- [[ContextEngineering/Knowledge-Graphs.md]] — KG for data lineage
- [[ModelArchitectures/Reasoning-Models.md]] — Causal reasoning
- [[ModelArchitectures/Mixture-of-Experts.md]] — MoE for analysis
- [[Domains/Coding-Agents.md]] — Code for analysis
- [[Domains/Research-Agents.md]] — Research methods
- [[Domains/Scientific-Agents.md]] — Scientific discovery
- [[Evaluation/Agent-Benchmarks.md]] — Benchmarks
- [[Testing/Test-Pyramid.md]] — Testing analysis code
- [[DevOps/Observability.md]] — Data observability
- [[DevOps/Infrastructure-as-Code.md]] — Data infrastructure
- [[Production/Observability.md]] — Data observability
- [[Production/Cost-Optimization.md]] — Compute optimization
- [[Frontiers/Scientific-Agents.md]] — Scientific discovery
- [[Frontiers/Self-Improving-Agents.md]] — Self-improving analysis
- [[Frontiers/Research-Directions.md]] — Research directions

---

*Sources: pandas, polars, DuckDB, Great Expectations, pandera, ydata-profiling, sweetviz, statsmodels, scipy, pingouin, causalml, econml, dowhy, prophet, neuralforecast, statsforecast, mlflow, wandb, clearml, dvc, lakefs, delta lake, great expectations, pandera, great_expectations, evidence-based data analysis, causal inference (Pearl, Rubin, Imbens), A/B testing (Kohavi, Kohavi), experimentation platforms (Netflix, Airbnb, Uber, Meta), visualization best practices (Few, Cairo, Tufte), grammar of graphics (Wilkinson), color theory (Ware, Brewer).*