Data Analysis Agents

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

Data Analysis Agent Capabilities

Data Analysis Agent Architectures

1. Exploratory Data Analysis (EDA) Agent

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

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

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

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

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

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

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

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

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

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

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

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

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

• Code Execution — Sandbox execution
• API Integration — Database APIs
• Static Analysis — Code quality for analysis
• Build Tools — Reproducible environments
• Reasoning Models — Reasoning for analysis
• Small Language Models — SLMs for analysis
• RAG Fundamentals — RAG for data docs
• Knowledge Graphs — KG for data lineage
• Reasoning Models — Causal reasoning
• Mixture of Experts — MoE for analysis
• Coding Agents — Code for analysis
• Research Agents — Research methods
• Scientific Agents — Scientific discovery
• Agent Benchmarks — Benchmarks
• Test Pyramid — Testing analysis code
• Observability — Data observability
• Infrastructure as Code — Data infrastructure
• Observability — Data observability
• Cost Optimization — Compute optimization
• Scientific Agents — Scientific discovery
• Self Improving Agents — Self-improving analysis
• Research Directions — 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).