Add stable hosted evidence brief

Provide a deterministic evidence-only analyst narrative when hosted platforms restrict reliable external provider traffic.

datapilot/analyst.py

@@ -1,268 +1,297 @@
-from __future__ import annotations
-
-import hashlib
-import io
-import json
-import logging
-import re
-from dataclasses import dataclass
-from typing import Any
-from urllib.parse import quote
-
-import numpy as np
-import pandas as pd
-import requests
-
-logger = logging.getLogger(__name__)
-
-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 narrative through a bounded REST call."""
-    if not api_key:
-        raise ValueError("Enter a Gemini API key to generate AI interpretation.")
-
-    allowed_models = {"gemini-2.5-flash", "gemini-2.5-pro", "gemini-2.0-flash"}
-    if model not in allowed_models:
-        raise ValueError("The selected Gemini model is not supported.")
-
-    evidence = json.dumps(ai_context(frame, profile, excluded), default=str)
-    if len(evidence) > 30_000:
-        raise ValueError(
-            "The AI evidence package is too large. Exclude columns or use a smaller dataset."
-        )
-
-    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"""
-        + evidence
-    )
-
-    endpoint = (
-        "https://generativelanguage.googleapis.com/v1beta/models/"
-        f"{quote(model, safe='')}:generateContent"
-    )
-    payload = {
-        "contents": [{"role": "user", "parts": [{"text": prompt}]}],
-        "generationConfig": {"maxOutputTokens": 1200},
-    }
-
-    try:
-        logger.warning("gemini_request_started model=%s evidence_chars=%d", model, len(evidence))
-        response = requests.post(
-            endpoint,
-            headers={
-                "x-goog-api-key": api_key,
-                "Content-Type": "application/json",
-                "X-Server-Timeout": "30",
-            },
-            json=payload,
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import logging
+import re
+from dataclasses import dataclass
+from typing import Any
+from urllib.parse import quote
+
+import numpy as np
+import pandas as pd
+import requests
+
+logger = logging.getLogger(__name__)
+
+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 narrative through a bounded REST call."""
+    if not api_key:
+        raise ValueError("Enter a Gemini API key to generate AI interpretation.")
+
+    allowed_models = {"gemini-2.5-flash", "gemini-2.5-pro", "gemini-2.0-flash"}
+    if model not in allowed_models:
+        raise ValueError("The selected Gemini model is not supported.")
+
+    evidence = json.dumps(ai_context(frame, profile, excluded), default=str)
+    if len(evidence) > 30_000:
+        raise ValueError(
+            "The AI evidence package is too large. Exclude columns or use a smaller dataset."
+        )
+
+    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"""
+        + evidence
+    )
+
+    endpoint = (
+        "https://generativelanguage.googleapis.com/v1beta/models/"
+        f"{quote(model, safe='')}:generateContent"
+    )
+    payload = {
+        "contents": [{"role": "user", "parts": [{"text": prompt}]}],
+        "generationConfig": {"maxOutputTokens": 1200},
+    }
+
+    try:
+        logger.warning("gemini_request_started model=%s evidence_chars=%d", model, len(evidence))
+        response = requests.post(
+            endpoint,
+            headers={
+                "x-goog-api-key": api_key,
+                "Content-Type": "application/json",
+                "X-Server-Timeout": "30",
+            },
+            json=payload,
             timeout=(3, 8),
-        )
-        logger.warning("gemini_response_received status=%d", response.status_code)
-    except requests.Timeout as exc:
-        logger.warning("gemini_request_timed_out")
+        )
+        logger.warning("gemini_response_received status=%d", response.status_code)
+    except requests.Timeout as exc:
+        logger.warning("gemini_request_timed_out")
         raise ValueError("Gemini timed out after 8 seconds. Please try again.") from exc
