Spaces:
Running
Running
File size: 8,852 Bytes
021e065 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 | """
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()
|