Spaces:
Running
Running
| from datetime import datetime | |
| from typing import List, Optional | |
| from .roles import InstitutionalRole, check_permission | |
| class ConversationalAnalytics: | |
| """ | |
| Institutional version of data conversation. | |
| RM/Risk teams talk to portfolio data. | |
| """ | |
| def query_portfolio( | |
| self, | |
| natural_language_query: str, | |
| role: InstitutionalRole, | |
| institution_id: str, | |
| db | |
| ) -> dict: | |
| """ | |
| Relationship Managers and Risk Analysts | |
| can ask questions about their portfolio. | |
| "Which customers are most at risk of default?" | |
| "Show me all loans above KES 500,000 from Q1" | |
| "Which branch has the best repayment rate?" | |
| "How many customers took their first loan | |
| in the last 30 days?" | |
| """ | |
| if not check_permission(role, "read_customers"): | |
| raise PermissionError("Insufficient role") | |
| # Parse the query into a safe database query | |
| # Never pass raw NL to SQL — always extract | |
| # entities and build query safely | |
| entities = self._extract_entities( | |
| natural_language_query | |
| ) | |
| results = self._safe_query( | |
| entities, institution_id, db | |
| ) | |
| return { | |
| "query": natural_language_query, | |
| "entities_extracted": entities, | |
| "results": results, | |
| "result_count": len(results), | |
| "generated_at": datetime.utcnow().isoformat(), | |
| "queried_by_role": role.value, | |
| "audit_logged": True | |
| } | |
| def _extract_entities(self, query: str) -> dict: | |
| """ | |
| Extract filters and metrics from NL. | |
| In production, this would use a more robust NER/intent model. | |
| """ | |
| query_lower = query.lower() | |
| entities = { | |
| "metric": "count", | |
| "filters": [], | |
| "time_range": "30d" | |
| } | |
| if "risk" in query_lower or "default" in query_lower: | |
| entities["metric"] = "risk_score" | |
| entities["filters"].append({"field": "risk_score", "op": ">", "val": 0.7}) | |
| if "loan" in query_lower: | |
| entities["target"] = "loans" | |
| if "repayment" in query_lower: | |
| entities["metric"] = "repayment_rate" | |
| return entities | |
| def _safe_query(self, entities: dict, institution_id: str, db) -> list: | |
| """ | |
| Execute safe queries based on extracted entities. | |
| Mock implementation for now. | |
| """ | |
| # This would normally be SQLAlchemy queries | |
| return [ | |
| {"customer_id": "CUST-001", "risk_score": 0.85, "status": "high_risk"}, | |
| {"customer_id": "CUST-042", "risk_score": 0.92, "status": "critical"} | |
| ] | |
| conversational_analytics = ConversationalAnalytics() | |