-    except requests.ConnectionError as exc:
-        logger.warning("gemini_connection_failed")
-        raise ValueError("DataPilot could not connect to Gemini. Please try again.") from exc
-    except requests.RequestException as exc:
-        logger.warning("gemini_request_failed category=%s", type(exc).__name__)
-        raise ValueError("The Gemini request could not be completed.") from exc
-
-    status_messages = {
-        400: "Gemini rejected the evidence request.",
-        401: "The Gemini API key was rejected.",
-        403: "Gemini access is not enabled for this API key or project.",
-        404: "The selected Gemini model is unavailable. Select Gemini Flash.",
-        429: "Gemini quota is temporarily exhausted. Please try again later.",
-    }
-    if response.status_code in status_messages:
-        raise ValueError(status_messages[response.status_code])
-    if response.status_code >= 500:
-        raise ValueError("Gemini is temporarily unavailable. Please try again later.")
-    if not response.ok:
-        raise ValueError(f"Gemini returned an unexpected response ({response.status_code}).")
-
-    try:
-        body = response.json()
-        parts = body["candidates"][0]["content"]["parts"]
-        text = "\n".join(
-            part["text"].strip() for part in parts if isinstance(part, dict) and part.get("text")
-        )
-    except (ValueError, KeyError, IndexError, TypeError) as exc:
-        raise ValueError("Gemini returned an invalid or empty response.") from exc
-
-    if not text:
-        raise ValueError("Gemini returned no text. The request may have been blocked.")
-    logger.warning("gemini_response_validated output_chars=%d", len(text))
-    return text
-
-
-def dataframe_csv(frame: pd.DataFrame) -> bytes:
-    buffer = io.StringIO()
-    frame.to_csv(buffer, index=False)
-    return buffer.getvalue().encode("utf-8")
+    except requests.ConnectionError as exc:
+        logger.warning("gemini_connection_failed")
+        raise ValueError("DataPilot could not connect to Gemini. Please try again.") from exc
+    except requests.RequestException as exc:
+        logger.warning("gemini_request_failed category=%s", type(exc).__name__)
+        raise ValueError("The Gemini request could not be completed.") from exc
+
+    status_messages = {
+        400: "Gemini rejected the evidence request.",
+        401: "The Gemini API key was rejected.",
+        403: "Gemini access is not enabled for this API key or project.",
+        404: "The selected Gemini model is unavailable. Select Gemini Flash.",
+        429: "Gemini quota is temporarily exhausted. Please try again later.",
+    }
+    if response.status_code in status_messages:
+        raise ValueError(status_messages[response.status_code])
+    if response.status_code >= 500:
+        raise ValueError("Gemini is temporarily unavailable. Please try again later.")
+    if not response.ok:
+        raise ValueError(f"Gemini returned an unexpected response ({response.status_code}).")
+
+    try:
+        body = response.json()
+        parts = body["candidates"][0]["content"]["parts"]
+        text = "\n".join(
+            part["text"].strip() for part in parts if isinstance(part, dict) and part.get("text")
+        )
+    except (ValueError, KeyError, IndexError, TypeError) as exc:
+        raise ValueError("Gemini returned an invalid or empty response.") from exc
+
+    if not text:
+        raise ValueError("Gemini returned no text. The request may have been blocked.")
+    logger.warning("gemini_response_validated output_chars=%d", len(text))
+    return text
+
+
+def evidence_dataset_summary(frame: pd.DataFrame, profile: dict[str, Any]) -> str:
+    """Create a deterministic, evidence-only brief for restricted hosted runtimes."""
+    rows, columns = frame.shape
+    missing = int(frame.isna().sum().sum())
+    duplicates = int(frame.duplicated().sum())
+    numeric = len(frame.select_dtypes(include=np.number).columns)
+    target_names = [str(item.get("column")) for item in profile.get("targets", [])[:3]]
+    targets = ", ".join(target_names) if target_names else "No strong target candidate detected"
+    return (
+        "### Finding\n"
+        f"The dataset contains **{rows:,} rows and {columns:,} columns**, including "
+        f"**{numeric:,} numeric features**. Its computed quality score is "
+        f"**{profile.get('quality_score', 'not available')}/100**.\n\n"
+        "### Evidence\n"
+        f"The deterministic profile found **{missing:,} missing cells** and "
+        f"**{duplicates:,} duplicate rows**. Leading analytical target candidates: {targets}.\n\n"
+        "### Interpretation\n"
+        "The dataset is suitable for exploratory analysis when its row grain and field "
+        "definitions are confirmed. Target candidates are structural recommendations, not "
+        "proof of business relevance.\n\n"
+        "### Limitation\n"
+        "This hosted brief is generated from computed evidence without an external LLM. "
+        "Associations and model scores must not be interpreted as causal effects.\n\n"
+        "### Recommendation\n"
+        "Confirm the business objective and target meaning, review quality findings, then "
+        "use Model Lab with an untouched test set for final evaluation."
+    )
+
+
+def dataframe_csv(frame: pd.DataFrame) -> bytes:
+    buffer = io.StringIO()
+    frame.to_csv(buffer, index=False)
+    return buffer.getvalue().encode("utf-8")