""" 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()