datasynth_graph/builders/
transaction_graph.rs1use std::collections::HashMap;
8
9use rust_decimal::Decimal;
10
11use datasynth_core::models::JournalEntry;
12
13use crate::models::{
14 AccountNode, EdgeType, Graph, GraphEdge, GraphNode, GraphType, NodeId, NodeType,
15 TransactionEdge,
16};
17
18#[derive(Debug, Clone)]
20pub struct TransactionGraphConfig {
21 pub include_vendors: bool,
23 pub include_customers: bool,
25 pub create_debit_credit_edges: bool,
27 pub include_document_nodes: bool,
29 pub min_edge_weight: f64,
31 pub aggregate_parallel_edges: bool,
33}
34
35impl Default for TransactionGraphConfig {
36 fn default() -> Self {
37 Self {
38 include_vendors: false,
39 include_customers: false,
40 create_debit_credit_edges: true,
41 include_document_nodes: false,
42 min_edge_weight: 0.0,
43 aggregate_parallel_edges: false,
44 }
45 }
46}
47
48pub struct TransactionGraphBuilder {
50 config: TransactionGraphConfig,
51 graph: Graph,
52 account_nodes: HashMap<String, NodeId>,
54 document_nodes: HashMap<String, NodeId>,
56 edge_aggregation: HashMap<(NodeId, NodeId), AggregatedEdge>,
58}
59
60impl TransactionGraphBuilder {
61 pub fn new(config: TransactionGraphConfig) -> Self {
63 Self {
64 config,
65 graph: Graph::new("transaction_network", GraphType::Transaction),
66 account_nodes: HashMap::new(),
67 document_nodes: HashMap::new(),
68 edge_aggregation: HashMap::new(),
69 }
70 }
71
72 pub fn add_journal_entry(&mut self, entry: &JournalEntry) {
74 if self.config.include_document_nodes {
75 self.add_journal_entry_with_document(entry);
76 } else if self.config.create_debit_credit_edges {
77 self.add_journal_entry_debit_credit(entry);
78 }
79 }
80
81 fn add_journal_entry_debit_credit(&mut self, entry: &JournalEntry) {
83 let debits: Vec<_> = entry
85 .lines
86 .iter()
87 .filter(|l| l.debit_amount > Decimal::ZERO)
88 .collect();
89
90 let credits: Vec<_> = entry
91 .lines
92 .iter()
93 .filter(|l| l.credit_amount > Decimal::ZERO)
94 .collect();
95
96 for debit in &debits {
98 let source_id = self.get_or_create_account_node(
99 debit.account_code(),
100 debit.account_description(),
101 entry.company_code(),
102 );
103
104 for credit in &credits {
105 let target_id = self.get_or_create_account_node(
106 credit.account_code(),
107 credit.account_description(),
108 entry.company_code(),
109 );
110
111 let total_debit: Decimal = debits.iter().map(|d| d.debit_amount).sum();
113 let total_credit: Decimal = credits.iter().map(|c| c.credit_amount).sum();
114
115 let proportion =
116 (debit.debit_amount / total_debit) * (credit.credit_amount / total_credit);
117 let edge_amount = debit.debit_amount * proportion;
118 let edge_weight: f64 = edge_amount.try_into().unwrap_or(0.0);
119
120 if edge_weight < self.config.min_edge_weight {
121 continue;
122 }
123
124 if self.config.aggregate_parallel_edges {
125 self.aggregate_edge(source_id, target_id, edge_weight, entry);
126 } else {
127 let mut tx_edge = TransactionEdge::new(
128 0,
129 source_id,
130 target_id,
131 entry.document_number(),
132 entry.posting_date(),
133 edge_amount,
134 true,
135 );
136 tx_edge.company_code = entry.company_code().to_string();
137 tx_edge.cost_center = debit.cost_center.clone();
138 tx_edge.compute_features();
139
140 self.graph.add_edge(tx_edge.edge);
141 }
142 }
143 }
144 }
145
146 fn add_journal_entry_with_document(&mut self, entry: &JournalEntry) {
148 let doc_id =
150 self.get_or_create_document_node(&entry.document_number(), entry.company_code());
151
152 for line in &entry.lines {
154 let account_id = self.get_or_create_account_node(
155 line.account_code(),
156 line.account_description(),
157 entry.company_code(),
158 );
159
160 let is_debit = line.debit_amount > Decimal::ZERO;
161 let amount = if is_debit {
162 line.debit_amount
163 } else {
164 line.credit_amount
165 };
166
167 let mut tx_edge = TransactionEdge::new(
168 0,
169 doc_id,
170 account_id,
171 entry.document_number(),
172 entry.posting_date(),
173 amount,
174 is_debit,
175 );
176 tx_edge.company_code = entry.company_code().to_string();
177 tx_edge.cost_center = line.cost_center.clone();
178 tx_edge.compute_features();
179
180 self.graph.add_edge(tx_edge.edge);
181 }
182 }
183
184 fn get_or_create_account_node(
186 &mut self,
187 account_code: &str,
188 account_name: &str,
189 company_code: &str,
190 ) -> NodeId {
191 let key = format!("{}_{}", company_code, account_code);
192
193 if let Some(&id) = self.account_nodes.get(&key) {
194 return id;
195 }
196
197 let mut account = AccountNode::new(
198 0,
199 account_code.to_string(),
200 account_name.to_string(),
201 Self::infer_account_type(account_code),
202 company_code.to_string(),
203 );
204 account.is_balance_sheet = Self::is_balance_sheet_account(account_code);
205 account.normal_balance = Self::infer_normal_balance(account_code);
206 account.compute_features();
207
208 let id = self.graph.add_node(account.node);
209 self.account_nodes.insert(key, id);
210 id
211 }
212
213 fn get_or_create_document_node(&mut self, document_number: &str, company_code: &str) -> NodeId {
215 let key = format!("{}_{}", company_code, document_number);
216
217 if let Some(&id) = self.document_nodes.get(&key) {
218 return id;
219 }
220
221 let node = GraphNode::new(
222 0,
223 NodeType::JournalEntry,
224 document_number.to_string(),
225 document_number.to_string(),
226 );
227
228 let id = self.graph.add_node(node);
229 self.document_nodes.insert(key, id);
230 id
231 }
232
233 fn aggregate_edge(
235 &mut self,
236 source: NodeId,
237 target: NodeId,
238 weight: f64,
239 entry: &JournalEntry,
240 ) {
241 let key = (source, target);
242 let agg = self.edge_aggregation.entry(key).or_insert(AggregatedEdge {
243 source,
244 target,
245 total_weight: 0.0,
246 count: 0,
247 first_date: entry.posting_date(),
248 last_date: entry.posting_date(),
249 });
250
251 agg.total_weight += weight;
252 agg.count += 1;
253 if entry.posting_date() < agg.first_date {
254 agg.first_date = entry.posting_date();
255 }
256 if entry.posting_date() > agg.last_date {
257 agg.last_date = entry.posting_date();
258 }
259 }
260
261 fn infer_account_type(account_code: &str) -> String {
263 if account_code.is_empty() {
264 return "Unknown".to_string();
265 }
266
267 match account_code.chars().next().unwrap() {
268 '1' => "Asset".to_string(),
269 '2' => "Liability".to_string(),
270 '3' => "Equity".to_string(),
271 '4' => "Revenue".to_string(),
272 '5' | '6' | '7' => "Expense".to_string(),
273 _ => "Unknown".to_string(),
274 }
275 }
276
277 fn is_balance_sheet_account(account_code: &str) -> bool {
279 if account_code.is_empty() {
280 return false;
281 }
282
283 matches!(account_code.chars().next().unwrap(), '1' | '2' | '3')
284 }
285
286 fn infer_normal_balance(account_code: &str) -> String {
288 if account_code.is_empty() {
289 return "Debit".to_string();
290 }
291
292 match account_code.chars().next().unwrap() {
293 '1' | '5' | '6' | '7' => "Debit".to_string(),
294 '2' | '3' | '4' => "Credit".to_string(),
295 _ => "Debit".to_string(),
296 }
297 }
298
299 pub fn build(mut self) -> Graph {
301 if self.config.aggregate_parallel_edges {
303 for ((source, target), agg) in self.edge_aggregation {
304 let mut edge = GraphEdge::new(0, source, target, EdgeType::Transaction)
305 .with_weight(agg.total_weight)
306 .with_timestamp(agg.last_date);
307
308 edge.features.push((agg.total_weight + 1.0).ln());
310 edge.features.push(agg.count as f64);
311
312 let duration = (agg.last_date - agg.first_date).num_days() as f64;
313 edge.features.push(duration);
314
315 self.graph.add_edge(edge);
316 }
317 }
318
319 self.graph.compute_statistics();
320 self.graph
321 }
322
323 pub fn add_journal_entries(&mut self, entries: &[JournalEntry]) {
325 for entry in entries {
326 self.add_journal_entry(entry);
327 }
328 }
329}
330
331#[allow(dead_code)]
333struct AggregatedEdge {
334 source: NodeId,
335 target: NodeId,
336 total_weight: f64,
337 count: usize,
338 first_date: chrono::NaiveDate,
339 last_date: chrono::NaiveDate,
340}
341
342#[cfg(test)]
343mod tests {
344 use super::*;
345 use datasynth_core::models::JournalEntryLine;
346 use rust_decimal_macros::dec;
347
348 fn create_test_entry() -> JournalEntry {
349 let mut entry = JournalEntry::new_simple(
350 "JE001".to_string(),
351 "1000".to_string(),
352 chrono::NaiveDate::from_ymd_opt(2024, 1, 15).unwrap(),
353 "Test Entry".to_string(),
354 );
355
356 let doc_id = entry.header.document_id;
357
358 entry.add_line(JournalEntryLine::debit(
359 doc_id,
360 1,
361 "1000".to_string(),
362 dec!(1000),
363 ));
364
365 entry.add_line(JournalEntryLine::credit(
366 doc_id,
367 2,
368 "4000".to_string(),
369 dec!(1000),
370 ));
371
372 entry
373 }
374
375 #[test]
376 fn test_build_transaction_graph() {
377 let mut builder = TransactionGraphBuilder::new(TransactionGraphConfig::default());
378 builder.add_journal_entry(&create_test_entry());
379
380 let graph = builder.build();
381
382 assert_eq!(graph.node_count(), 2); assert_eq!(graph.edge_count(), 1); }
385
386 #[test]
387 fn test_with_document_nodes() {
388 let config = TransactionGraphConfig {
389 include_document_nodes: true,
390 create_debit_credit_edges: false,
391 ..Default::default()
392 };
393
394 let mut builder = TransactionGraphBuilder::new(config);
395 builder.add_journal_entry(&create_test_entry());
396
397 let graph = builder.build();
398
399 assert_eq!(graph.node_count(), 3); assert_eq!(graph.edge_count(), 2); }
402}