Spaces:
Running
Running
| """ | |
| Finance Knowledge Graph for Senti AI. | |
| Structured relationships between financial entities. | |
| What KAG knows that RAG does not: | |
| RAG: "What does section 12 of the Finance Act say?" | |
| KAG: "This invoice uses account 2020. 2020 is linked | |
| to VAT Payable. VAT Payable links to KRA obligation. | |
| KRA obligation has a 20th-of-month deadline rule. | |
| Therefore: this invoice triggers a tax obligation." | |
| KAG handles multi-hop reasoning like: | |
| "Does this transaction affect my PAYE?" | |
| Step 1: transaction → expense category | |
| Step 2: expense category → account code | |
| Step 3: account code → is_payroll_related? | |
| Step 4: is_payroll_related → PAYE implications | |
| Pre-built graphs: | |
| 1. Chart of Accounts graph | |
| 2. Tax rule dependency graph | |
| 3. Kenyan entity hierarchy graph (regulatory bodies) | |
| """ | |
| import networkx as nx | |
| from typing import Optional, List | |
| class FinanceKnowledgeGraph: | |
| def __init__(self): | |
| self.coa_graph = self._build_coa_graph() | |
| self.tax_graph = self._build_tax_graph() | |
| self.entity_graph = self._build_entity_graph() | |
| def _build_coa_graph(self) -> nx.DiGraph: | |
| """ | |
| Chart of Accounts relationship graph. | |
| Nodes: account codes | |
| Edges: parent-child + related-to relationships | |
| """ | |
| G = nx.DiGraph() | |
| # Account nodes with metadata | |
| accounts = [ | |
| ("1010", {"name": "Cash at Hand", "type": "asset", "liquidity": "high"}), | |
| ("1020", {"name": "M-Pesa Wallet", "type": "asset", "liquidity": "high"}), | |
| ("1040", {"name": "Accounts Receivable", "type": "asset", "liquidity": "medium"}), | |
| ("1050", {"name": "Inventory", "type": "asset", "liquidity": "medium"}), | |
| ("2010", {"name": "Accounts Payable", "type": "liability", "urgency": "medium"}), | |
| ("2020", {"name": "VAT Payable", "type": "liability", "urgency": "high", "tax": "VAT"}), | |
| ("2030", {"name": "PAYE Payable", "type": "liability", "urgency": "high", "tax": "PAYE"}), | |
| ("2031", {"name": "NSSF Payable", "type": "liability", "urgency": "high", "tax": "NSSF"}), | |
| ("2040", {"name": "TOT Payable", "type": "liability", "urgency": "high", "tax": "TOT"}), | |
| ("4010", {"name": "Sales Revenue", "type": "revenue", "kra_relevant": True}), | |
| ("5020", {"name": "Purchases", "type": "expense", "cogs": True}), | |
| ("6010", {"name": "Staff Salaries", "type": "expense", "payroll": True}), | |
| ] | |
| for code, attrs in accounts: | |
| G.add_node(code, **attrs) | |
| # Relationships | |
| edges = [ | |
| # Revenue triggers TOT obligation | |
| ("4010", "2040", {"relation": "triggers_obligation", "condition": "revenue > 500000 annually"}), | |
| # Salary expense triggers PAYE | |
| ("6010", "2030", {"relation": "triggers_obligation", "condition": "always"}), | |
| ("6010", "2031", {"relation": "triggers_obligation", "condition": "always"}), | |
| # VAT on purchases creates input VAT | |
| ("5020", "2020", {"relation": "input_vat_eligible", "condition": "if vat registered"}), | |
| # Cash and M-Pesa are liquidity sources | |
| ("1010", "1020", {"relation": "liquid_substitute"}), | |
| # Receivables convert to cash | |
| ("1040", "1010", {"relation": "converts_to", "condition": "on collection"}), | |
| ] | |
| for src, dst, attrs in edges: | |
| G.add_edge(src, dst, **attrs) | |
| return G | |
| def _build_tax_graph(self) -> nx.DiGraph: | |
| """ | |
| Kenya tax rule dependency graph. | |
| Nodes: tax types, thresholds, obligations. | |
| """ | |
| G = nx.DiGraph() | |
| G.add_nodes_from([ | |
| ("PAYE", {"rate": "10-35%", "due": "20th", "authority": "KRA", "filing": "iTax"}), | |
| ("TOT", {"rate": "3%", "due": "20th", "authority": "KRA", | |
| "threshold_min_annual": 500000, "threshold_max_annual": 25000000}), | |
| ("VAT", {"rate": "16%", "due": "20th", "authority": "KRA", | |
| "threshold_annual": 5000000}), | |
| ("NSSF", {"rate": "6%", "due": "15th", "authority": "NSSF", | |
| "employer_match": True}), | |
| ("SHA", {"rate": "2.75%", "due": "15th", "authority": "SHA"}), | |
| ("HOUSING", {"rate": "1.5%", "due": "9th", "authority": "NHCF"}), | |
| ("CORP_TAX", {"rate": "30%", "due": "annual","authority": "KRA"}), | |
| ]) | |
| G.add_edges_from([ | |
| ("PAYE", "KRA", {"action": "file_and_pay"}), | |
| ("TOT", "KRA", {"action": "file_and_pay", "paybill": "572572"}), | |
| ("VAT", "KRA", {"action": "file_and_pay"}), | |
| ("NSSF", "NSSF", {"action": "remit"}), | |
| ("SHA", "SHA", {"action": "remit"}), | |
| ("HOUSING","NHCF", {"action": "remit"}), | |
| ]) | |
| return G | |
| def _build_entity_graph(self) -> nx.DiGraph: | |
| """ | |
| Kenya financial regulatory entity graph. | |
| Who regulates whom. Where to file what. | |
| """ | |
| G = nx.DiGraph() | |
| G.add_nodes_from([ | |
| ("KRA", {"full_name": "Kenya Revenue Authority", "website": "kra.go.ke", "portal": "itax.kra.go.ke"}), | |
| ("CBK", {"full_name": "Central Bank of Kenya", "website": "centralbank.go.ke"}), | |
| ("CMA", {"full_name": "Capital Markets Authority", "website": "cma.or.ke"}), | |
| ("SASRA", {"full_name": "SACCO Societies Regulatory Authority", "website": "sasra.go.ke"}), | |
| ("NSSF", {"full_name": "National Social Security Fund"}), | |
| ("SHA", {"full_name": "Social Health Authority"}), | |
| ("NHCF", {"full_name": "National Housing Corp Fund"}), | |
| ("NSE", {"full_name": "Nairobi Securities Exchange","website": "nse.co.ke"}), | |
| ]) | |
| G.add_edges_from([ | |
| ("CBK", "BANKS", {"relation": "regulates"}), | |
| ("CMA", "FUNDS", {"relation": "regulates"}), | |
| ("SASRA", "SACCOS", {"relation": "regulates"}), | |
| ("KRA", "TAXPAYERS", {"relation": "collects_from"}), | |
| ]) | |
| return G | |
| def query( | |
| self, | |
| start_node: str, | |
| query_type: str, | |
| graph_name: str = "coa" | |
| ) -> dict: | |
| """ | |
| Answer a structured question using graph traversal. | |
| query_type: obligations|dependencies|path_to|related | |
| """ | |
| graph = { | |
| "coa": self.coa_graph, | |
| "tax": self.tax_graph, | |
| "entity": self.entity_graph, | |
| }.get(graph_name, self.coa_graph) | |
| if start_node not in graph: | |
| return {"found": False, "node": start_node} | |
| node_data = graph.nodes[start_node] | |
| result = {"node": start_node, "data": node_data} | |
| if query_type == "obligations": | |
| # Find what tax obligations this account triggers | |
| obligations = [] | |
| for _, target, edge_data in graph.out_edges(start_node, data=True): | |
| if edge_data.get("relation") == "triggers_obligation": | |
| target_data = graph.nodes.get(target, {}) | |
| obligations.append({ | |
| "obligation": target, | |
| "name": target_data.get("name", target), | |
| "tax_type": target_data.get("tax"), | |
| "condition": edge_data.get("condition"), | |
| }) | |
| result["obligations"] = obligations | |
| elif query_type == "related": | |
| # Direct neighbors | |
| neighbors = list(graph.neighbors(start_node)) | |
| result["related"] = [ | |
| {"code": n, **graph.nodes[n]} | |
| for n in neighbors | |
| ] | |
| elif query_type == "path_to": | |
| # Placeholder for multi-hop path finding | |
| result["paths"] = [] | |
| return result | |
| def is_tax_relevant(self, account_code: str) -> dict: | |
| """Quick check: does this account have tax implications?""" | |
| if account_code not in self.coa_graph: | |
| return {"tax_relevant": False} | |
| obligations = self.query(account_code, "obligations") | |
| has_obligations = len(obligations.get("obligations", [])) > 0 | |
| node = self.coa_graph.nodes.get(account_code, {}) | |
| is_tax_account = "tax" in node | |
| return { | |
| "tax_relevant": has_obligations or is_tax_account, | |
| "obligations": obligations.get("obligations", []), | |
| "account_type": node.get("type"), | |
| } | |
| def explain_transaction( | |
| self, amount: float, account_code: str | |
| ) -> str: | |
| """ | |
| Explain what a transaction means in KAG terms. | |
| Used to enrich LLM context for complex queries. | |
| """ | |
| tax_check = self.is_tax_relevant(account_code) | |
| node = self.coa_graph.nodes.get(account_code, {}) | |
| lines = [f"Account {account_code} ({node.get('name', 'Unknown')})"] | |
| if tax_check["tax_relevant"]: | |
| for obl in tax_check["obligations"]: | |
| tax = obl.get("tax_type") or obl.get("name") | |
| if tax: | |
| lines.append(f" → Triggers {tax} obligation") | |
| if node.get("payroll"): | |
| lines.append(" → Payroll-related: PAYE, NSSF, SHA, Housing Levy apply") | |
| if node.get("kra_relevant"): | |
| lines.append(f" → KES {amount:,.0f} is KRA-reportable revenue") | |
| return "\n".join(lines) | |
| finance_kg = FinanceKnowledgeGraph() | |