Create ai_assistant.py

analytics/ai_assistant.py

@@ -0,0 +1,157 @@
+import os
+import pandas as pd
+
+try:
+    from huggingface_hub import InferenceApi
+except ImportError as exc:
+    raise ImportError(
+        "huggingface_hub is required for AI assistant support. "
+        "Install it with `pip install huggingface_hub`."
+    ) from exc
+
+HF_MODEL_ID = os.environ.get("HF_MODEL_ID", "gpt2")
+HF_TOKEN = os.environ.get("HUGGINGFACE_API_TOKEN")
+HF_MAX_TOKENS = int(os.environ.get("HF_MAX_TOKENS", 256))
+HF_TEMPERATURE = float(os.environ.get("HF_TEMPERATURE", 0.3))
+
+
+def _detect_date_column(df: pd.DataFrame):
+    candidates = [
+        "reporting_month",
+        "observation_date",
+        "observation_month",
+        "obs_date",
+        "date",
+        "calendar_month",
+        "month",
+        "report_date"
+    ]
+    for c in candidates:
+        if c in df.columns:
+            return c
+    return None
+
+
+def _filter_by_month(df: pd.DataFrame, as_of_month: str):
+    if not as_of_month:
+        return df.copy()
+
+    date_col = _detect_date_column(df)
+    if date_col is None:
+        return df.copy()
+
+    ser = pd.to_datetime(df[date_col], errors="coerce").dt.to_period("M").astype(str)
+    return df[ser == as_of_month].copy()
+
+
+def _fmt_pct(value):
+    try:
+        return f"{round(float(value), 2)}%"
+    except Exception:
+        return "N/A"
+
+
+def build_portfolio_context(df: pd.DataFrame, as_of_month: str = None, segment: str = None):
+    df = _filter_by_month(df, as_of_month)
+
+    total_accounts = int(df["account_id"].nunique()) if "account_id" in df.columns else 0
+    open_accounts = int(df.loc[df["balance"] > 0, "account_id"].nunique()) if "balance" in df.columns else total_accounts
+    bad_accounts = int(df.loc[df["dpd"].fillna(0) >= 30, "account_id"].nunique()) if "dpd" in df.columns else 0
+    total_balance = float(df["balance"].sum(skipna=True)) if "balance" in df.columns else 0.0
+
+    ncl_cols = [c for c in df.columns if "ncl" in c.lower()]
+    if len(ncl_cols) > 0 and total_balance > 0:
+        overall_ncl_rate = df[ncl_cols[0]].sum(skipna=True) / total_balance * 100
+    elif total_balance > 0 and "dpd" in df.columns:
+        bad_balance = float(df.loc[df["dpd"].fillna(0) >= 30, "balance"].sum())
+        overall_ncl_rate = bad_balance / total_balance * 100
+    else:
+        overall_ncl_rate = None
+
+    if overall_ncl_rate is None:
+        overall_ncl_rate_text = "N/A"
+    else:
+        overall_ncl_rate_text = _fmt_pct(overall_ncl_rate)
+
+    if "fico_score" in df.columns:
+        avg_fico = round(df["fico_score"].dropna().mean(), 1)
+    elif "fico_band" in df.columns:
+        def band_mid(val):
+            try:
+                lo, hi = val.split("-")
+                return (int(lo) + int(hi)) / 2
+            except Exception:
+                return None
+        mid_vals = df["fico_band"].dropna().apply(band_mid).dropna()
+        avg_fico = round(mid_vals.mean(), 1) if not mid_vals.empty else None
+    else:
+        avg_fico = None
+
+    lines = [
+        f"As of month: {as_of_month or 'latest available'}",
+        f"Total accounts: {total_accounts}",
+        f"Open accounts: {open_accounts}",
+        f"Bad accounts (dpd>=30): {bad_accounts}",
+        f"Overall NCL rate: {overall_ncl_rate_text}",
+        f"Average FICO: {avg_fico if avg_fico is not None else 'N/A'}"
+    ]
+
+    if segment and segment in df.columns:
+        segment_summary = (
+            df.groupby(segment)
+            .agg(
+                accounts=("account_id", "nunique"),
+                balance=("balance", "sum"),
+                bad_balance=("balance", lambda x: x[df.loc[x.index, "dpd"].fillna(0) >= 30].sum() if "dpd" in df.columns else 0)
+            )
+            .reset_index()
+        )
+        if "balance" in df.columns:
+            segment_summary["ncl_rate"] = (segment_summary["bad_balance"] / segment_summary["balance"] * 100).round(2).fillna(0)
+        else:
+            segment_summary["ncl_rate"] = 0
+
+        lines.append(f"Segment breakdown by {segment}:")
+        for _, row in segment_summary.sort_values("ncl_rate", ascending=False).head(5).iterrows():
+            lines.append(
+                f"  - {row[segment]}: accounts={int(row['accounts'])}, balance={row['balance']:.0f}, ncl={_fmt_pct(row['ncl_rate'])}"
+            )
+
+    return "\n".join(lines)
+
+
+def _get_inference_client():
+    if not HF_TOKEN:
+        raise RuntimeError(
+            "HUGGINGFACE_API_TOKEN is required for AI assistant. "
+            "Set it in your environment before running the app."
+        )
+    return InferenceApi(repo_id=HF_MODEL_ID, token=HF_TOKEN)
+
+
+def generate_ai_answer(question: str, df: pd.DataFrame, as_of_month: str = None, segment: str = None):
+    summary = build_portfolio_context(df, as_of_month=as_of_month, segment=segment)
+    prompt = (
+        "You are a senior risk manager assistant responding to portfolio analytics questions. "
+        "Use the risk context below and answer the user clearly, describing what is happening, why it is happening, and what actions should be considered. "
+        "If you cannot answer from the data, say so clearly.
+
+" +
+        "Context:\n" + summary + "\n\n" +
+        "Question: " + question + "\n\n" +
+        "Answer as a risk manager with practical, concise guidance."
+    )
+
+    client = _get_inference_client()
+    response = client(prompt, parameters={
+        "max_new_tokens": HF_MAX_TOKENS,
+        "temperature": HF_TEMPERATURE,
+        "top_k": 50,
+        "top_p": 0.95,
+    })
+
+    if isinstance(response, dict):
+        return response.get("generated_text") or response.get("text") or str(response)
+    if isinstance(response, list) and len(response) > 0:
+        return response[0].get("generated_text", str(response[0]))
+    return str(response)