Update main.py

main.py

@@ -396,6 +396,24 @@ class IrisReportEngine:
     - Never uses LLM for numbers. LLM only for narration elsewhere.
     """
 
+    # ---- Canonical column names (single source of truth; no magic strings sprinkled around) ----
+    COL_INVOICE   = "_Invoice"
+    COL_PRODUCT   = "_Product"
+    COL_TELLER    = "_Teller"
+    COL_TXNTYPE   = "_TxnType"
+    COL_BRANCH    = "_Branch"
+    COL_CUSTOMER  = "_Customer"
+    COL_DT        = "_datetime"
+    COL_AMOUNT    = "_Amount"
+    COL_UNITS     = "_Units"
+    COL_UNITCOST  = "_UnitCost"
+    COL_REVENUE   = "_Revenue"
+    COL_COGS      = "_COGS"
+    COL_GP        = "_GrossProfit"
+    COL_HOUR      = "_Hour"
+    COL_DOW       = "_DOW"
+    COL_DOWI      = "_DOW_idx"
+
     DEFAULT_PARAMS = {
         "top_k": 5,
         "min_revenue_for_margin_pct": 50.0,
@@ -414,8 +432,8 @@ class IrisReportEngine:
         profile_id: str,
         transactions_data: List[dict],
         llm_instance,
-        stock_feed: Optional[List[Dict[str, Any]]] = None,   # optional: [{product, stock_on_hand, reorder_point, lead_time_days, min_order_qty}]
-        cash_float_feed: Optional[List[Dict[str, Any]]] = None,  # optional: [{branch, date, opening_float, closing_float, drops, petty_cash, declared_cash}]
+        stock_feed: Optional[List[Dict[str, Any]]] = None,
+        cash_float_feed: Optional[List[Dict[str, Any]]] = None,
         params: Optional[Dict[str, Any]] = None,
     ):
         self.profile_id = profile_id
@@ -428,6 +446,26 @@ class IrisReportEngine:
         self.df = self._load_and_prepare_data(self.raw)
         self.currency = self._get_primary_currency()
 
+    # ------------------------- small helpers -------------------------
+    @staticmethod
+    def _rfm_to_list(df: pd.DataFrame) -> List[Dict[str, Any]]:
+        if df is None or df.empty:
+            return []
+        out = []
+        for _, r in df.iterrows():
+            out.append({
+                "customer": str(r.get("customer")),
+                "orders": int(r.get("orders", 0)),
+                "revenue": float(r.get("revenue", 0.0)),
+                "gp": float(r.get("gp", 0.0)),
+                "recency_days": float(r.get("recency_days", np.nan)) if pd.notna(r.get("recency_days")) else None,
+                "avg_basket_value": float(r.get("avg_basket_value", np.nan)) if pd.notna(r.get("avg_basket_value")) else None,
+            })
+        return out
+
+    def _has(self, df: pd.DataFrame, col: str) -> bool:
+        return isinstance(df, pd.DataFrame) and col in df.columns
+
     # ------------------------- load/prepare -------------------------
 
     def _load_and_prepare_data(self, df: pd.DataFrame) -> pd.DataFrame:
@@ -439,21 +477,21 @@ class IrisReportEngine:
         emit_kpi_debug(self.profile_id, "column_map", mapping)
 
         # Numerics
-        amt_col = mapping["amount"] or "Settled_Amount" if "Settled_Amount" in df.columns else None
+        amt_col = mapping["amount"] or ("Settled_Amount" if "Settled_Amount" in df.columns else None)
         if amt_col and amt_col in df:
-            df["_Amount"] = pd.to_numeric(df[amt_col], errors="coerce")
+            df[self.COL_AMOUNT] = pd.to_numeric(df[amt_col], errors="coerce")
         else:
-            df["_Amount"] = pd.Series(dtype=float)
+            df[self.COL_AMOUNT] = pd.Series(dtype=float)
 
         if mapping["units"] and mapping["units"] in df:
-            df["_Units"] = pd.to_numeric(df[mapping["units"]], errors="coerce").fillna(0)
+            df[self.COL_UNITS] = pd.to_numeric(df[mapping["units"]], errors="coerce").fillna(0)
         else:
-            df["_Units"] = 0
+            df[self.COL_UNITS] = 0
 
         if mapping["unit_cost"] and mapping["unit_cost"] in df:
-            df["_UnitCost"] = pd.to_numeric(df[mapping["unit_cost"]], errors="coerce").fillna(0.0)
+            df[self.COL_UNITCOST] = pd.to_numeric(df[mapping["unit_cost"]], errors="coerce").fillna(0.0)
         else:
-            df["_UnitCost"] = 0.0
+            df[self.COL_UNITCOST] = 0.0
 
         # Datetime
         if mapping["date"] and mapping["date"] in df:
@@ -475,46 +513,46 @@ class IrisReportEngine:
         except Exception:
             pass
 
-        df["_datetime"] = dt_series
-        df = df.dropna(subset=["_datetime"]).copy()
+        df[self.COL_DT] = dt_series
+        df = df.dropna(subset=[self.COL_DT]).copy()
 
         # Canonical dims
-        df["_Invoice"] = df[mapping["invoice"]] if mapping["invoice"] and mapping["invoice"] in df else None
-        df["_Product"] = df[mapping["product"]] if mapping["product"] and mapping["product"] in df else None
-        df["_Teller"]  = df[mapping["teller"]]  if mapping["teller"]  and mapping["teller"]  in df else None
-        df["_TxnType"] = (df[mapping["txn_type"]].astype(str).str.lower()
-                          if mapping["txn_type"] and mapping["txn_type"] in df else df.get("Transaction_Type", "").astype(str).str.lower())
-        df["_Branch"]  = df.get("Branch")
-        df["_Customer"] = df.get("Customer_Reference")
-
-        # Sales filter: keep explicit sales OR positive amounts
+        df[self.COL_INVOICE]  = df[mapping["invoice"]] if mapping["invoice"] and mapping["invoice"] in df else None
+        df[self.COL_PRODUCT]  = df[mapping["product"]] if mapping["product"] and mapping["product"] in df else None
+        df[self.COL_TELLER]   = df[mapping["teller"]]  if mapping["teller"]  and mapping["teller"]  in df else None
+        df[self.COL_TXNTYPE]  = (df[mapping["txn_type"]].astype(str).str.lower()
+                                 if mapping["txn_type"] and mapping["txn_type"] in df
+                                 else df.get("Transaction_Type", "").astype(str).str.lower())
+        df[self.COL_BRANCH]   = df.get("Branch")
+        df[self.COL_CUSTOMER] = df.get("Customer_Reference")
+
+        # Sales filter: keep explicit sales OR Transaction_Type_ID 21 OR positive amounts
+        txid_series = df.get("Transaction_Type_ID")
         sales_mask = (
-            df["_TxnType"].isin(["sale", "sales", "invoice"]) |
-            df.get("Transaction_Type_ID", pd.Series(dtype=float)).isin([21])
+            df[self.COL_TXNTYPE].isin(["sale", "sales", "invoice"]) |
+            (pd.Series(False, index=df.index) if txid_series is None else txid_series.isin([21])) |
+            (df[self.COL_AMOUNT] > 0)
         )
         working = df[sales_mask].copy()
-        if working["_Amount"].isna().all():
-            working = working.copy()
-        # Remove clearly non-sale placeholder SKUs from product analytics later using params["blocked_products"]
 
         # Derive measures
-        working["_Revenue"]     = working["_Amount"].fillna(0.0)
-        working["_COGS"]        = (working["_UnitCost"] * working["_Units"]).fillna(0.0)
-        working["_GrossProfit"] = (working["_Revenue"] - working["_COGS"]).fillna(0.0)
-        working["_Hour"]        = working["_datetime"].dt.hour
-        working["_DOW"]         = working["_datetime"].dt.day_name()
-        working["_DOW_idx"]     = working["_datetime"].dt.dayofweek  # 0=Mon .. 6=Sun
+        working[self.COL_REVENUE] = working[self.COL_AMOUNT].fillna(0.0)
+        working[self.COL_COGS]    = (working[self.COL_UNITCOST] * working[self.COL_UNITS]).fillna(0.0)
+        working[self.COL_GP]      = (working[self.COL_REVENUE] - working[self.COL_COGS]).fillna(0.0)
+        working[self.COL_HOUR]    = working[self.COL_DT].dt.hour
+        working[self.COL_DOW]     = working[self.COL_DT].dt.day_name()
+        working[self.COL_DOWI]    = working[self.COL_DT].dt.dayofweek  # 0=Mon .. 6=Sun
 
         # Deduplicate exact duplicate sale lines
         before = len(working)
-        dedupe_keys = ["Transaction_ID", "_Invoice", "_Product", "_Units", "_Amount", "_datetime"]
+        dedupe_keys = ["Transaction_ID", self.COL_INVOICE, self.COL_PRODUCT, self.COL_UNITS, self.COL_AMOUNT, self.COL_DT]
         existing_keys = [k for k in dedupe_keys if k in working.columns]
         if existing_keys:
             working = working.drop_duplicates(subset=existing_keys)
         duplicates_dropped = before - len(working)
 
-        # Drop zero-rows if both revenue and cost are zero to avoid noise
-        working = working[(working["_Revenue"].abs() > 0) | (working["_COGS"].abs() > 0)]
+        # Drop zero rows if both revenue and cost are zero
+        working = working[(working[self.COL_REVENUE].abs() > 0) | (working[self.COL_COGS].abs() > 0)]
 
         emit_kpi_debug(self.profile_id, "prepared_counts", {
             "raw_rows": int(len(self.raw)),
@@ -527,7 +565,6 @@ class IrisReportEngine:
         return working
 
     def _get_primary_currency(self) -> str:
-        candidates = ["USD", "ZAR", "ZWL", "EUR", "GBP"]
         try:
             mapping = ColumnResolver.map(self.raw)
             if mapping["currency"] and mapping["currency"] in self.raw:
@@ -551,8 +588,8 @@ class IrisReportEngine:
         start_prev = start_cur - pd.Timedelta(days=7)
         end_prev = start_cur - pd.Timedelta(seconds=1)
 
-        current_df = self.df[(self.df["_datetime"] >= start_cur) & (self.df["_datetime"] <= end_cur)]
-        previous_df = self.df[(self.df["_datetime"] >= start_prev) & (self.df["_datetime"] <= end_prev)]
+        current_df = self.df[(self.df[self.COL_DT] >= start_cur) & (self.df[self.COL_DT] <= end_cur)]
+        previous_df = self.df[(self.df[self.COL_DT] >= start_prev) & (self.df[self.COL_DT] <= end_prev)]
 
         meta = {
             "period_label": "This Week vs. Last Week",
@@ -573,17 +610,17 @@ class IrisReportEngine:
         return f"{((cur - prev) / prev) * 100:+.1f}%"
 
     def _headline(self, cur_df: pd.DataFrame, prev_df: pd.DataFrame) -> Dict[str, Any]:
-        cur_rev = float(cur_df["_Revenue"].sum()) if not cur_df.empty else 0.0
-        prev_rev = float(prev_df["_Revenue"].sum()) if not prev_df.empty else 0.0
-        cur_gp  = float(cur_df["_GrossProfit"].sum()) if not cur_df.empty else 0.0
-        prev_gp = float(prev_df["_GrossProfit"].sum()) if not prev_df.empty else 0.0
+        cur_rev = float(cur_df[self.COL_REVENUE].sum()) if not cur_df.empty else 0.0
+        prev_rev = float(prev_df[self.COL_REVENUE].sum()) if not prev_df.empty else 0.0
+        cur_gp  = float(cur_df[self.COL_GP].sum()) if not cur_df.empty else 0.0
+        prev_gp = float(prev_df[self.COL_GP].sum()) if not prev_df.empty else 0.0
 
-        if "_Invoice" in cur_df.columns and cur_df["_Invoice"].notna().any():
-            tx_now = int(cur_df["_Invoice"].nunique())
+        if self._has(cur_df, self.COL_INVOICE) and cur_df[self.COL_INVOICE].notna().any():
+            tx_now = int(cur_df[self.COL_INVOICE].nunique())
         else:
             tx_now = int(len(cur_df))
-        if "_Invoice" in prev_df.columns and prev_df["_Invoice"].notna().any():
-            tx_prev = int(prev_df["_Invoice"].nunique())
+        if self._has(prev_df, self.COL_INVOICE) and prev_df[self.COL_INVOICE].notna().any():
+            tx_prev = int(prev_df[self.COL_INVOICE].nunique())
         else:
             tx_prev = int(len(prev_df))
 
@@ -607,39 +644,31 @@ class IrisReportEngine:
     def _build_product_aggregates(self, cur_df: pd.DataFrame) -> pd.DataFrame:
         if cur_df.empty:
             return pd.DataFrame(columns=[
-                "_Product","revenue","units","cogs","gross_profit","margin_pct","avg_selling_price","avg_unit_cost","tx_count"
+                self.COL_PRODUCT,"revenue","units","cogs","gross_profit","margin_pct",
+                "avg_selling_price","avg_unit_cost","tx_count"
             ])
 
         df = cur_df.copy()
         # Exclude blocked products for leaderboards/affinity, but keep them in totals if needed
         if self.params["blocked_products"]:
-            df = df[~df["_Product"].astype(str).str.strip().isin(self.params["blocked_products"])]
+            df = df[~df[self.COL_PRODUCT].astype(str).str.strip().isin(self.params["blocked_products"])]
 
         # Tx count via invoice nunique if available
-        if "_Invoice" in df.columns and df["_Invoice"].notna().any():
-            g = df.groupby("_Product", dropna=False).agg(
-                revenue=("_Revenue","sum"),
-                units=("_Units","sum"),
-                cogs=("_COGS","sum"),
-                gp=("_GrossProfit","sum"),
-                tx=(" _Invoice","nunique")  # typo trap; fix next line
-            )
-        # fix groupby with invoice nunique
-        if "_Invoice" in df.columns and df["_Invoice"].notna().any():
-            g = df.groupby("_Product", dropna=False).agg(
-                revenue=("_Revenue","sum"),
-                units=("_Units","sum"),
-                cogs=("_COGS","sum"),
-                gp=("_GrossProfit","sum"),
-                tx=("_Invoice","nunique")
+        if self._has(df, self.COL_INVOICE) and df[self.COL_INVOICE].notna().any():
+            g = df.groupby(self.COL_PRODUCT, dropna=False).agg(
+                revenue=(self.COL_REVENUE,"sum"),
+                units=(self.COL_UNITS,"sum"),
+                cogs=(self.COL_COGS,"sum"),
+                gp=(self.COL_GP,"sum"),
+                tx=(self.COL_INVOICE,"nunique")
             )
         else:
-            g = df.groupby("_Product", dropna=False).agg(
-                revenue=("_Revenue","sum"),
-                units=("_Units","sum"),
-                cogs=("_COGS","sum"),
-                gp=("_GrossProfit","sum"),
-                tx=("_Product","size")
+            g = df.groupby(self.COL_PRODUCT, dropna=False).agg(
+                revenue=(self.COL_REVENUE,"sum"),
+                units=(self.COL_UNITS,"sum"),
+                cogs=(self.COL_COGS,"sum"),
+                gp=(self.COL_GP,"sum"),
+                tx=(self.COL_PRODUCT,"size")
             )
 
         g = g.rename(columns={"gp":"gross_profit", "tx":"tx_count"}).reset_index()
@@ -649,24 +678,21 @@ class IrisReportEngine:
         g["avg_selling_price"] = np.where(g["units"] > 0, g["revenue"] / g["units"], np.nan)
         g["avg_unit_cost"] = np.where(g["units"] > 0, g["cogs"] / g["units"], np.nan)
 
-        # velocity (units/day) needs window length
-        # Set later when we know the time window length; store raw fields for now
         return g
 
     def _build_basket_table(self, cur_df: pd.DataFrame) -> pd.DataFrame:
-        if cur_df.empty:
-            return pd.DataFrame(columns=["_Invoice","basket_revenue","basket_gp","basket_items","_datetime_max"])
-        # per invoice sums
-        b = cur_df.groupby("_Invoice", dropna=False).agg(
-            basket_revenue=("_Revenue","sum"),
-            basket_gp=("_GrossProfit","sum"),
-            basket_items=("_Units","sum"),
-            _datetime_max=("_datetime","max"),
+        if cur_df.empty or not self._has(cur_df, self.COL_INVOICE):
+            return pd.DataFrame(columns=[self.COL_INVOICE,"basket_revenue","basket_gp","basket_items","_datetime_max"])
+        b = cur_df.groupby(self.COL_INVOICE, dropna=False).agg(
+            basket_revenue=(self.COL_REVENUE,"sum"),
+            basket_gp=(self.COL_GP,"sum"),
+            basket_items=(self.COL_UNITS,"sum"),
+            _datetime_max=(self.COL_DT,"max"),
         ).reset_index()
         return b
 
     def _basket_kpis(self, basket_df: pd.DataFrame) -> Dict[str, Any]:
-        if basket_df.empty:
+        if basket_df is None or basket_df.empty:
             return {
                 "avg_items_per_basket": "N/A",
                 "avg_gross_profit_per_basket": "N/A",
@@ -677,10 +703,9 @@ class IrisReportEngine:
         avg_items = float(basket_df["basket_items"].mean())
         avg_gp = float(basket_df["basket_gp"].mean())
         median_value = float(basket_df["basket_revenue"].median())
-        # size histogram
         sizes = basket_df["basket_items"].fillna(0)
         bins = {
-            "1": int(((sizes == 1).sum())),
+            "1": int((sizes == 1).sum()),
             "2-3": int(((sizes >= 2) & (sizes <= 3)).sum()),
             "4-5": int(((sizes >= 4) & (sizes <= 5)).sum()),
             "6_plus": int((sizes >= 6).sum()),
@@ -694,37 +719,32 @@ class IrisReportEngine:
 
     def _affinity_pairs(self, cur_df: pd.DataFrame, basket_df: pd.DataFrame) -> Dict[str, Any]:
         # Build unique product sets per invoice, count pairs
-        if cur_df.empty or basket_df.empty or "_Product" not in cur_df.columns:
+        if cur_df.empty or basket_df.empty or not self._has(cur_df, self.COL_PRODUCT) or not self._has(cur_df, self.COL_INVOICE):
             return {"params": self._affinity_params(), "top_pairs": []}
 
-        # Per-basket unique product set (exclude null/blocked)
-        tmp = cur_df[["_Invoice","_Product"]].dropna()
+        tmp = cur_df[[self.COL_INVOICE, self.COL_PRODUCT]].dropna()
         if tmp.empty:
             return {"params": self._affinity_params(), "top_pairs": []}
 
         blocked = set(self.params.get("blocked_products", []) or [])
-        tmp = tmp[~tmp["_Product"].astype(str).str.strip().isin(blocked)]
+        tmp = tmp[~tmp[self.COL_PRODUCT].astype(str).str.strip().isin(blocked)]
         if tmp.empty:
             return {"params": self._affinity_params(), "top_pairs": []}
 
-        products_per_invoice = tmp.groupby("_Invoice")["_Product"].agg(lambda s: sorted(set(map(str, s)))).reset_index()
+        products_per_invoice = tmp.groupby(self.COL_INVOICE)[self.COL_PRODUCT].agg(lambda s: sorted(set(map(str, s)))).reset_index()
         total_baskets = int(len(products_per_invoice))
         if total_baskets == 0:
             return {"params": self._affinity_params(), "top_pairs": []}
 
-        # Limit explosion: optionally cap to top-N frequent products first
-        # Count single supports
         from collections import Counter
         single_counter = Counter()
-        for prods in products_per_invoice["_Product"]:
+        for prods in products_per_invoice[self.COL_PRODUCT]:
             single_counter.update(prods)
 
-        # Pair counting
         pair_counter = Counter()
-        for prods in products_per_invoice["_Product"]:
+        for prods in products_per_invoice[self.COL_PRODUCT]:
             if len(prods) < 2:
                 continue
-            # 2-combinations
             for i in range(len(prods)):
                 for j in range(i+1, len(prods)):
                     a, b = prods[i], prods[j]
@@ -736,11 +756,9 @@ class IrisReportEngine:
         top_k = int(self.params["top_k"])
 
         rows = []
-        # Average pair revenue across baskets containing both (optional; approximate via filtering once)
-        inv_with_products = cur_df.groupby("_Invoice")["_Product"].apply(lambda s: set(map(str, s.dropna())))
-
-        # Precompute basket revenue by invoice for avg pair revenue
-        rev_by_inv = cur_df.groupby("_Invoice")["_Revenue"].sum()
+        # Average pair revenue across baskets containing both (optional; approximate)
+        inv_with_products = cur_df.groupby(self.COL_INVOICE)[self.COL_PRODUCT].apply(lambda s: set(map(str, s.dropna())))
+        rev_by_inv = cur_df.groupby(self.COL_INVOICE)[self.COL_REVENUE].sum()
 
         for (a, b), ab_count in pair_counter.items():
             if ab_count < min_support_baskets:
@@ -755,7 +773,6 @@ class IrisReportEngine:
             if not np.isfinite(lift) or lift < min_lift:
                 continue
 
-            # avg pair revenue over baskets that include both
             inv_mask = inv_with_products.apply(lambda s: (a in s) and (b in s))
             pair_invoices = inv_mask[inv_mask].index
             avg_pair_revenue = float(rev_by_inv.loc[pair_invoices].mean()) if len(pair_invoices) else np.nan
@@ -791,34 +808,29 @@ class IrisReportEngine:
                 "dow_series": [],
                 "profit_heatmap_7x24": []
             }
-        # Hourly
-        gh = cur_df.groupby("_Hour", dropna=False).agg(
-            revenue=("_Revenue","sum"),
-            gross_profit=("_GrossProfit","sum")
+        gh = cur_df.groupby(self.COL_HOUR, dropna=False).agg(
+            revenue=(self.COL_REVENUE,"sum"),
+            gross_profit=(self.COL_GP,"sum")
         ).reset_index()
-        best_hour_idx = int(gh.loc[gh["gross_profit"].idxmax(), "_Hour"]) if not gh.empty else None
+        best_hour_idx = int(gh.loc[gh["gross_profit"].idxmax(), self.COL_HOUR]) if not gh.empty else None
         best_hour_gp  = float(gh["gross_profit"].max()) if not gh.empty else None
 
-        # DOW
-        gd = cur_df.groupby("_DOW", dropna=False).agg(
-            revenue=("_Revenue","sum"),
-            gross_profit=("_GrossProfit","sum")
+        gd = cur_df.groupby(self.COL_DOW, dropna=False).agg(
+            revenue=(self.COL_REVENUE,"sum"),
+            gross_profit=(self.COL_GP,"sum")
         ).reset_index()
-        # enforce Mon..Sun order using _DOW_idx
-        order_map = cur_df.groupby("_DOW")["_DOW_idx"].max().to_dict()
-        gd["__ord"] = gd["_DOW"].map(order_map)
+        order_map = cur_df.groupby(self.COL_DOW)[self.COL_DOWI].max().to_dict()
+        gd["__ord"] = gd[self.COL_DOW].map(order_map)
         gd = gd.sort_values("__ord", kind="stable")
         best_day_row = gd.loc[gd["gross_profit"].idxmax()] if not gd.empty else None
-        best_day = {"day": str(best_day_row["_DOW"]), "gross_profit": float(best_day_row["gross_profit"])} if best_day_row is not None else None
+        best_day = {"day": str(best_day_row[self.COL_DOW]), "gross_profit": float(best_day_row["gross_profit"])} if best_day_row is not None else None
 
-        # Heatmap (7x24 by _DOW_idx then _Hour)
-        m = cur_df.groupby(["_DOW_idx","_Hour"], dropna=False)["_GrossProfit"].sum().unstack(fill_value=0)
-        # ensure full 7x24
+        m = cur_df.groupby([self.COL_DOWI, self.COL_HOUR], dropna=False)[self.COL_GP].sum().unstack(fill_value=0)
         m = m.reindex(index=range(0,7), columns=range(0,24), fill_value=0)
         heatmap = [[float(x) for x in row] for row in m.values.tolist()]
 
-        hourly_series = gh.rename(columns={"_Hour":"hour"}).to_dict(orient="records")
-        dow_series = gd[["_DOW","revenue","gross_profit"]].rename(columns={"_DOW":"day"}).to_dict(orient="records")
+        hourly_series = gh.rename(columns={self.COL_HOUR:"hour"}).to_dict(orient="records")
+        dow_series = gd[[self.COL_DOW,"revenue","gross_profit"]].rename(columns={self.COL_DOW:"day"}).to_dict(orient="records")
 
         return {
             "best_hour_by_profit": {"hour": best_hour_idx, "gross_profit": round(best_hour_gp, 2)} if best_hour_idx is not None else None,
@@ -829,24 +841,26 @@ class IrisReportEngine:
         }
 
     def _customer_value(self, cur_df: pd.DataFrame, basket_df: pd.DataFrame) -> Dict[str, Any]:
-        if cur_df.empty or "_Customer" not in cur_df.columns:
+        if cur_df.empty or not self._has(cur_df, self.COL_CUSTOMER):
             return {
                 "params": {"rfm_window_days": int(self.params["rfm_window_days"]), "retention_factor": float(self.params["retention_factor"]), "vip_count": 20},
                 "leaderboards": {"top_customers_by_gp": [], "at_risk": [], "new_customers": []},
                 "rfm_summary": {"unique_customers": 0, "median_recency_days": None, "median_orders": None, "median_gp": None}
             }
         df = cur_df.copy()
-        # Build per-customer aggregates
-        last_date = df.groupby("_Customer")["_datetime"].max()
-        orders = (df.dropna(subset=["_Invoice"])
-                    .groupby("_Customer")["_Invoice"].nunique())
-        revenue = df.groupby("_Customer")["_Revenue"].sum()
-        gp = df.groupby("_Customer")["_GrossProfit"].sum()
+
+        last_date = df.groupby(self.COL_CUSTOMER)[self.COL_DT].max()
+        if self._has(df, self.COL_INVOICE):
+            orders = df.dropna(subset=[self.COL_INVOICE]).groupby(self.COL_CUSTOMER)[self.COL_INVOICE].nunique()
+        else:
+            orders = df.groupby(self.COL_CUSTOMER).size()
+        revenue = df.groupby(self.COL_CUSTOMER)[self.COL_REVENUE].sum()
+        gp = df.groupby(self.COL_CUSTOMER)[self.COL_GP].sum()
 
         # Avg basket value per customer (from their invoices)
-        if not basket_df.empty and "_Invoice" in df.columns:
-            inv_to_rev = basket_df.set_index("_Invoice")["basket_revenue"]
-            cust_invoices = df.dropna(subset=["_Invoice"]).groupby("_Customer")["_Invoice"].agg(lambda x: sorted(set(x)))
+        if not basket_df.empty and self._has(df, self.COL_INVOICE):
+            inv_to_rev = basket_df.set_index(self.COL_INVOICE)["basket_revenue"]
+            cust_invoices = df.dropna(subset=[self.COL_INVOICE]).groupby(self.COL_CUSTOMER)[self.COL_INVOICE].agg(lambda x: sorted(set(x)))
             avg_basket_val = {}
             for cust, invs in cust_invoices.items():
                 vals = inv_to_rev.reindex(invs).dropna()
@@ -867,24 +881,20 @@ class IrisReportEngine:
             "avg_basket_value": avg_basket.reindex(last_date.index).values
         }).fillna({"avg_basket_value": np.nan})
 
-        # Leaderboards
         vip = rfm.sort_values(["gp","orders","revenue"], ascending=[False, False, False]).head(20)
-        # At-risk: top quartile gp but recency > 30 days (tunable)
         if len(rfm):
             gp_q3 = rfm["gp"].quantile(0.75)
             at_risk = rfm[(rfm["gp"] >= gp_q3) & (rfm["recency_days"] > 30)].sort_values(["gp","recency_days"], ascending=[False, False]).head(20)
         else:
             at_risk = rfm.head(0)
-        # New customers: first seen within current window (approx via last_date inside window and orders==1)
-        # (More precise would need a historical first_seen; we infer using current window)
         new_customers = rfm[(rfm["orders"] == 1) & (rfm["recency_days"] <= 7)].sort_values("gp", ascending=False).head(20)
 
         out = {
             "params": {"rfm_window_days": int(self.params["rfm_window_days"]), "retention_factor": float(self.params["retention_factor"]), "vip_count": 20},
             "leaderboards": {
-                "top_customers_by_gp": rfm_to_list(vip),
-                "at_risk": rfm_to_list(at_risk),
-                "new_customers": rfm_to_list(new_customers)
+                "top_customers_by_gp": self._rfm_to_list(vip),
+                "at_risk": self._rfm_to_list(at_risk),
+                "new_customers": self._rfm_to_list(new_customers)
             },
             "rfm_summary": {
                 "unique_customers": int(rfm["customer"].nunique()),
@@ -905,21 +915,17 @@ class IrisReportEngine:
         start_cur, end_cur = current_bounds
         days = max(1.0, (end_cur - start_cur).total_seconds() / 86400.0)
 
-        # velocity from product_agg
-        pa = product_agg.copy()
+        pa = (product_agg or pd.DataFrame()).copy()
         if pa.empty:
             return {"status": "no_stock_data", "products": [], "alerts": {"low_stock": [], "stockout_risk": [], "dead_stock": []}}
 
         pa["units_per_day"] = pa["units"] / days
 
-        # merge stock feed on product
         sf = self.stock_feed.copy()
-        # Normalize join keys
         sf["product_key"] = sf.get("product", sf.get("Product", "")).astype(str).str.strip()
-        pa["product_key"] = pa["_Product"].astype(str).str.strip()
+        pa["product_key"] = pa[self.COL_PRODUCT].astype(str).str.strip()
         merged = pa.merge(sf, on="product_key", how="right", suffixes=("", "_stock"))
 
-        # If a product exists in stock but didn’t sell in window, units_per_day may be NaN→0
         merged["units_per_day"] = merged["units_per_day"].fillna(0.0)
         merged["stock_on_hand"] = pd.to_numeric(merged.get("stock_on_hand", np.nan), errors="coerce")
         merged["reorder_point"] = pd.to_numeric(merged.get("reorder_point", np.nan), errors="coerce")
@@ -930,7 +936,7 @@ class IrisReportEngine:
         def status_row(r):
             if pd.isna(r.get("stock_on_hand")):
                 return "unknown"
-            if r["stock_on_hand"] <= 0:
+            if (r["stock_on_hand"] or 0) <= 0:
                 return "stockout"
             if pd.notna(r.get("reorder_point")) and r["stock_on_hand"] <= r["reorder_point"]:
                 return "low"
@@ -942,11 +948,10 @@ class IrisReportEngine:
 
         merged["status"] = merged.apply(status_row, axis=1)
 
-        products_out = []
-        low_stock, stockout_risk, dead_stock = [], [], []
+        products_out, low_stock, stockout_risk, dead_stock = [], [], [], []
         for _, r in merged.iterrows():
             rec = {
-                "product": str(r.get("_Product") or r.get("product_key")),
+                "product": str(r.get(self.COL_PRODUCT) or r.get("product_key")),
                 "stock_on_hand": float(r["stock_on_hand"]) if pd.notna(r["stock_on_hand"]) else None,
                 "reorder_point": float(r["reorder_point"]) if pd.notna(r["reorder_point"]) else None,
                 "lead_time_days": float(r["lead_time_days"]) if pd.notna(r["lead_time_days"]) else None,
@@ -976,7 +981,6 @@ class IrisReportEngine:
         if self.cash_float_feed.empty:
             return {"status": "no_cash_data"}
 
-        # We expect cash_float_feed rows with branch, date (YYYY-MM-DD), opening_float, closing_float, drops, petty_cash, declared_cash
         cf = self.cash_float_feed.copy()
         out_days = []
         high_var_days = 0
@@ -986,16 +990,15 @@ class IrisReportEngine:
             cash_sales = pd.DataFrame(columns=["branch","date","cash_sales"])
         else:
             df = cur_df.copy()
-            df["date"] = df["_datetime"].dt.strftime("%Y-%m-%d")
+            df["date"] = df[self.COL_DT].dt.strftime("%Y-%m-%d")
             df["is_cash"] = (df.get("Money_Type","").astype(str).str.lower() == "cash")
-            cash_sales = df[df["is_cash"]].groupby(["_Branch","date"])["_Revenue"].sum().reset_index()
-            cash_sales = cash_sales.rename(columns={"_Branch":"branch","_Revenue":"cash_sales"})
+            cash_sales = df[df["is_cash"]].groupby([self.COL_BRANCH,"date"])[self.COL_REVENUE].sum().reset_index()
+            cash_sales = cash_sales.rename(columns={self.COL_BRANCH:"branch", self.COL_REVENUE:"cash_sales"})
 
         cf["date"] = cf["date"].astype(str).str[:10]
         merged = cf.merge(cash_sales, on=["branch","date"], how="left")
         merged["cash_sales"] = merged["cash_sales"].fillna(0.0)
 
-        # Expected Cash = Opening + CashSales – Drops – PettyCash – Closing
         for _, r in merged.iterrows():
             opening = float(r.get("opening_float") or 0.0)
             closing = float(r.get("closing_float") or 0.0)
@@ -1034,119 +1037,110 @@ class IrisReportEngine:
     # ------------------------- branch analytics -------------------------
 
     def _per_branch_blocks(self, cur_df: pd.DataFrame, previous_df: pd.DataFrame, current_bounds: Tuple[pd.Timestamp,pd.Timestamp]) -> Dict[str, Any]:
-        if cur_df.empty or "_Branch" not in cur_df.columns:
+        if cur_df.empty or not self._has(cur_df, self.COL_BRANCH):
             return {"params": self._branch_params(), "per_branch": {}, "cross_branch": {}}
 
         per_branch = {}
-        branches = sorted(map(str, cur_df["_Branch"].dropna().unique().tolist()))
+        branches = sorted(map(str, cur_df[self.COL_BRANCH].dropna().unique().tolist()))
         start_cur, end_cur = current_bounds
         days = max(1.0, (end_cur - start_cur).total_seconds() / 86400.0)
 
         branch_summary_rows = []
 
         for br in branches:
-            d = cur_df[cur_df["_Branch"] == br]
-            if d.empty:
-                continue
-
-            # headline-like
-            revenue = float(d["_Revenue"].sum())
-            cogs    = float(d["_COGS"].sum())
-            gp      = float(d["_GrossProfit"].sum())
-            margin_pct = (gp / revenue) if revenue > 0 else None
-            tx = int(d["_Invoice"].nunique()) if "_Invoice" in d.columns and d["_Invoice"].notna().any() else int(len(d))
-            items = float(d["_Units"].sum())
-
-            # baskets
-            basket_df = self._build_basket_table(d)
-            basket_kpis = self._basket_kpis(basket_df)
-
-            # temporal
-            temporal = self._temporal_patterns(d)
-
-            # product leaderboards
-            pagg = self._build_product_aggregates(d)
-            if not pagg.empty:
-                pagg["units_per_day"] = pagg["units"] / days
-                product_lb = self._product_leaderboards(pagg)
-            else:
-                product_lb = self._empty_product_leaderboards()
-
-            # affinity
-            affinity = self._affinity_pairs(d, basket_df)
-
-            # customers
-            customers = self._customer_value(d, basket_df)
-
-            # cash recon slice
-            cash_recon = self._cash_recon_block(d)
-
-            per_branch[br] = {
-                "kpis": {
-                    "revenue": round(revenue, 2),
-                    "cogs": round(cogs, 2),
-                    "gross_profit": round(gp, 2),
-                    "gp_margin_pct": float(round(margin_pct, 4)) if margin_pct is not None else None,
-                    "transactions": tx,
-                    "items_sold": round(items, 2),
-                    "avg_basket_value": basket_kpis.get("median_basket_value"),
-                    "avg_items_per_basket": basket_kpis.get("avg_items_per_basket"),
-                    "avg_gp_per_basket": basket_kpis.get("avg_gross_profit_per_basket"),
-                },
-                "temporal": temporal,
-                "products": product_lb,
-                "affinity": affinity,
-                "customer_value": customers,
-                "cash_recon": cash_recon,
-                "data_quality": {
-                    "duplicates_dropped": self._prepared_dupes_dropped,
-                    "non_sale_rows_excluded": self._non_sale_excluded,
-                    "currency_mixed": False  # set if you add multi-currency detection
+            try:
+                d = cur_df[cur_df[self.COL_BRANCH] == br]
+                if d.empty:
+                    continue
+
+                revenue = float(d[self.COL_REVENUE].sum())
+                cogs    = float(d[self.COL_COGS].sum())
+                gp      = float(d[self.COL_GP].sum())
+                margin_pct = (gp / revenue) if revenue > 0 else None
+                tx = int(d[self.COL_INVOICE].nunique()) if self._has(d, self.COL_INVOICE) and d[self.COL_INVOICE].notna().any() else int(len(d))
+                items = float(d[self.COL_UNITS].sum())
+
+                basket_df = self._build_basket_table(d)
+                basket_kpis = self._basket_kpis(basket_df)
+                temporal = self._temporal_patterns(d)
+
+                pagg = self._build_product_aggregates(d)
+                if not pagg.empty:
+                    pagg["units_per_day"] = pagg["units"] / days
+                    product_lb = self._product_leaderboards(pagg)
+                else:
+                    product_lb = self._empty_product_leaderboards()
+
+                affinity = self._affinity_pairs(d, basket_df)
+                customers = self._customer_value(d, basket_df)
+                cash_recon = self._cash_recon_block(d)
+
+                per_branch[br] = {
+                    "kpis": {
+                        "revenue": round(revenue, 2),
+                        "cogs": round(cogs, 2),
+                        "gross_profit": round(gp, 2),
+                        "gp_margin_pct": float(round(margin_pct, 4)) if margin_pct is not None else None,
+                        "transactions": tx,
+                        "items_sold": round(items, 2),
+                        "avg_basket_value": basket_kpis.get("median_basket_value"),
+                        "avg_items_per_basket": basket_kpis.get("avg_items_per_basket"),
+                        "avg_gp_per_basket": basket_kpis.get("avg_gross_profit_per_basket"),
+                    },
+                    "temporal": temporal,
+                    "products": product_lb,
+                    "affinity": affinity,
+                    "customer_value": customers,
+                    "cash_recon": cash_recon,
+                    "data_quality": {
+                        "duplicates_dropped": self._prepared_dupes_dropped,
+                        "non_sale_rows_excluded": self._non_sale_excluded,
+                        "currency_mixed": False
+                    }
                 }
-            }
 
-            branch_summary_rows.append({"branch": br, "revenue": revenue, "gp": gp, "gp_margin_pct": margin_pct or 0.0})
+                branch_summary_rows.append({"branch": br, "revenue": revenue, "gp": gp, "gp_margin_pct": margin_pct or 0.0})
+            except Exception as e:
+                emit_kpi_debug(self.profile_id, "branch_block_error", {"branch": br, "error": str(e)})
 
-        # cross-branch comparisons
         cross = {}
         if branch_summary_rows:
-            bs = pd.DataFrame(branch_summary_rows)
-            cross["rankings"] = {
-                "by_revenue": bs.sort_values("revenue", ascending=False)[["branch","revenue"]].to_dict(orient="records"),
-                "by_gp_margin_pct": bs.sort_values("gp_margin_pct", ascending=False)[["branch","gp_margin_pct"]].to_dict(orient="records"),
-            }
-            cross["spread"] = {
-                "gp_margin_pct_max": float(bs["gp_margin_pct"].max()) if len(bs) else None,
-                "gp_margin_pct_min": float(bs["gp_margin_pct"].min()) if len(bs) else None,
-                "gap_pct_points": float((bs["gp_margin_pct"].max() - bs["gp_margin_pct"].min())) if len(bs) else None,
-            }
-            # revenue share & HHI
-            tot_rev = float(bs["revenue"].sum())
-            shares = []
-            hhi = 0.0
-            for _, r in bs.iterrows():
-                sh = (r["revenue"] / tot_rev) if tot_rev > 0 else 0.0
-                shares.append({"branch": r["branch"], "share": float(round(sh, 6))})
-                hhi += sh*sh
-            cross["concentration"] = {"share_by_branch": shares, "hhi_revenue": float(round(hhi, 6))}
-            # week-over-week deltas per branch (best-effort: compute previous per-branch)
-            if not previous_df.empty:
-                prev_g = previous_df.groupby("_Branch").agg(
-                    revenue=("_Revenue","sum"),
-                    gp=("_GrossProfit","sum")
-                ).reset_index().rename(columns={"_Branch":"branch"})
-                cur_g = pd.DataFrame(branch_summary_rows)
-                cur_g = cur_g.rename(columns={"branch":"branch"})
-                m = cur_g.merge(prev_g, on="branch", suffixes=("_cur","_prev"), how="left").fillna(0.0)
-                wow_rows = []
-                for _, r in m.iterrows():
-                    wow_rows.append({
-                        "branch": r["branch"],
-                        "revenue_wow": float(((r["revenue_cur"] - r["revenue_prev"]) / r["revenue_prev"])*100) if r["revenue_prev"]>0 else (100.0 if r["revenue_cur"]>0 else 0.0),
-                        "gp_wow": float(((r["gp_cur"] - r["gp_prev"]) / r["gp_prev"])*100) if r["gp_prev"]>0 else (100.0 if r["gp_cur"]>0 else 0.0),
-                        "avg_basket_wow": None  # compute if you persist prev basket median
-                    })
-                cross["trend_wow"] = wow_rows
+            try:
+                bs = pd.DataFrame(branch_summary_rows)
+                cross["rankings"] = {
+                    "by_revenue": bs.sort_values("revenue", ascending=False)[["branch","revenue"]].to_dict(orient="records"),
+                    "by_gp_margin_pct": bs.sort_values("gp_margin_pct", ascending=False)[["branch","gp_margin_pct"]].to_dict(orient="records"),
+                }
+                cross["spread"] = {
+                    "gp_margin_pct_max": float(bs["gp_margin_pct"].max()) if len(bs) else None,
+                    "gp_margin_pct_min": float(bs["gp_margin_pct"].min()) if len(bs) else None,
+                    "gap_pct_points": float((bs["gp_margin_pct"].max() - bs["gp_margin_pct"].min())) if len(bs) else None,
+                }
+                tot_rev = float(bs["revenue"].sum())
+                shares, hhi = [], 0.0
+                for _, r in bs.iterrows():
+                    sh = (r["revenue"] / tot_rev) if tot_rev > 0 else 0.0
+                    shares.append({"branch": r["branch"], "share": float(round(sh, 6))})
+                    hhi += sh*sh
+                cross["concentration"] = {"share_by_branch": shares, "hhi_revenue": float(round(hhi, 6))}
+                if not previous_df.empty and self._has(previous_df, self.COL_BRANCH):
+                    prev_g = previous_df.groupby(self.COL_BRANCH).agg(
+                        revenue=(self.COL_REVENUE,"sum"),
+                        gp=(self.COL_GP,"sum")
+                    ).reset_index().rename(columns={self.COL_BRANCH:"branch"})
+                    cur_g = pd.DataFrame(branch_summary_rows)
+                    m = cur_g.merge(prev_g, on="branch", suffixes=("_cur","_prev"), how="left").fillna(0.0)
+                    wow_rows = []
+                    for _, r in m.iterrows():
+                        wow_rows.append({
+                            "branch": r["branch"],
+                            "revenue_wow": float(((r["revenue_cur"] - r["revenue_prev"]) / r["revenue_prev"])*100) if r["revenue_prev"]>0 else (100.0 if r["revenue_cur"]>0 else 0.0),
+                            "gp_wow": float(((r["gp_cur"] - r["gp_prev"]) / r["gp_prev"])*100) if r["gp_prev"]>0 else (100.0 if r["gp_cur"]>0 else 0.0),
+                            "avg_basket_wow": None
+                        })
+                    cross["trend_wow"] = wow_rows
+            except Exception as e:
+                emit_kpi_debug(self.profile_id, "branch_cross_error", {"error": str(e)})
 
         return {"params": self._branch_params(), "per_branch": per_branch, "cross_branch": cross}
 
@@ -1163,7 +1157,6 @@ class IrisReportEngine:
 
     def _product_leaderboards(self, g: pd.DataFrame) -> Dict[str, Any]:
         top_k = int(self.params["top_k"])
-        # margin % floor
         g_marginpct = g.copy()
         g_marginpct = g_marginpct[
             (g_marginpct["revenue"] >= float(self.params["min_revenue_for_margin_pct"])) &
@@ -1176,7 +1169,7 @@ class IrisReportEngine:
             d = df.sort_values(col, ascending=asc).head(top_k)
             return [
                 {
-                    "product": str(r["_Product"]),
+                    "product": str(r[self.COL_PRODUCT]),
                     "revenue": round(float(r["revenue"]), 2),
                     "units": float(r["units"]),
                     "gross_profit": round(float(r["gross_profit"]), 2),
@@ -1219,7 +1212,6 @@ class IrisReportEngine:
                 "revenue_pareto_top20pct_share": 0.0,
                 "gini_revenue": 0.0
             }
-        # shares
         total_rev = float(g["revenue"].sum())
         total_units = float(g["units"].sum())
         rev_sorted = g.sort_values("revenue", ascending=False)["revenue"].values
@@ -1228,7 +1220,6 @@ class IrisReportEngine:
         share_top5_rev = (rev_sorted[:5].sum() / total_rev) if total_rev > 0 else 0.0
         share_top5_units = (units_sorted[:5].sum() / total_units) if total_units > 0 else 0.0
 
-        # Pareto top 20% products by count
         n = len(rev_sorted)
         if n == 0:
             pareto = 0.0
@@ -1236,11 +1227,9 @@ class IrisReportEngine:
             k = max(1, int(np.ceil(0.2 * n)))
             pareto = rev_sorted[:k].sum() / total_rev if total_rev > 0 else 0.0
 
-        # Gini on revenue
         if total_rev <= 0 or n == 0:
             gini = 0.0
         else:
-            # Gini for array x >=0: G = 1 - 2 * sum((n+1-i)*x_i) / (n * sum(x))
             x = np.sort(rev_sorted)  # ascending
             cum = np.cumsum(x)
             gini = 1.0 - 2.0 * np.sum(cum) / (n * np.sum(x)) + 1.0 / n
@@ -1264,82 +1253,119 @@ class IrisReportEngine:
             emit_kpi_debug(self.profile_id, "briefing", {"status": "no_current_period_data", **tfmeta})
             return {"Status": f"No sales data for the current period ({tfmeta.get('period_label', 'N/A')}).", "meta": tfmeta}
 
-        headline = self._headline(current_df, previous_df)
+        snapshot = {}
+        section_errors = {}
+
+        # Headline
+        try:
+            headline = self._headline(current_df, previous_df)
+            snapshot["Summary Period"] = tfmeta.get("period_label", "This Week vs. Last Week")
+            snapshot["Performance Snapshot (vs. Prior Period)"] = {
+                "Total Revenue": f"{headline['total_revenue_fmt']} ({headline['total_revenue_change']})",
+                "Gross Profit": f"{headline['gross_profit_fmt']} ({headline['gross_profit_change']})",
+                "Transactions": f"{headline['transactions_value']} ({headline['transactions_change']})",
+            }
+        except Exception as e:
+            section_errors["headline"] = str(e)
 
         # Basket & affinity
-        basket_df = self._build_basket_table(current_df)
-        basket_kpis = self._basket_kpis(basket_df)
-        affinity = self._affinity_pairs(current_df, basket_df)
+        try:
+            basket_df = self._build_basket_table(current_df)
+            snapshot["Basket Analysis"] = self._basket_kpis(basket_df)
+        except Exception as e:
+            section_errors["basket"] = str(e)
+            snapshot["Basket Analysis"] = {"avg_items_per_basket": "N/A", "avg_gross_profit_per_basket": "N/A", "median_basket_value": "N/A", "basket_size_distribution": {}, "low_sample": True}
+
+        try:
+            if 'basket_df' in locals():
+                snapshot["Product Affinity"] = self._affinity_pairs(current_df, basket_df)
+            else:
+                snapshot["Product Affinity"] = {"params": self._affinity_params(), "top_pairs": []}
+        except Exception as e:
+            section_errors["affinity"] = str(e)
+            snapshot["Product Affinity"] = {"params": self._affinity_params(), "top_pairs": []}
 
         # Temporal
-        temporal = self._temporal_patterns(current_df)
+        try:
+            snapshot["Temporal Patterns"] = self._temporal_patterns(current_df)
+        except Exception as e:
+            section_errors["temporal"] = str(e)
+            snapshot["Temporal Patterns"] = {"best_hour_by_profit": None, "best_day_by_profit": None, "hourly_series": [], "dow_series": [], "profit_heatmap_7x24": []}
 
         # Product aggregates + leaderboards + concentration
-        start_cur = pd.Timestamp(tfmeta["current_start"])
-        end_cur = pd.Timestamp(tfmeta["current_end"])
-        days = max(1.0, (end_cur - start_cur).total_seconds() / 86400.0)
+        try:
+            start_cur = pd.Timestamp(tfmeta["current_start"])
+            end_cur = pd.Timestamp(tfmeta["current_end"])
+            days = max(1.0, (end_cur - start_cur).total_seconds() / 86400.0)
+
+            g_products = self._build_product_aggregates(current_df)
+            if not g_products.empty:
+                g_products["units_per_day"] = g_products["units"] / days
+                product_lb = self._product_leaderboards(g_products)
+                concentration = self._concentration_block(g_products)
+            else:
+                product_lb = self._empty_product_leaderboards()
+                concentration = self._concentration_block(pd.DataFrame(columns=["revenue","units"]))
 
-        g_products = self._build_product_aggregates(current_df)
-        if not g_products.empty:
-            g_products["units_per_day"] = g_products["units"] / days
-            product_lb = self._product_leaderboards(g_products)
-            concentration = self._concentration_block(g_products)
-        else:
-            product_lb = self._empty_product_leaderboards()
-            concentration = self._concentration_block(pd.DataFrame(columns=["revenue","units"]))
+            snapshot["Product KPIs"] = {"leaderboards": product_lb, "concentration": concentration}
+        except Exception as e:
+            section_errors["products"] = str(e)
+            snapshot["Product KPIs"] = {"leaderboards": self._empty_product_leaderboards(), "concentration": self._concentration_block(pd.DataFrame(columns=["revenue","units"]))}
 
         # Customer value (RFM)
-        customer_val = self._customer_value(current_df, basket_df)
+        try:
+            # basket_df may or may not exist:
+            bdf = locals().get("basket_df", pd.DataFrame())
+            snapshot["Customer Value"] = self._customer_value(current_df, bdf)
+        except Exception as e:
+            section_errors["customer_value"] = str(e)
+            snapshot["Customer Value"] = {
+                "params": {"rfm_window_days": int(self.params["rfm_window_days"]), "retention_factor": float(self.params["retention_factor"]), "vip_count": 20},
+                "leaderboards": {"top_customers_by_gp": [], "at_risk": [], "new_customers": []},
+                "rfm_summary": {"unique_customers": 0, "median_recency_days": None, "median_orders": None, "median_gp": None}
+            }
 
         # Inventory (optional)
-        inventory = self._inventory_block(current_df, g_products, (start_cur, end_cur))
+        try:
+            g_products_for_inv = locals().get("g_products", pd.DataFrame())
+            snapshot["Inventory"] = self._inventory_block(current_df, g_products_for_inv, (start_cur, end_cur))
+        except Exception as e:
+            section_errors["inventory"] = str(e)
+            snapshot["Inventory"] = {"status": "no_stock_data", "products": [], "alerts": {"low_stock": [], "stockout_risk": [], "dead_stock": []}}
 
         # Branch analytics
-        branch_block = self._per_branch_blocks(current_df, previous_df, (start_cur, end_cur))
-
-        # Old snapshot maintained + new blocks appended
-        snapshot = {
-            "Summary Period": tfmeta.get("period_label", "This Week vs. Last Week"),
-            "Performance Snapshot (vs. Prior Period)": {
-                "Total Revenue": f"{headline['total_revenue_fmt']} ({headline['total_revenue_change']})",
-                "Gross Profit": f"{headline['gross_profit_fmt']} ({headline['gross_profit_change']})",
-                "Transactions": f"{headline['transactions_value']} ({headline['transactions_change']})",
+        try:
+            snapshot["Branch Analytics"] = self._per_branch_blocks(current_df, previous_df, (start_cur, end_cur))
+        except Exception as e:
+            section_errors["branch"] = str(e)
+            snapshot["Branch Analytics"] = {"params": self._branch_params(), "per_branch": {}, "cross_branch": {}}
+
+        # Meta
+        snapshot["meta"] = {
+            "timeframes": tfmeta,
+            "kpi_params": {
+                "top_k": int(self.params["top_k"]),
+                "min_revenue_for_margin_pct": float(self.params["min_revenue_for_margin_pct"]),
+                "min_tx_for_margin_pct": int(self.params["min_tx_for_margin_pct"]),
+                "rfm_window_days": int(self.params["rfm_window_days"]),
+                "retention_factor": float(self.params["retention_factor"]),
+                "min_support_baskets": int(self.params["min_support_baskets"]),
+                "min_lift": float(self.params["min_lift"]),
+                "blocked_products": list(self.params["blocked_products"]),
+                "cash_variance_threshold_abs": float(self.params["cash_variance_threshold_abs"]),
+                "cash_variance_threshold_pct": float(self.params["cash_variance_threshold_pct"]),
             },
-            "Basket Analysis": basket_kpis,
-            "Product Affinity": affinity,
-            "Temporal Patterns": temporal,
-            "Customer Value": customer_val,
-            "Product KPIs": {
-                "leaderboards": product_lb,
-                "concentration": concentration
+            "row_counts": {
+                "input": int(len(self.raw)),
+                "prepared": int(len(self.df)),
+                "current_period": int(len(current_df)),
+                "previous_period": int(len(previous_df)),
             },
-            "Inventory": inventory,
-            "Branch Analytics": branch_block,
-            "meta": {
-                "timeframes": tfmeta,
-                "kpi_params": {
-                    "top_k": int(self.params["top_k"]),
-                    "min_revenue_for_margin_pct": float(self.params["min_revenue_for_margin_pct"]),
-                    "min_tx_for_margin_pct": int(self.params["min_tx_for_margin_pct"]),
-                    "rfm_window_days": int(self.params["rfm_window_days"]),
-                    "retention_factor": float(self.params["retention_factor"]),
-                    "min_support_baskets": int(self.params["min_support_baskets"]),
-                    "min_lift": float(self.params["min_lift"]),
-                    "blocked_products": list(self.params["blocked_products"]),
-                    "cash_variance_threshold_abs": float(self.params["cash_variance_threshold_abs"]),
-                    "cash_variance_threshold_pct": float(self.params["cash_variance_threshold_pct"]),
-                },
-                "row_counts": {
-                    "input": int(len(self.raw)),
-                    "prepared": int(len(self.df)),
-                    "current_period": int(len(current_df)),
-                    "previous_period": int(len(previous_df)),
-                },
-                "notes": [
-                    "Non-sales transaction types excluded (e.g., Transaction_Type_ID != 21).",
-                    f"Duplicates dropped: {getattr(self, '_prepared_dupes_dropped', 0)}",
-                ],
-            }
+            "notes": [
+                "Non-sales transaction types excluded (e.g., Transaction_Type_ID != 21).",
+                f"Duplicates dropped: {getattr(self, '_prepared_dupes_dropped', 0)}",
+            ],
+            "section_errors": section_errors,  # surfaced to the client for your debug panel
         }
 
         emit_kpi_debug(self.profile_id, "briefing_done", snapshot["meta"])
@@ -1352,7 +1378,6 @@ class IrisReportEngine:
         Safe for PandasAI exception fallback.
         """
         try:
-            tz = TZ
             prompt = (
                 "You are Iris, a concise business analyst.\n"
                 "IMPORTANT RULES:\n"
@@ -1367,15 +1392,10 @@ class IrisReportEngine:
                 "Business Data (authoritative; JSON):\n"
                 f"{json.dumps(json_safe(briefing), ensure_ascii=False)}\n"
             )
-
             resp = self.llm.invoke(prompt)
-            # ChatGoogleGenerativeAI returns an object with .content
             text = getattr(resp, "content", None) or str(resp)
-            # Final safety scrub (remove accidental code fences / tracebacks)
             return sanitize_answer(text)
-            
         except Exception as e:
-            # Absolute last resort: dump a compact JSON view so the UI shows *something*
             fallback = {
                 "note": "Narrative fallback failed; returning raw snapshot.",
                 "error": str(e)[:200],