+
+
+
+
Upload a dataset and DataPilot immediately inspects its structure, surfaces quality risks, + recommends analytical targets, creates interactive evidence, and prepares a leakage-safe + machine-learning study—with Gemini available for grounded interpretation.
+ +Load a complete classification or regression dataset and see the full analyst workflow.
iris · Run run_f13555d6077f
| Metric | Value |
|---|---|
| accuracy | 0.9333 |
| balanced_accuracy | 0.9333 |
| f1_weighted | 0.9333 |
| Severity | Code | Observation |
|---|---|---|
| warning | DUPLICATE_ROWS | 1 exact duplicate rows can bias validation. |
Generated from computed evidence. Predictive findings do not establish causality.
+ \ No newline at end of file diff --git a/artifacts/run_f13555d6077f/metrics.json b/artifacts/run_f13555d6077f/metrics.json new file mode 100644 index 0000000000000000000000000000000000000000..94f0396d8b89d7eb6efba1cddd23e49454a0d5a6 --- /dev/null +++ b/artifacts/run_f13555d6077f/metrics.json @@ -0,0 +1,228 @@ +{ + "run_id": "run_f13555d6077f", + "status": "completed", + "dataset_name": "iris", + "profile": { + "rows": 150, + "columns": 5, + "numeric_columns": [ + "sepal_length_(cm)", + "sepal_width_(cm)", + "petal_length_(cm)", + "petal_width_(cm)", + "target" + ], + "categorical_columns": [], + "datetime_columns": [], + "duplicate_rows": 1, + "missing_cells": 0, + "missing_rate": 0.0, + "memory_mb": 0.006, + "target": "target", + "task_type": "classification", + "target_cardinality": 3 + }, + "plan": { + "objective": "Predict 'target' and produce reproducible, evidence-backed insights.", + "target": "target", + "task_type": "classification", + "primary_metric": "balanced_accuracy", + "validation_strategy": "Training-only stratified cross-validation; untouched final test evaluation", + "candidate_models": [ + "Logistic Regression", + "Random Forest", + "Extra Trees", + "Histogram Gradient Boosting", + "XGBoost (when installed)" + ], + "risk_controls": [ + "Drop rows with missing target before split", + "Fit imputers, encoders, and scalers on training folds only", + "Flag leakage-like names and identifier cardinality", + "Require critic quality gate before explanation" + ] + }, + "quality_issues": [ + { + "code": "DUPLICATE_ROWS", + "severity": "warning", + "column": null, + "message": "1 exact duplicate rows can bias validation.", + "evidence_ids": [ + "EV-9000641D" + ] + } + ], + "evidence": [ + { + "evidence_id": "EV-84BCE8AD", + "claim": "Dataset missingness was measured across all cells.", + "metric": "missing_rate", + "value": 0.0, + "source": "uploaded_dataset", + "method": "pandas.isna" + }, + { + "evidence_id": "EV-9000641D", + "claim": "Exact duplicate rows were counted before splitting.", + "metric": "duplicate_rows", + "value": 1, + "source": "uploaded_dataset", + "method": "pandas.duplicated" + }, + { + "evidence_id": "EV-A9BE42C5", + "claim": "Rows with missing labels cannot be used for supervised training.", + "metric": "missing_target_rows", + "value": 0, + "source": "column:target", + "method": "pandas.isna" + }, + { + "evidence_id": "EV-76E9ED71", + "claim": "Class imbalance was measured using the minority-class share.", + "metric": "minority_class_share", + "value": 0.3333, + "source": "column:target", + "method": "normalized value counts" + } + ], + "model_results": [ + { + "name": "Logistic Regression", + "primary_metric": "balanced_accuracy", + "primary_score": 0.9583333333333334, + "metrics": { + "accuracy": 0.9333, + "balanced_accuracy": 0.9333, + "f1_weighted": 0.9333 + }, + "cross_validation_mean": 0.9583333333333334, + "cross_validation_std": 0.026352313834736508, + "training_seconds": 0.228, + "selection_score": 0.9583333333333334, + "final_test_score": 0.9333, + "final_test_metrics": { + "accuracy": 0.9333, + "balanced_accuracy": 0.9333, + "f1_weighted": 0.9333 + } + }, + { + "name": "Extra Trees", + "primary_metric": "balanced_accuracy", + "primary_score": 0.9583333333333334, + "metrics": {}, + "cross_validation_mean": 0.9583333333333334, + "cross_validation_std": 0.026352313834736508, + "training_seconds": 2.901, + "selection_score": 0.9583333333333334, + "final_test_score": null, + "final_test_metrics": {} + }, + { + "name": "Random Forest", + "primary_metric": "balanced_accuracy", + "primary_score": 0.95, + "metrics": {}, + "cross_validation_mean": 0.95, + "cross_validation_std": 0.0311804782231162, + "training_seconds": 2.641, + "selection_score": 0.95, + "final_test_score": null, + "final_test_metrics": {} + } + ], + "model_failures": [], + "best_model": "Logistic Regression", + "critic": { + "approved": true, + "score": 0.9583333333333334, + "threshold": 0.55, + "reasons": [ + "Performance and validation consistency passed the configured quality gate." + ], + "retry_number": 0 + }, + "explainability": { + "method": "Permutation importance", + "feature_importance": { + "petal_width_(cm)": 0.213333, + "petal_length_(cm)": 0.2, + "sepal_length_(cm)": 0.06, + "sepal_width_(cm)": 0.046667 + }, + "caveats": [ + "Permutation importance can dilute importance among correlated features.", + "Feature importance is predictive, not causal." + ] + }, + "executive_summary": [ + "The analysis used 150 rows and 5 columns for a classification task targeting 'target'.", + "Logistic Regression ranked first with training CV balanced_accuracy 0.958; its one-time test score was 0.933.", + "1 data-quality observations were recorded; 0 are critical.", + "The strongest predictive signals were petal_width_(cm), petal_length_(cm), sepal_length_(cm) according to permutation importance." + ], + "recommendations": [ + "Validate performance on fresh, out-of-time data before production deployment.", + "Review suspected leakage and identifier columns with a domain owner.", + "Monitor input drift and the primary metric after deployment." + ], + "artifacts": {}, + "trace": [ + { + "agent": "Data Quality Agent", + "status": "completed", + "duration_seconds": 0.005, + "detail": "Recorded 1 quality observations." + }, + { + "agent": "EDA Agent", + "status": "completed", + "duration_seconds": 0.058, + "detail": "Computed DuckDB-backed dataset overview." + }, + { + "agent": "Statistical Analysis Agent", + "status": "completed", + "duration_seconds": 0.001, + "detail": "Measured target distribution and associations." + }, + { + "agent": "Planning Agent", + "status": "completed", + "duration_seconds": 0.0, + "detail": "Selected balanced_accuracy as primary metric." + }, + { + "agent": "Feature Engineering Agent", + "status": "completed", + "duration_seconds": 0.0, + "detail": "Created leakage-safe ColumnTransformer plan." + }, + { + "agent": "Modeling Agent", + "status": "completed", + "duration_seconds": 5.916, + "detail": "Compared 3 models; Logistic Regression ranked first." + }, + { + "agent": "Evaluation / Critic Agent", + "status": "completed", + "duration_seconds": 0.0, + "detail": "Approved analysis." + }, + { + "agent": "Explainability Agent", + "status": "completed", + "duration_seconds": 0.331, + "detail": "Generated Permutation importance explanations." + }, + { + "agent": "Executive Insights Agent", + "status": "completed", + "duration_seconds": 0.0, + "detail": "Created evidence-grounded narrative with deterministic metric provenance." + } + ] +} \ No newline at end of file diff --git a/artifacts/run_f13555d6077f/reproduction.json b/artifacts/run_f13555d6077f/reproduction.json new file mode 100644 index 0000000000000000000000000000000000000000..22d8fbbaaca40b4c392ce7cfbda9ee6152ad889c --- /dev/null +++ b/artifacts/run_f13555d6077f/reproduction.json @@ -0,0 +1,8 @@ +{ + "run_id": "run_f13555d6077f", + "random_state": 42, + "test_size": 0.2, + "target": "target", + "task_type": "classification", + "best_model": "Logistic Regression" +} \ No newline at end of file diff --git a/datapilot/__init__.py b/datapilot/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..3bfce83e0105b6db285851def6182be97fa4fc38 --- /dev/null +++ b/datapilot/__init__.py @@ -0,0 +1,3 @@ +"""DataPilot AI: evidence-grounded autonomous data science.""" + +__version__ = "1.0.0" diff --git a/datapilot/analyst.py b/datapilot/analyst.py new file mode 100644 index 0000000000000000000000000000000000000000..37d57708211974c28037bcf0993bedc05fc030b2 --- /dev/null +++ b/datapilot/analyst.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import hashlib +import io +import json +import re +from dataclasses import dataclass +from typing import Any + +import numpy as np +import pandas as pd + +TARGET_WORDS = { + "target": 1.0, + "label": 1.0, + "outcome": 0.95, + "class": 0.9, + "churn": 0.95, + "fraud": 0.95, + "default": 0.9, + "price": 0.8, + "revenue": 0.75, + "sales": 0.75, + "diagnosis": 0.9, + "status": 0.7, + "response": 0.8, + "converted": 0.9, +} +PII_PATTERN = re.compile(r"(email|phone|mobile|address|ssn|passport|account|name)", re.I) + + +@dataclass +class DatasetBrief: + fingerprint: str + rows: int + columns: int + numeric: int + categorical: int + datetime: int + missing_cells: int + duplicate_rows: int + memory_mb: float + + +def inspect_dataset(frame: pd.DataFrame) -> dict[str, Any]: + """Compute an immediate analyst profile without requiring a target.""" + numeric = list(frame.select_dtypes(include=np.number).columns) + datetime = list(frame.select_dtypes(include=["datetime", "datetimetz"]).columns) + categorical = [c for c in frame.columns if c not in numeric and c not in datetime] + missing = frame.isna().sum() + brief = DatasetBrief( + fingerprint=hashlib.sha256( + pd.util.hash_pandas_object(frame, index=True).values.tobytes() + ).hexdigest()[:12], + rows=len(frame), + columns=len(frame.columns), + numeric=len(numeric), + categorical=len(categorical), + datetime=len(datetime), + missing_cells=int(missing.sum()), + duplicate_rows=int(frame.duplicated().sum()), + memory_mb=round(float(frame.memory_usage(deep=True).sum() / 1_048_576), 2), + ) + dictionary = build_data_dictionary(frame) + quality_score = max( + 0, + round( + 100 + - (brief.missing_cells / max(1, frame.size)) * 45 + - (brief.duplicate_rows / max(1, brief.rows)) * 25 + - sum(dictionary["issue_count"].clip(upper=3)) / max(1, len(dictionary)) * 4 + ), + ) + return { + "brief": brief, + "dictionary": dictionary, + "targets": rank_target_candidates(frame), + "quality_score": quality_score, + "missing": missing.sort_values(ascending=False), + "numeric": numeric, + "categorical": categorical, + "datetime": datetime, + "correlation": frame[numeric].corr(numeric_only=True) + if len(numeric) > 1 + else pd.DataFrame(), + } + + +def build_data_dictionary(frame: pd.DataFrame) -> pd.DataFrame: + """Create an evidence-based data dictionary for every column.""" + rows: list[dict[str, Any]] = [] + for column in frame.columns: + series = frame[column] + unique, missing = int(series.nunique(dropna=True)), int(series.isna().sum()) + issues: list[str] = [] + if missing: + issues.append("Missing values") + if unique <= 1: + issues.append("Constant") + if len(frame) and unique / len(frame) > 0.98: + issues.append("Identifier-like") + if PII_PATTERN.search(str(column)): + issues.append("Potential PII") + role = "Numeric feature" if pd.api.types.is_numeric_dtype(series) else "Categorical feature" + if pd.api.types.is_datetime64_any_dtype(series): + role = "Datetime" + elif unique == len(frame) and len(frame) > 10: + role = "Identifier" + rows.append( + { + "column": str(column), + "type": str(series.dtype), + "role": role, + "unique": unique, + "missing": missing, + "missing_%": round(missing / max(1, len(frame)) * 100, 2), + "example_values": ", ".join(map(str, series.dropna().astype(str).unique()[:3]))[ + :90 + ], + "issues": ", ".join(issues) or "None detected", + "issue_count": len(issues), + } + ) + return pd.DataFrame(rows) + + +def rank_target_candidates(frame: pd.DataFrame) -> list[dict[str, Any]]: + """Rank targets while explicitly leaving confirmation to the user.""" + ranked: list[dict[str, Any]] = [] + for position, column in enumerate(frame.columns): + series, name = frame[column], str(column).lower().strip() + cardinality = int(series.nunique(dropna=True)) + score = max((v for word, v in TARGET_WORDS.items() if word in name), default=0.0) + if 2 <= cardinality <= max(20, int(len(frame) * 0.1)): + score += 0.22 + if position == len(frame.columns) - 1: + score += 0.12 + if cardinality >= max(10, int(len(frame) * 0.95)): + score -= 0.45 + if series.isna().mean() > 0.5 or cardinality < 2: + score -= 0.5 + task = "classification" if cardinality <= max(20, int(len(frame) * 0.05)) else "regression" + if pd.api.types.is_numeric_dtype(series) and cardinality > 20: + task = "regression" + ranked.append( + { + "column": str(column), + "task": task, + "confidence": round(max(0, min(score, 0.99)), 2), + "reason": f"{cardinality:,} distinct values; " + + ("name/position signals detected" if score > 0.3 else "weak heuristic evidence"), + } + ) + return sorted(ranked, key=lambda item: item["confidence"], reverse=True)[:5] + + +def ai_context(frame: pd.DataFrame, profile: dict[str, Any], excluded: list[str]) -> dict[str, Any]: + """Build a bounded, redacted payload suitable for AI interpretation.""" + safe = frame.drop(columns=[c for c in excluded if c in frame], errors="ignore").copy() + pii = [c for c in safe.columns if PII_PATTERN.search(str(c))] + safe = safe.drop(columns=pii, errors="ignore") + return { + "shape": list(frame.shape), + "columns": profile["dictionary"].drop(columns=["issue_count"]).to_dict(orient="records"), + "numeric_summary": safe.select_dtypes(include=np.number).describe().round(3).to_dict(), + "sample": safe.head(3).replace({np.nan: None}).to_dict(orient="records"), + "excluded_columns": sorted(set(excluded + pii)), + "quality_score": profile["quality_score"], + "target_candidates": profile["targets"], + } + + +def gemini_dataset_summary( + frame: pd.DataFrame, profile: dict[str, Any], api_key: str, model: str, excluded: list[str] +) -> str: + """Ask Gemini for an evidence-bounded analyst narrative.""" + if not api_key: + raise ValueError("Enter a Gemini API key to generate AI interpretation.") + try: + from google import genai + except ImportError as exc: + raise ValueError("Install the AI extra: pip install google-genai") from exc + prompt = """You are DataPilot, a rigorous senior data analyst. Use ONLY the supplied JSON. +Return concise Markdown with exactly these headings: Finding, Evidence, Interpretation, +Limitation, Recommendation. Explain likely row grain and useful business questions, but label +uncertain semantics as assumptions. Never invent values, origin, or causal claims. +DATA:\n""" + json.dumps(ai_context(frame, profile, excluded), default=str) + try: + response = genai.Client(api_key=api_key).models.generate_content( + model=model, contents=prompt + ) + return response.text + except Exception as exc: + raise ValueError(f"Gemini request failed: {str(exc)[:240]}") from exc + + +def dataframe_csv(frame: pd.DataFrame) -> bytes: + buffer = io.StringIO() + frame.to_csv(buffer, index=False) + return buffer.getvalue().encode("utf-8") diff --git a/datapilot/config.py b/datapilot/config.py new file mode 100644 index 0000000000000000000000000000000000000000..552fe0ef1594a55aa42d76c228be90ed31521b52 --- /dev/null +++ b/datapilot/config.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path + +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore") + + app_name: str = "DataPilot AI" + environment: str = "development" + artifact_root: Path = Path("artifacts") + database_url: str = "sqlite:///artifacts/datapilot.db" + max_upload_mb: int = Field(default=25, ge=1, le=250) + max_rows: int = Field(default=100_000, ge=100) + max_columns: int = Field(default=250, ge=2) + max_categories_per_feature: int = Field(default=100, ge=10, le=10_000) + max_encoded_features: int = Field(default=5_000, ge=100, le=100_000) + api_key: str | None = None + requests_per_minute: int = Field(default=30, ge=1, le=10_000) + random_state: int = 42 + test_size: float = Field(default=0.2, gt=0.05, lt=0.5) + max_critic_retries: int = Field(default=1, ge=0, le=3) + min_classification_score: float = 0.55 + min_regression_score: float = 0.15 + optuna_trials: int = Field(default=8, ge=0, le=50) + enable_mlflow: bool = False + mlflow_tracking_uri: str = "file:./artifacts/mlruns" + gemini_api_key: str | None = None + gemini_model: str = "gemini-2.5-flash" + cors_origins: str = "http://localhost:8501" + + def ensure_directories(self) -> None: + self.artifact_root.mkdir(parents=True, exist_ok=True) + + +@lru_cache +def get_settings() -> Settings: + settings = Settings() + settings.ensure_directories() + return settings diff --git a/datapilot/data.py b/datapilot/data.py new file mode 100644 index 0000000000000000000000000000000000000000..edf8858e36fe9a38f2232f9566a3d421a4ad7bb4 --- /dev/null +++ b/datapilot/data.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import io +from pathlib import Path +from typing import BinaryIO + +import duckdb +import pandas as pd +from sklearn.datasets import load_breast_cancer, load_diabetes, load_iris + +from datapilot.config import Settings + +SAMPLE_DATASETS = { + "Iris classification": "iris", + "Breast cancer classification": "breast_cancer", + "Diabetes progression regression": "diabetes", +} + + +def load_sample(name: str) -> tuple[pd.DataFrame, str, str]: + loaders = { + "iris": load_iris, + "breast_cancer": load_breast_cancer, + "diabetes": load_diabetes, + } + if name not in loaders: + raise ValueError(f"Unknown sample dataset: {name}") + bundle = loaders[name](as_frame=True) + frame = bundle.frame.copy() + frame.columns = [str(column).replace(" ", "_") for column in frame.columns] + target = str(bundle.target.name).replace(" ", "_") + return frame, target, name + + +def read_dataset( + source: bytes | BinaryIO | str | Path, + filename: str, + settings: Settings, +) -> pd.DataFrame: + suffix = Path(filename).suffix.lower() + if suffix not in {".csv", ".parquet"}: + raise ValueError("Only CSV and Parquet files are supported.") + if isinstance(source, bytes): + if len(source) > settings.max_upload_mb * 1024 * 1024: + raise ValueError(f"File exceeds the {settings.max_upload_mb} MB upload limit.") + stream: BinaryIO | str | Path = io.BytesIO(source) + else: + stream = source + frame = pd.read_csv(stream) if suffix == ".csv" else pd.read_parquet(stream) + validate_shape(frame, settings) + frame.columns = _unique_columns([str(column).strip() for column in frame.columns]) + return frame + + +def validate_shape(frame: pd.DataFrame, settings: Settings) -> None: + if frame.empty: + raise ValueError("The dataset is empty.") + if len(frame) > settings.max_rows: + raise ValueError(f"Dataset has {len(frame):,} rows; limit is {settings.max_rows:,}.") + if len(frame.columns) > settings.max_columns: + raise ValueError( + f"Dataset has {len(frame.columns):,} columns; limit is {settings.max_columns:,}." + ) + if len(frame.columns) < 2: + raise ValueError("At least one feature and one target column are required.") + + +def _unique_columns(columns: list[str]) -> list[str]: + seen: dict[str, int] = {} + result: list[str] = [] + for raw in columns: + name = raw or "unnamed" + count = seen.get(name, 0) + seen[name] = count + 1 + result.append(name if count == 0 else f"{name}_{count}") + return result + + +def duckdb_overview(frame: pd.DataFrame) -> dict[str, object]: + connection = duckdb.connect(database=":memory:") + try: + connection.register("dataset", frame) + row = connection.execute( + """ + SELECT + COUNT(*) AS row_count + FROM dataset + """ + ).fetchone() + numeric = frame.select_dtypes(include="number") + correlations = ( + numeric.corr(numeric_only=True).round(4).fillna(0).to_dict() + if len(numeric.columns) > 1 + else {} + ) + return { + "row_count": int(row[0]), + "duplicate_rows": int(frame.duplicated().sum()), + "correlations": correlations, + } + finally: + connection.close() diff --git a/datapilot/evaluation.py b/datapilot/evaluation.py new file mode 100644 index 0000000000000000000000000000000000000000..5aadcad504a018dc2344bc7a8ea535339e801004 --- /dev/null +++ b/datapilot/evaluation.py @@ -0,0 +1,58 @@ +"""Deterministic quality gates for evidence-grounded LLM narratives.""" + +from __future__ import annotations + +import re +from collections.abc import Iterable +from dataclasses import asdict, dataclass + +EVIDENCE_ID = re.compile(r"\bEVD-[A-Z0-9-]+\b") +NUMBER = re.compile(r"(? dict[str, object]: + return asdict(self) + + +def evaluate_narrative( + output: str, + allowed_evidence_ids: Iterable[str], + allowed_numbers: Iterable[str | float | int], +) -> NarrativeEvaluation: + """Score an AI narrative for evidence citations, numeric faithfulness, and safety.""" + allowed_ids = set(allowed_evidence_ids) + cited = set(EVIDENCE_ID.findall(output)) + supported_ids = cited.issubset(allowed_ids) and bool(cited) + allowed_numeric = {_normalize_number(str(value)) for value in allowed_numbers} + narrative_without_ids = EVIDENCE_ID.sub("", output) + found = {_normalize_number(value) for value in NUMBER.findall(narrative_without_ids)} + unsupported = sorted(value for value in found if value not in allowed_numeric) + leaked_pii = bool(PII.search(output)) + injection = bool(INJECTION.search(output)) + penalties = (0 if supported_ids else 0.35) + min(0.35, len(unsupported) * 0.1) + penalties += 0.2 if leaked_pii else 0 + penalties += 0.1 if injection else 0 + return NarrativeEvaluation( + supported_evidence_ids=supported_ids, + unsupported_numbers=unsupported, + leaked_pii=leaked_pii, + prompt_injection_echo=injection, + score=round(max(0.0, 1.0 - penalties), 3), + ) + + +def _normalize_number(value: str) -> str: + return value.replace(",", "").rstrip("%").lstrip("+") diff --git a/datapilot/insights.py b/datapilot/insights.py new file mode 100644 index 0000000000000000000000000000000000000000..0fcbd8fba609b703909d1098c6d52257f8f2f5e3 --- /dev/null +++ b/datapilot/insights.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +from typing import Any + +from datapilot.config import Settings +from datapilot.schemas import Evidence + + +def deterministic_insights(state: dict[str, Any]) -> tuple[list[str], list[str]]: + profile = state["profile"] + best = state["model_bundle"].results[0] + quality = state["quality_issues"] + explainability = state["explainability"] + important = list(explainability.feature_importance)[:3] + summary = [ + ( + f"The analysis used {profile.rows:,} rows and {profile.columns:,} columns for a " + f"{profile.task_type.value} task targeting '{profile.target}'." + ), + ( + f"{best.name} ranked first with training CV {best.primary_metric} {best.primary_score:.3f}; " + f"its one-time test score was {best.final_test_score:.3f}." + ), + ( + f"{len(quality)} data-quality observations were recorded; " + f"{sum(issue.severity.value == 'critical' for issue in quality)} are critical." + ), + ] + if important: + summary.append( + f"The strongest predictive signals were {', '.join(important)} " + f"according to {explainability.method.lower()}." + ) + recommendations = [ + "Validate performance on fresh, out-of-time data before production deployment.", + "Review suspected leakage and identifier columns with a domain owner.", + "Monitor input drift and the primary metric after deployment.", + ] + if profile.missing_rate > 0.1: + recommendations.insert(0, "Investigate upstream causes of missing data before retraining.") + return summary, recommendations + + +def optional_llm_narrative( + state: dict[str, Any], evidence: list[Evidence], settings: Settings +) -> list[str] | None: + """Generate narrative only from bounded evidence; calculations remain deterministic.""" + if not settings.gemini_api_key: + return None + try: + from google import genai + + client = genai.Client(api_key=settings.gemini_api_key) + payload = { + "profile": state["profile"].model_dump(), + "best_model": state["model_bundle"].results[0].model_dump(), + "critic": state["critic"].model_dump(), + "evidence": [item.model_dump() for item in evidence[:25]], + } + prompt = ( + "You are a senior data scientist. Return exactly four concise markdown bullet points. " + "Use only the JSON evidence below. Cite supporting evidence IDs in square brackets. " + "Do not add numbers, causal claims, or facts absent from the payload.\n" + + json.dumps(payload, default=str) + ) + response = client.models.generate_content(model=settings.gemini_model, contents=prompt) + lines = [line.strip("- ").strip() for line in response.text.splitlines() if line.strip()] + return lines[:4] or None + except Exception: + return None + + +def answer_follow_up(run: dict[str, Any], question: str) -> str: + lowered = question.lower() + if any(token in lowered for token in {"best model", "which model", "winner"}): + top = run["model_results"][0] + return ( + f"The best model was **{top['name']}**, with {top['primary_metric']} " + f"training-CV **{top['selection_score']:.3f}** and one-time test " + f"**{top['final_test_score']:.3f}**." + ) + if any(token in lowered for token in {"feature", "important", "driver"}): + importance = run["explainability"]["feature_importance"] + top = list(importance.items())[:5] + return ( + "Top predictive features: " + + ", ".join(f"**{name}** ({value:.4f})" for name, value in top) + + ". These are associations, not causal effects." + ) + if any(token in lowered for token in {"quality", "missing", "leak", "risk"}): + issues = run["quality_issues"] + if not issues: + return "No material quality flags were detected by the configured checks." + return "Quality observations: " + "; ".join(item["message"] for item in issues[:6]) + if any(token in lowered for token in {"metric", "performance", "score"}): + top = run["model_results"][0] + formatted = ", ".join( + f"{key}={value:.3f}" for key, value in top["final_test_metrics"].items() + ) + return f"Selected-model one-time test metrics: {formatted}." + return ( + "I can answer evidence-backed questions about the best model, performance metrics, " + "data quality, leakage risk, and feature importance for this run." + ) diff --git a/datapilot/jobs.py b/datapilot/jobs.py new file mode 100644 index 0000000000000000000000000000000000000000..c19e6e1a0163d055053845a8f146bfcaade5fc3b --- /dev/null +++ b/datapilot/jobs.py @@ -0,0 +1,78 @@ +"""Bounded asynchronous analysis jobs with a production-queue compatible contract.""" + +from __future__ import annotations + +import threading +from collections.abc import Callable +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any +from uuid import uuid4 + + +@dataclass +class Job: + job_id: str + status: str = "queued" + progress: int = 0 + created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) + updated_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) + result: Any | None = None + error: str | None = None + future: Future[Any] | None = field(default=None, repr=False) + + def public(self) -> dict[str, Any]: + return { + "job_id": self.job_id, + "status": self.status, + "progress": self.progress, + "created_at": self.created_at, + "updated_at": self.updated_at, + "result": self.result if self.status == "completed" else None, + "error": self.error, + } + + +class JobManager: + """In-process development queue; replaceable by ARQ/Celery behind the same API.""" + + def __init__(self, workers: int = 2) -> None: + self._executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="datapilot") + self._jobs: dict[str, Job] = {} + self._lock = threading.Lock() + + def submit(self, operation: Callable[[], Any]) -> Job: + job = Job(job_id=f"job_{uuid4().hex[:12]}") + with self._lock: + self._jobs[job.job_id] = job + job.future = self._executor.submit(self._run, job, operation) + return job + + def get(self, job_id: str) -> Job | None: + with self._lock: + return self._jobs.get(job_id) + + def cancel(self, job_id: str) -> bool: + job = self.get(job_id) + if job is None or job.future is None or job.status not in {"queued", "running"}: + return False + cancelled = job.future.cancel() + if cancelled: + job.status, job.progress = "cancelled", 0 + job.updated_at = datetime.now(UTC).isoformat() + return cancelled + + @staticmethod + def _run(job: Job, operation: Callable[[], Any]) -> None: + job.status, job.progress = "running", 10 + job.updated_at = datetime.now(UTC).isoformat() + try: + value = operation() + job.result = value.model_dump(mode="json") if hasattr(value, "model_dump") else value + job.status, job.progress = "completed", 100 + except Exception: + job.status, job.progress = "failed", 100 + job.error = "Analysis failed. Use the correlation ID in server logs for support." + finally: + job.updated_at = datetime.now(UTC).isoformat() diff --git a/datapilot/modeling.py b/datapilot/modeling.py new file mode 100644 index 0000000000000000000000000000000000000000..a701636c2df226f5311314db60c03713c435c61f --- /dev/null +++ b/datapilot/modeling.py @@ -0,0 +1,389 @@ +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Any + +import numpy as np +import pandas as pd +from sklearn.base import BaseEstimator +from sklearn.compose import ColumnTransformer +from sklearn.ensemble import ( + ExtraTreesClassifier, + ExtraTreesRegressor, + RandomForestClassifier, + RandomForestRegressor, +) +from sklearn.impute import SimpleImputer +from sklearn.inspection import permutation_importance +from sklearn.linear_model import LinearRegression, LogisticRegression +from sklearn.metrics import ( + accuracy_score, + balanced_accuracy_score, + f1_score, + mean_absolute_error, + mean_squared_error, + r2_score, + roc_auc_score, +) +from sklearn.model_selection import cross_val_score, train_test_split +from sklearn.pipeline import Pipeline +from sklearn.preprocessing import OneHotEncoder, StandardScaler + +from datapilot.config import Settings +from datapilot.schemas import ( + CriticDecision, + ExplainabilityResult, + ModelFailure, + ModelResult, + TaskType, +) +from datapilot.tuning import tune_random_forest + + +@dataclass +class TrainingBundle: + pipeline: Pipeline + results: list[ModelResult] + best_model: str + test_features: pd.DataFrame + test_target: pd.Series + retry_number: int + failures: list[ModelFailure] + + +logger = logging.getLogger(__name__) + + +def _preprocessor(features: pd.DataFrame, settings: Settings) -> ColumnTransformer: + numeric = features.select_dtypes(include=np.number).columns.tolist() + categorical = [column for column in features.columns if column not in numeric] + numeric_pipeline = Pipeline( + [ + ("imputer", SimpleImputer(strategy="median")), + ("scaler", StandardScaler()), + ] + ) + categorical_pipeline = Pipeline( + [ + ("imputer", SimpleImputer(strategy="most_frequent")), + ( + "encoder", + OneHotEncoder( + handle_unknown="infrequent_if_exist", + min_frequency=2, + max_categories=settings.max_categories_per_feature, + sparse_output=True, + ), + ), + ] + ) + return ColumnTransformer( + [ + ("numeric", numeric_pipeline, numeric), + ("categorical", categorical_pipeline, categorical), + ], + remainder="drop", + verbose_feature_names_out=False, + ) + + +def _candidate_models(task: TaskType, settings: Settings, retry: int) -> dict[str, BaseEstimator]: + if task == TaskType.classification: + models: dict[str, BaseEstimator] = { + "Logistic Regression": LogisticRegression( + max_iter=1_000, class_weight="balanced", random_state=settings.random_state + ), + "Random Forest": RandomForestClassifier( + n_estimators=180 + retry * 80, + min_samples_leaf=max(1, 2 - retry), + class_weight="balanced", + n_jobs=-1, + random_state=settings.random_state, + ), + "Extra Trees": ExtraTreesClassifier( + n_estimators=180 + retry * 80, + class_weight="balanced", + n_jobs=-1, + random_state=settings.random_state, + ), + } + else: + models = { + "Linear Regression": LinearRegression(), + "Random Forest": RandomForestRegressor( + n_estimators=180 + retry * 80, + min_samples_leaf=max(1, 2 - retry), + n_jobs=-1, + random_state=settings.random_state, + ), + "Extra Trees": ExtraTreesRegressor( + n_estimators=180 + retry * 80, + n_jobs=-1, + random_state=settings.random_state, + ), + } + try: + if task == TaskType.classification: + from xgboost import XGBClassifier + + models["XGBoost"] = XGBClassifier( + n_estimators=160 + retry * 60, + max_depth=4 + retry, + learning_rate=0.07, + eval_metric="logloss", + n_jobs=-1, + random_state=settings.random_state, + ) + else: + from xgboost import XGBRegressor + + models["XGBoost"] = XGBRegressor( + n_estimators=160 + retry * 60, + max_depth=4 + retry, + learning_rate=0.07, + n_jobs=-1, + random_state=settings.random_state, + ) + except ImportError: + pass + return models + + +def train_models( + frame: pd.DataFrame, + target: str, + task: TaskType, + settings: Settings, + retry_number: int = 0, +) -> TrainingBundle: + if target not in frame.columns: + raise ValueError(f"Target column '{target}' does not exist.") + clean = frame.dropna(subset=[target]).drop_duplicates().reset_index(drop=True) + features = clean.drop(columns=[target]).copy() + labels = clean[target].copy() + if features.empty: + raise ValueError("No usable feature columns remain after removing the target.") + if labels.nunique(dropna=True) < 2: + raise ValueError("The target must contain at least two distinct non-null values.") + categorical = features.select_dtypes(exclude=np.number) + estimated_width = len(features.select_dtypes(include=np.number).columns) + sum( + min(int(categorical[column].nunique(dropna=True)), settings.max_categories_per_feature) + for column in categorical.columns + ) + if estimated_width > settings.max_encoded_features: + raise ValueError( + f"Estimated encoded width {estimated_width:,} exceeds the safe limit of " + f"{settings.max_encoded_features:,}. Reduce high-cardinality columns or increase " + "MAX_ENCODED_FEATURES after reviewing memory capacity." + ) + stratify = ( + labels if task == TaskType.classification and labels.value_counts().min() >= 2 else None + ) + x_train, x_test, y_train, y_test = train_test_split( + features, + labels, + test_size=settings.test_size, + random_state=settings.random_state, + stratify=stratify, + ) + results: list[ModelResult] = [] + failures: list[ModelFailure] = [] + cv = min(5, max(2, int(len(x_train) / 20))) + if task == TaskType.classification: + smallest_class = int(y_train.value_counts().min()) + cv = min(cv, smallest_class) if smallest_class >= 2 else 0 + scoring = "balanced_accuracy" + else: + scoring = "r2" + if cv < 2: + raise ValueError("The training partition is too small for reliable cross-validation.") + + candidates = _candidate_models(task, settings, retry_number) + tuned_parameters = tune_random_forest( + task, + _preprocessor(x_train, settings), + x_train, + y_train, + settings, + cv, + ) + if tuned_parameters: + candidates["Random Forest"].set_params(**tuned_parameters) + + for name, estimator in candidates.items(): + pipeline = Pipeline( + [("preprocessor", _preprocessor(x_train, settings)), ("model", estimator)] + ) + started = time.perf_counter() + try: + cv_scores = ( + cross_val_score(pipeline, x_train, y_train, scoring=scoring, cv=cv, n_jobs=1) + if cv >= 2 + else np.array([]) + ) + result = ModelResult( + name=name, + primary_metric="balanced_accuracy" if task == TaskType.classification else "r2", + primary_score=float(cv_scores.mean()), + metrics={}, + cross_validation_mean=float(cv_scores.mean()) if len(cv_scores) else None, + cross_validation_std=float(cv_scores.std()) if len(cv_scores) else None, + training_seconds=round(time.perf_counter() - started, 3), + selection_score=float(cv_scores.mean()) if len(cv_scores) else None, + ) + results.append(result) + except Exception as exc: + logger.warning( + "Candidate %s failed during cross-validation: %s", name, type(exc).__name__ + ) + failures.append( + ModelFailure( + name=name, + stage="cross_validation", + exception_category=type(exc).__name__, + sanitized_error="Candidate failed during cross-validation; inspect structured logs.", + training_seconds=round(time.perf_counter() - started, 3), + expected=isinstance(exc, (ValueError, TypeError)), + ) + ) + if not results: + raise RuntimeError("Every candidate model failed; review the data types and target.") + results.sort(key=lambda item: item.selection_score or float("-inf"), reverse=True) + best = results[0] + final_pipeline = Pipeline( + [ + ("preprocessor", _preprocessor(x_train, settings)), + ("model", candidates[best.name]), + ] + ) + final_pipeline.fit(x_train, y_train) + predictions = final_pipeline.predict(x_test) + final_metrics = _metrics(task, y_test, predictions, final_pipeline, x_test) + final_score = ( + final_metrics["balanced_accuracy"] + if task == TaskType.classification + else final_metrics["r2"] + ) + best.final_test_score = float(final_score) + best.final_test_metrics = final_metrics + best.metrics = final_metrics + return TrainingBundle( + pipeline=final_pipeline, + results=results, + best_model=best.name, + test_features=x_test, + test_target=y_test, + retry_number=retry_number, + failures=failures, + ) + + +def _metrics( + task: TaskType, + truth: pd.Series, + predictions: np.ndarray, + pipeline: Pipeline, + features: pd.DataFrame, +) -> dict[str, float]: + if task == TaskType.classification: + metrics = { + "accuracy": round(float(accuracy_score(truth, predictions)), 4), + "balanced_accuracy": round(float(balanced_accuracy_score(truth, predictions)), 4), + "f1_weighted": round(float(f1_score(truth, predictions, average="weighted")), 4), + } + if truth.nunique() == 2 and hasattr(pipeline, "predict_proba"): + probabilities = pipeline.predict_proba(features)[:, 1] + metrics["roc_auc"] = round(float(roc_auc_score(truth, probabilities)), 4) + return metrics + return { + "r2": round(float(r2_score(truth, predictions)), 4), + "rmse": round(float(mean_squared_error(truth, predictions) ** 0.5), 4), + "mae": round(float(mean_absolute_error(truth, predictions)), 4), + } + + +def critic_decision(bundle: TrainingBundle, task: TaskType, settings: Settings) -> CriticDecision: + best = bundle.results[0] + threshold = ( + settings.min_classification_score + if task == TaskType.classification + else settings.min_regression_score + ) + reasons: list[str] = [] + quality_score = best.cross_validation_mean if best.cross_validation_mean is not None else -1.0 + if quality_score < threshold: + reasons.append( + f"Training CV {best.primary_metric} {quality_score:.3f} is below {threshold:.3f}." + ) + if ( + best.cross_validation_mean is not None + and best.final_test_score is not None + and abs(best.final_test_score - best.cross_validation_mean) > 0.2 + ): + reasons.append("Holdout and cross-validation scores diverge by more than 0.20.") + approved = not reasons or bundle.retry_number >= settings.max_critic_retries + if not reasons: + reasons.append("Performance and validation consistency passed the configured quality gate.") + elif approved: + reasons.append("Retry budget exhausted; result is retained with an explicit limitation.") + return CriticDecision( + approved=approved, + score=quality_score, + threshold=threshold, + reasons=reasons, + retry_number=bundle.retry_number, + ) + + +def explain_model(bundle: TrainingBundle) -> ExplainabilityResult: + sample_size = min(300, len(bundle.test_features)) + features = bundle.test_features.iloc[:sample_size] + target = bundle.test_target.iloc[:sample_size] + try: + transformed = bundle.pipeline.named_steps["preprocessor"].transform(features) + transformed_names = bundle.pipeline.named_steps["preprocessor"].get_feature_names_out() + estimator = bundle.pipeline.named_steps["model"] + import shap + + explainer = shap.Explainer(estimator, transformed) + values = explainer(transformed) + raw = np.asarray(values.values) + if raw.ndim == 3: + raw = np.abs(raw).mean(axis=(0, 2)) + else: + raw = np.abs(raw).mean(axis=0) + importance = _top_importance(transformed_names, raw) + return ExplainabilityResult( + method="SHAP", + feature_importance=importance, + caveats=["SHAP values explain this fitted model, not causal effects."], + ) + except Exception: + permutation = permutation_importance( + bundle.pipeline, + features, + target, + n_repeats=5, + random_state=42, + n_jobs=1, + ) + importance = _top_importance(features.columns, np.abs(permutation.importances_mean)) + return ExplainabilityResult( + method="Permutation importance", + feature_importance=importance, + caveats=[ + "Permutation importance can dilute importance among correlated features.", + "Feature importance is predictive, not causal.", + ], + ) + + +def _top_importance(names: Any, values: np.ndarray, limit: int = 15) -> dict[str, float]: + pairs = sorted( + zip([str(name) for name in names], values.tolist(), strict=False), + key=lambda item: item[1], + reverse=True, + )[:limit] + return {name: round(float(value), 6) for name, value in pairs} diff --git a/datapilot/observability.py b/datapilot/observability.py new file mode 100644 index 0000000000000000000000000000000000000000..52183f335c43f5a573e62ce52cfa58af76871658 --- /dev/null +++ b/datapilot/observability.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from typing import Any + +from datapilot.config import Settings + + +def log_to_mlflow(payload: dict[str, Any], settings: Settings) -> bool: + """Record a run when MLflow is enabled; local demos remain dependency-light.""" + if not settings.enable_mlflow: + return False + try: + import mlflow + + mlflow.set_tracking_uri(settings.mlflow_tracking_uri) + mlflow.set_experiment("datapilot-ai") + best = payload["model_results"][0] + with mlflow.start_run(run_name=payload["run_id"]): + mlflow.log_params( + { + "dataset": payload["dataset_name"], + "task_type": payload["profile"]["task_type"], + "target": payload["profile"]["target"], + "selected_model": payload["best_model"], + } + ) + mlflow.log_metrics({key: float(value) for key, value in best["metrics"].items()}) + return True + except Exception: + return False diff --git a/datapilot/persistence.py b/datapilot/persistence.py new file mode 100644 index 0000000000000000000000000000000000000000..a029507002c68e491d49e14eb7ebcd624f861cbd --- /dev/null +++ b/datapilot/persistence.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from sqlalchemy import DateTime, String, Text, create_engine +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker + +from datapilot.config import Settings + + +class Base(DeclarativeBase): + pass + + +class AnalysisRun(Base): + __tablename__ = "analysis_runs" + + run_id: Mapped[str] = mapped_column(String(64), primary_key=True) + dataset_name: Mapped[str] = mapped_column(String(255)) + status: Mapped[str] = mapped_column(String(32), index=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + payload_json: Mapped[str] = mapped_column(Text) + + +class RunStore: + """SQLite locally; set DATABASE_URL to PostgreSQL in production.""" + + def __init__(self, settings: Settings): + self.engine = create_engine(settings.database_url, future=True) + Base.metadata.create_all(self.engine) + self.sessions = sessionmaker(self.engine, expire_on_commit=False) + + def save(self, run_id: str, dataset_name: str, status: str, payload: dict[str, Any]) -> None: + now = datetime.now(UTC) + with self.sessions.begin() as session: + record = session.get(AnalysisRun, run_id) + if record is None: + record = AnalysisRun( + run_id=run_id, + dataset_name=dataset_name, + status=status, + created_at=now, + updated_at=now, + payload_json=json.dumps(payload, default=str), + ) + session.add(record) + else: + record.status = status + record.updated_at = now + record.payload_json = json.dumps(payload, default=str) + + def get(self, run_id: str) -> dict[str, Any] | None: + with self.sessions() as session: + record = session.get(AnalysisRun, run_id) + return json.loads(record.payload_json) if record else None + + def list_recent(self, limit: int = 20) -> list[dict[str, Any]]: + from sqlalchemy import select + + with self.sessions() as session: + records = session.scalars( + select(AnalysisRun).order_by(AnalysisRun.created_at.desc()).limit(limit) + ) + return [ + { + "run_id": record.run_id, + "dataset_name": record.dataset_name, + "status": record.status, + "created_at": record.created_at.isoformat(), + } + for record in records + ] + + +class ArtifactStore: + """Local artifact storage with an interface that can be replaced by S3/MinIO.""" + + def __init__(self, root: Path): + self.root = root + self.root.mkdir(parents=True, exist_ok=True) + + def run_directory(self, run_id: str) -> Path: + directory = (self.root / run_id).resolve() + if self.root.resolve() not in directory.parents: + raise ValueError("Invalid run identifier.") + directory.mkdir(parents=True, exist_ok=True) + return directory diff --git a/datapilot/quality.py b/datapilot/quality.py new file mode 100644 index 0000000000000000000000000000000000000000..09a5a4e373f44895cb7e9fc87c45319041535816 --- /dev/null +++ b/datapilot/quality.py @@ -0,0 +1,243 @@ +from __future__ import annotations + +import re +from uuid import uuid4 + +import numpy as np +import pandas as pd + +from datapilot.schemas import ( + DatasetProfile, + Evidence, + QualityIssue, + Severity, + TaskType, +) + +LEAKAGE_PATTERNS = re.compile( + r"(target|label|outcome|result|prediction|predicted|probability|score)$", + re.IGNORECASE, +) + + +def infer_task_type(target: pd.Series) -> TaskType: + unique = int(target.nunique(dropna=True)) + if ( + not pd.api.types.is_numeric_dtype(target) + or pd.api.types.is_bool_dtype(target) + or unique <= 20 + or unique / max(len(target), 1) < 0.05 + ): + return TaskType.classification + return TaskType.regression + + +def build_profile(frame: pd.DataFrame, target: str) -> DatasetProfile: + if target not in frame.columns: + raise ValueError(f"Target column '{target}' is not present.") + numeric = frame.select_dtypes(include=np.number).columns.tolist() + categorical = frame.select_dtypes(include=["object", "category", "bool"]).columns.tolist() + datetime = frame.select_dtypes(include=["datetime", "datetimetz"]).columns.tolist() + missing_cells = int(frame.isna().sum().sum()) + return DatasetProfile( + rows=len(frame), + columns=len(frame.columns), + numeric_columns=numeric, + categorical_columns=categorical, + datetime_columns=datetime, + duplicate_rows=int(frame.duplicated().sum()), + missing_cells=missing_cells, + missing_rate=round(missing_cells / max(frame.size, 1), 4), + memory_mb=round(frame.memory_usage(deep=True).sum() / 1_048_576, 3), + target=target, + task_type=infer_task_type(frame[target]), + target_cardinality=int(frame[target].nunique(dropna=True)), + ) + + +def audit_quality( + frame: pd.DataFrame, profile: DatasetProfile +) -> tuple[list[QualityIssue], list[Evidence]]: + issues: list[QualityIssue] = [] + evidence: list[Evidence] = [] + + def add_evidence(claim: str, metric: str, value: object, source: str, method: str) -> str: + evidence_id = f"EV-{uuid4().hex[:8].upper()}" + evidence.append( + Evidence( + evidence_id=evidence_id, + claim=claim, + metric=metric, + value=value, + source=source, + method=method, + ) + ) + return evidence_id + + missing_id = add_evidence( + "Dataset missingness was measured across all cells.", + "missing_rate", + profile.missing_rate, + "uploaded_dataset", + "pandas.isna", + ) + if profile.missing_rate > 0.2: + issues.append( + QualityIssue( + code="HIGH_MISSINGNESS", + severity=Severity.critical, + message=f"{profile.missing_rate:.1%} of dataset cells are missing.", + evidence_ids=[missing_id], + ) + ) + elif profile.missing_rate > 0: + issues.append( + QualityIssue( + code="MISSING_VALUES", + severity=Severity.warning, + message=f"{profile.missing_rate:.1%} of dataset cells are missing.", + evidence_ids=[missing_id], + ) + ) + + duplicate_id = add_evidence( + "Exact duplicate rows were counted before splitting.", + "duplicate_rows", + profile.duplicate_rows, + "uploaded_dataset", + "pandas.duplicated", + ) + if profile.duplicate_rows: + issues.append( + QualityIssue( + code="DUPLICATE_ROWS", + severity=Severity.warning, + message=f"{profile.duplicate_rows:,} exact duplicate rows can bias validation.", + evidence_ids=[duplicate_id], + ) + ) + + target = frame[profile.target] + target_missing = int(target.isna().sum()) + target_missing_id = add_evidence( + "Rows with missing labels cannot be used for supervised training.", + "missing_target_rows", + target_missing, + f"column:{profile.target}", + "pandas.isna", + ) + if target_missing: + issues.append( + QualityIssue( + code="MISSING_TARGET", + severity=Severity.critical, + column=profile.target, + message=f"{target_missing:,} rows have no target value and will be excluded.", + evidence_ids=[target_missing_id], + ) + ) + + if profile.task_type == TaskType.classification: + distribution = target.value_counts(normalize=True, dropna=True) + minority_share = float(distribution.min()) if not distribution.empty else 0.0 + imbalance_id = add_evidence( + "Class imbalance was measured using the minority-class share.", + "minority_class_share", + round(minority_share, 4), + f"column:{profile.target}", + "normalized value counts", + ) + if minority_share < 0.1: + issues.append( + QualityIssue( + code="CLASS_IMBALANCE", + severity=Severity.warning, + column=profile.target, + message=f"Minority class represents only {minority_share:.1%} of labeled rows.", + evidence_ids=[imbalance_id], + ) + ) + + feature_frame = frame.drop(columns=[profile.target]) + for column in feature_frame.columns: + normalized = column.strip().lower() + leakage_risk = bool(LEAKAGE_PATTERNS.search(normalized)) + if feature_frame[column].nunique(dropna=True) == len(feature_frame): + leakage_risk = leakage_risk or normalized.endswith(("_id", "id")) + if leakage_risk: + evidence_id = add_evidence( + "A feature name or cardinality pattern may reveal the target or row identity.", + "suspected_leakage_feature", + column, + f"column:{column}", + "name and cardinality heuristic", + ) + issues.append( + QualityIssue( + code="LEAKAGE_RISK", + severity=Severity.warning, + column=column, + message=f"'{column}' may leak target or row identity; review before deployment.", + evidence_ids=[evidence_id], + ) + ) + + numeric = feature_frame.select_dtypes(include=np.number) + for column in numeric.columns: + series = numeric[column].dropna() + if len(series) < 8: + continue + q1, q3 = series.quantile([0.25, 0.75]) + iqr = q3 - q1 + if iqr == 0: + continue + outlier_rate = float(((series < q1 - 1.5 * iqr) | (series > q3 + 1.5 * iqr)).mean()) + if outlier_rate > 0.05: + evidence_id = add_evidence( + "Potential outliers were detected with the 1.5×IQR rule.", + "outlier_rate", + round(outlier_rate, 4), + f"column:{column}", + "Tukey IQR", + ) + issues.append( + QualityIssue( + code="OUTLIER_RATE", + severity=Severity.info, + column=column, + message=f"'{column}' has {outlier_rate:.1%} potential outliers.", + evidence_ids=[evidence_id], + ) + ) + + return issues, evidence + + +def drift_report(reference: pd.DataFrame, current: pd.DataFrame) -> list[dict[str, object]]: + """Population stability index for numeric columns shared by two datasets.""" + reports: list[dict[str, object]] = [] + shared = reference.select_dtypes(include=np.number).columns.intersection( + current.select_dtypes(include=np.number).columns + ) + for column in shared: + baseline = reference[column].dropna() + observed = current[column].dropna() + if baseline.nunique() < 2 or observed.empty: + continue + edges = np.unique(baseline.quantile(np.linspace(0, 1, 11)).to_numpy()) + if len(edges) < 3: + continue + expected_counts, _ = np.histogram(baseline, bins=edges) + actual_counts, _ = np.histogram(observed, bins=edges) + expected = np.clip(expected_counts / max(expected_counts.sum(), 1), 1e-6, None) + actual = np.clip(actual_counts / max(actual_counts.sum(), 1), 1e-6, None) + psi = float(np.sum((actual - expected) * np.log(actual / expected))) + reports.append( + { + "column": column, + "psi": round(psi, 4), + "status": "high" if psi >= 0.25 else "moderate" if psi >= 0.1 else "stable", + } + ) + return reports diff --git a/datapilot/reports.py b/datapilot/reports.py new file mode 100644 index 0000000000000000000000000000000000000000..3c569d4c6eef3799c5da40eb88ddb602725fd656 --- /dev/null +++ b/datapilot/reports.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import html +import json +from pathlib import Path +from typing import Any + +import joblib + + +def export_artifacts( + run_id: str, + state: dict[str, Any], + run_directory: Path, +) -> dict[str, str]: + run_directory.mkdir(parents=True, exist_ok=True) + bundle = state["model_bundle"] + summary = state["summary_payload"] + + pipeline_path = run_directory / "model_pipeline.joblib" + joblib.dump(bundle.pipeline, pipeline_path) + + metrics_path = run_directory / "metrics.json" + metrics_path.write_text(json.dumps(summary, indent=2, default=str), encoding="utf-8") + + model_card_path = run_directory / "MODEL_CARD.md" + model_card_path.write_text(_model_card(summary), encoding="utf-8") + + report_path = run_directory / "analysis_report.html" + report_path.write_text(_html_report(summary), encoding="utf-8") + + requirements_path = run_directory / "reproduction.json" + requirements_path.write_text( + json.dumps( + { + "run_id": run_id, + "random_state": state["settings"].random_state, + "test_size": state["settings"].test_size, + "target": summary["profile"]["target"], + "task_type": summary["profile"]["task_type"], + "best_model": summary["best_model"], + }, + indent=2, + ), + encoding="utf-8", + ) + return { + "pipeline": str(pipeline_path), + "metrics": str(metrics_path), + "model_card": str(model_card_path), + "report": str(report_path), + "reproduction": str(requirements_path), + } + + +def _model_card(run: dict[str, Any]) -> str: + best = run["model_results"][0] + profile = run["profile"] + issues = ( + "\n".join(f"- {item['message']}" for item in run["quality_issues"]) or "- None detected" + ) + return f"""# Model Card — {run["dataset_name"]} + +## Model details + +- Run ID: `{run["run_id"]}` +- Task: {profile["task_type"]} +- Target: `{profile["target"]}` +- Selected model: **{run["best_model"]}** +- Training-CV selection metric: `{best["primary_metric"]} = {best["selection_score"]:.4f}` +- One-time untouched test metric: `{best["primary_metric"]} = {best["final_test_score"]:.4f}` +- Training rows before split: {profile["rows"]:,} + +## Intended use + +Exploratory decision support and portfolio demonstration. Validate with domain-specific, +out-of-time data before any consequential or production use. + +## Evaluation + +```json +{json.dumps(best["final_test_metrics"], indent=2)} +``` + +## Data-quality observations + +{issues} + +## Explainability + +Method: **{run["explainability"]["method"]}**. Importance values are predictive associations, +not evidence of causation. + +## Limitations + +- Results depend on the uploaded dataset and chosen target. +- Automated task inference can be wrong; a domain owner should confirm the objective. +- Fairness, privacy, and legal review are outside the automatic approval gate. +""" + + +def _html_report(run: dict[str, Any]) -> str: + best = run["model_results"][0] + summary_items = "".join(f"{html.escape(run["dataset_name"])} · Run {html.escape(run["run_id"])}
| Metric | Value |
|---|
| Severity | Code | Observation |
|---|
Generated from computed evidence. Predictive findings do not establish causality.
+""" diff --git a/datapilot/safety.py b/datapilot/safety.py new file mode 100644 index 0000000000000000000000000000000000000000..d5ba978fb9c9b465a7e4f56ee05daf9b317ec000 --- /dev/null +++ b/datapilot/safety.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import ast + +BLOCKED_NODES = ( + ast.Import, + ast.ImportFrom, + ast.Global, + ast.Nonlocal, + ast.With, + ast.AsyncWith, + ast.Try, + ast.Raise, + ast.ClassDef, + ast.FunctionDef, + ast.AsyncFunctionDef, +) +BLOCKED_CALLS = { + "eval", + "exec", + "compile", + "open", + "input", + "__import__", + "breakpoint", + "getattr", + "setattr", + "delattr", +} + + +class UnsafeCodeError(ValueError): + pass + + +def validate_generated_expression(expression: str, max_characters: int = 2_000) -> ast.Expression: + """Validate a calculation expression before it is sent to an isolated worker. + + DataPilot's primary workflow does not execute LLM-generated Python. This validator + exists for an optional restricted calculation worker and accepts expressions only. + """ + if len(expression) > max_characters: + raise UnsafeCodeError("Expression exceeds the configured size limit.") + try: + tree = ast.parse(expression, mode="eval") + except SyntaxError as exc: + raise UnsafeCodeError("Expression is not valid Python.") from exc + for node in ast.walk(tree): + if isinstance(node, BLOCKED_NODES): + raise UnsafeCodeError(f"Blocked syntax: {type(node).__name__}.") + if isinstance(node, ast.Attribute) and node.attr.startswith("__"): + raise UnsafeCodeError("Dunder attribute access is blocked.") + if isinstance(node, ast.Name) and node.id.startswith("__"): + raise UnsafeCodeError("Dunder names are blocked.") + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Name) and node.func.id in BLOCKED_CALLS: + raise UnsafeCodeError(f"Blocked call: {node.func.id}.") + if isinstance(node.func, ast.Attribute) and node.func.attr in BLOCKED_CALLS: + raise UnsafeCodeError(f"Blocked call: {node.func.attr}.") + return tree diff --git a/datapilot/schemas.py b/datapilot/schemas.py new file mode 100644 index 0000000000000000000000000000000000000000..0223f4700c6f742dbb12ed02454d07aa613d8642 --- /dev/null +++ b/datapilot/schemas.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, Field + + +class TaskType(StrEnum): + classification = "classification" + regression = "regression" + + +class Severity(StrEnum): + info = "info" + warning = "warning" + critical = "critical" + + +class Evidence(BaseModel): + evidence_id: str + claim: str + metric: str + value: float | int | str + source: str + method: str + + +class QualityIssue(BaseModel): + code: str + severity: Severity + column: str | None = None + message: str + evidence_ids: list[str] = Field(default_factory=list) + + +class DatasetProfile(BaseModel): + rows: int + columns: int + numeric_columns: list[str] + categorical_columns: list[str] + datetime_columns: list[str] + duplicate_rows: int + missing_cells: int + missing_rate: float + memory_mb: float + target: str + task_type: TaskType + target_cardinality: int + + +class AnalysisPlan(BaseModel): + objective: str + target: str + task_type: TaskType + primary_metric: str + validation_strategy: str + candidate_models: list[str] + risk_controls: list[str] + + +class ModelResult(BaseModel): + name: str + primary_metric: str + primary_score: float + metrics: dict[str, float] + cross_validation_mean: float | None = None + cross_validation_std: float | None = None + training_seconds: float + selection_score: float | None = None + final_test_score: float | None = None + final_test_metrics: dict[str, float] = Field(default_factory=dict) + + +class ModelFailure(BaseModel): + name: str + stage: str + exception_category: str + sanitized_error: str + training_seconds: float + expected: bool = False + + +class CriticDecision(BaseModel): + approved: bool + score: float + threshold: float + reasons: list[str] + retry_number: int + + +class ExplainabilityResult(BaseModel): + method: str + feature_importance: dict[str, float] + caveats: list[str] + + +class RunSummary(BaseModel): + run_id: str + status: str + dataset_name: str + profile: DatasetProfile + plan: AnalysisPlan + quality_issues: list[QualityIssue] + evidence: list[Evidence] + model_results: list[ModelResult] + model_failures: list[ModelFailure] = Field(default_factory=list) + best_model: str + critic: CriticDecision + explainability: ExplainabilityResult + executive_summary: list[str] + recommendations: list[str] + artifacts: dict[str, str] + trace: list[dict[str, Any]] diff --git a/datapilot/tuning.py b/datapilot/tuning.py new file mode 100644 index 0000000000000000000000000000000000000000..65be86774ed468f3443102c148e418be781a2c5d --- /dev/null +++ b/datapilot/tuning.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from typing import Any + +import pandas as pd +from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor +from sklearn.model_selection import cross_val_score +from sklearn.pipeline import Pipeline + +from datapilot.config import Settings +from datapilot.schemas import TaskType + + +def tune_random_forest( + task: TaskType, + preprocessor: Any, + features: pd.DataFrame, + target: pd.Series, + settings: Settings, + cv: int, +) -> dict[str, Any]: + """Run a bounded Optuna study when the optional AutoML extra is installed.""" + if settings.optuna_trials <= 0 or cv < 2: + return {} + try: + import optuna + except ImportError: + return {} + + optuna.logging.set_verbosity(optuna.logging.WARNING) + scoring = "balanced_accuracy" if task == TaskType.classification else "r2" + + def objective(trial): + parameters = { + "n_estimators": trial.suggest_int("n_estimators", 120, 360, step=60), + "max_depth": trial.suggest_int("max_depth", 3, 14), + "min_samples_leaf": trial.suggest_int("min_samples_leaf", 1, 6), + "max_features": trial.suggest_categorical("max_features", ["sqrt", "log2", 0.8]), + } + common = { + **parameters, + "n_jobs": -1, + "random_state": settings.random_state, + } + if task == TaskType.classification: + estimator = RandomForestClassifier(**common, class_weight="balanced") + else: + estimator = RandomForestRegressor(**common) + pipeline = Pipeline([("preprocessor", preprocessor), ("model", estimator)]) + scores = cross_val_score( + pipeline, + features, + target, + cv=cv, + scoring=scoring, + n_jobs=1, + ) + return float(scores.mean()) + + study = optuna.create_study(direction="maximize") + study.optimize( + objective, + n_trials=settings.optuna_trials, + timeout=90, + show_progress_bar=False, + ) + return study.best_params diff --git a/datapilot/workflow.py b/datapilot/workflow.py new file mode 100644 index 0000000000000000000000000000000000000000..edb5708047003594994441e162159d1679ff3698 --- /dev/null +++ b/datapilot/workflow.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import time +from typing import Any, TypedDict +from uuid import uuid4 + +import numpy as np +import pandas as pd +from langgraph.graph import END, START, StateGraph + +from datapilot.config import Settings, get_settings +from datapilot.data import duckdb_overview +from datapilot.insights import deterministic_insights, optional_llm_narrative +from datapilot.modeling import ( + TrainingBundle, + critic_decision, + explain_model, + train_models, +) +from datapilot.observability import log_to_mlflow +from datapilot.persistence import ArtifactStore, RunStore +from datapilot.quality import audit_quality, build_profile +from datapilot.reports import export_artifacts +from datapilot.schemas import AnalysisPlan, RunSummary, TaskType + + +class AgentState(TypedDict, total=False): + run_id: str + dataset_name: str + frame: pd.DataFrame + target: str + settings: Settings + profile: Any + quality_issues: list[Any] + evidence: list[Any] + eda: dict[str, Any] + statistics: dict[str, Any] + plan: AnalysisPlan + feature_plan: dict[str, Any] + model_bundle: TrainingBundle + critic: Any + explainability: Any + executive_summary: list[str] + recommendations: list[str] + summary_payload: dict[str, Any] + artifacts: dict[str, str] + trace: list[dict[str, Any]] + + +def _trace(state: AgentState, agent: str, started: float, detail: str) -> list[dict[str, Any]]: + trace = list(state.get("trace", [])) + trace.append( + { + "agent": agent, + "status": "completed", + "duration_seconds": round(time.perf_counter() - started, 3), + "detail": detail, + } + ) + return trace + + +def data_quality_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + profile = build_profile(state["frame"], state["target"]) + issues, evidence = audit_quality(state["frame"], profile) + return { + "profile": profile, + "quality_issues": issues, + "evidence": evidence, + "trace": _trace( + state, "Data Quality Agent", started, f"Recorded {len(issues)} quality observations." + ), + } + + +def eda_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + frame = state["frame"] + overview = duckdb_overview(frame) + overview["numeric_summary"] = ( + frame.select_dtypes(include=np.number).describe().round(4).to_dict() + ) + overview["categorical_cardinality"] = { + column: int(frame[column].nunique(dropna=True)) + for column in frame.select_dtypes(exclude=np.number).columns + } + return { + "eda": overview, + "trace": _trace(state, "EDA Agent", started, "Computed DuckDB-backed dataset overview."), + } + + +def statistical_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + frame = state["frame"] + target = state["target"] + numeric = frame.select_dtypes(include=np.number) + correlations: dict[str, float] = {} + if target in numeric.columns and len(numeric.columns) > 1: + correlations = ( + numeric.corr(numeric_only=True)[target] + .drop(labels=[target]) + .abs() + .sort_values(ascending=False) + .head(10) + .round(4) + .to_dict() + ) + statistics = { + "top_absolute_target_correlations": correlations, + "target_distribution": frame[target].value_counts(dropna=False).head(20).to_dict(), + } + return { + "statistics": statistics, + "trace": _trace( + state, + "Statistical Analysis Agent", + started, + "Measured target distribution and associations.", + ), + } + + +def planning_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + profile = state["profile"] + if profile.task_type == TaskType.classification: + metric = "balanced_accuracy" + candidates = [ + "Logistic Regression", + "Random Forest", + "Extra Trees", + "Histogram Gradient Boosting", + "XGBoost (when installed)", + ] + else: + metric = "r2" + candidates = [ + "Linear Regression", + "Random Forest", + "Extra Trees", + "Histogram Gradient Boosting", + "XGBoost (when installed)", + ] + plan = AnalysisPlan( + objective=f"Predict '{profile.target}' and produce reproducible, evidence-backed insights.", + target=profile.target, + task_type=profile.task_type, + primary_metric=metric, + validation_strategy="Training-only stratified cross-validation; untouched final test evaluation" + if profile.task_type == TaskType.classification + else "Training-only cross-validation; untouched final test evaluation", + candidate_models=candidates, + risk_controls=[ + "Drop rows with missing target before split", + "Fit imputers, encoders, and scalers on training folds only", + "Flag leakage-like names and identifier cardinality", + "Require critic quality gate before explanation", + ], + ) + return { + "plan": plan, + "trace": _trace(state, "Planning Agent", started, f"Selected {metric} as primary metric."), + } + + +def feature_engineering_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + profile = state["profile"] + feature_plan = { + "numeric": "Median imputation followed by standard scaling", + "categorical": "Most-frequent imputation followed by unknown-safe one-hot encoding", + "fit_scope": "Preprocessing is fitted inside each sklearn Pipeline after splitting", + "dropped": ["exact duplicate rows", "rows with missing target"], + "feature_count": profile.columns - 1, + } + return { + "feature_plan": feature_plan, + "trace": _trace( + state, + "Feature Engineering Agent", + started, + "Created leakage-safe ColumnTransformer plan.", + ), + } + + +def modeling_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + retry = state.get("model_bundle").retry_number + 1 if state.get("model_bundle") else 0 + bundle = train_models( + state["frame"], + state["target"], + state["profile"].task_type, + state["settings"], + retry_number=retry, + ) + return { + "model_bundle": bundle, + "trace": _trace( + state, + "Modeling Agent", + started, + f"Compared {len(bundle.results)} models; {bundle.best_model} ranked first.", + ), + } + + +def evaluation_critic_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + decision = critic_decision(state["model_bundle"], state["profile"].task_type, state["settings"]) + detail = "Approved analysis." if decision.approved else "Rejected analysis and requested retry." + return { + "critic": decision, + "trace": _trace(state, "Evaluation / Critic Agent", started, detail), + } + + +def critic_route(state: AgentState) -> str: + return "explainability" if state["critic"].approved else "retry_modeling" + + +def explainability_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + result = explain_model(state["model_bundle"]) + return { + "explainability": result, + "trace": _trace( + state, "Explainability Agent", started, f"Generated {result.method} explanations." + ), + } + + +def executive_insights_agent(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + summary, recommendations = deterministic_insights(state) + llm_summary = optional_llm_narrative(state, state["evidence"], state["settings"]) + if llm_summary: + summary = llm_summary + return { + "executive_summary": summary, + "recommendations": recommendations, + "trace": _trace( + state, + "Executive Insights Agent", + started, + "Created evidence-grounded narrative with deterministic metric provenance.", + ), + } + + +def build_graph(): + graph = StateGraph(AgentState) + graph.add_node("data_quality", data_quality_agent) + graph.add_node("eda", eda_agent) + graph.add_node("statistics", statistical_agent) + graph.add_node("planning", planning_agent) + graph.add_node("feature_engineering", feature_engineering_agent) + graph.add_node("modeling", modeling_agent) + graph.add_node("critic", evaluation_critic_agent) + graph.add_node("explainability", explainability_agent) + graph.add_node("executive_insights", executive_insights_agent) + graph.add_edge(START, "data_quality") + graph.add_edge("data_quality", "eda") + graph.add_edge("eda", "statistics") + graph.add_edge("statistics", "planning") + graph.add_edge("planning", "feature_engineering") + graph.add_edge("feature_engineering", "modeling") + graph.add_edge("modeling", "critic") + graph.add_conditional_edges( + "critic", + critic_route, + {"retry_modeling": "modeling", "explainability": "explainability"}, + ) + graph.add_edge("explainability", "executive_insights") + graph.add_edge("executive_insights", END) + return graph.compile() + + +def run_analysis( + frame: pd.DataFrame, + target: str, + dataset_name: str, + settings: Settings | None = None, +) -> RunSummary: + settings = settings or get_settings() + run_id = f"run_{uuid4().hex[:12]}" + store = RunStore(settings) + artifact_store = ArtifactStore(settings.artifact_root) + store.save(run_id, dataset_name, "running", {"run_id": run_id, "status": "running"}) + initial: AgentState = { + "run_id": run_id, + "dataset_name": dataset_name, + "frame": frame, + "target": target, + "settings": settings, + "trace": [], + } + try: + final = build_graph().invoke(initial) + payload = _summary_payload(final, run_id, dataset_name) + final["summary_payload"] = payload + log_to_mlflow(payload, settings) + artifacts = export_artifacts(run_id, final, artifact_store.run_directory(run_id)) + payload["artifacts"] = artifacts + payload["status"] = "completed" + store.save(run_id, dataset_name, "completed", payload) + return RunSummary.model_validate(payload) + except Exception as exc: + store.save( + run_id, + dataset_name, + "failed", + {"run_id": run_id, "dataset_name": dataset_name, "status": "failed", "error": str(exc)}, + ) + raise + finally: + store.engine.dispose() + + +def _summary_payload(state: AgentState, run_id: str, dataset_name: str) -> dict[str, Any]: + bundle = state["model_bundle"] + return { + "run_id": run_id, + "status": "completed", + "dataset_name": dataset_name, + "profile": state["profile"].model_dump(mode="json"), + "plan": state["plan"].model_dump(mode="json"), + "quality_issues": [item.model_dump(mode="json") for item in state["quality_issues"]], + "evidence": [item.model_dump(mode="json") for item in state["evidence"]], + "model_results": [item.model_dump(mode="json") for item in bundle.results], + "model_failures": [item.model_dump(mode="json") for item in bundle.failures], + "best_model": bundle.best_model, + "critic": state["critic"].model_dump(mode="json"), + "explainability": state["explainability"].model_dump(mode="json"), + "executive_summary": state["executive_summary"], + "recommendations": state["recommendations"], + "artifacts": {}, + "trace": state["trace"], + } diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000000000000000000000000000000000000..2fc7dea3152f0289f7ad4bd2ac6850ef5d7591f4 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,63 @@ +services: + ui: + build: + context: . + dockerfile: Dockerfile + ports: + - "8501:8501" + environment: + ARTIFACT_ROOT: /app/artifacts + DATABASE_URL: sqlite:////app/artifacts/datapilot.db + volumes: + - datapilot_artifacts:/app/artifacts + read_only: true + tmpfs: + - /tmp:size=128m + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + mem_limit: 2g + cpus: 2 + + api: + build: + context: . + dockerfile: Dockerfile.api + ports: + - "8000:8000" + environment: + ARTIFACT_ROOT: /app/artifacts + DATABASE_URL: sqlite:////app/artifacts/datapilot.db + CORS_ORIGINS: http://localhost:8501 + volumes: + - datapilot_artifacts:/app/artifacts + read_only: true + tmpfs: + - /tmp:size=128m + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + mem_limit: 2g + cpus: 2 + + worker: + build: + context: . + dockerfile: Dockerfile.worker + network_mode: none + read_only: true + tmpfs: + - /tmp:size=32m,noexec,nosuid + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + mem_limit: 256m + cpus: 0.5 + pids_limit: 64 + +volumes: + datapilot_artifacts: + diff --git a/docs/API_EXAMPLES.md b/docs/API_EXAMPLES.md new file mode 100644 index 0000000000000000000000000000000000000000..2e43a5705d4ed7538991d6e10979727ef50ba3b1 --- /dev/null +++ b/docs/API_EXAMPLES.md @@ -0,0 +1,37 @@ +# API examples + +Start the service with `uvicorn api.main:app --port 8000`. If `API_KEY` is configured, include +`X-API-Key: