datasynth_runtime/
je_network.rs

1//! Shared Method-A / Method-B/C edge-list builder for the JE network export.
2//!
3//! v5.10: extracted from [`output_writer::write_je_network_csv`] so the
4//! same logic can be reused by the `datasynth-group` aggregate emitter,
5//! which builds both per-entity and consolidated edge lists from many
6//! per-entity JE batches.
7//!
8//! The builder is pure — no I/O — and returns a `Vec<JeNetworkEdge>`.
9//! Single-entity callers feed the result into the existing CSV writer
10//! (preserving v5.8.0 byte-identical output); group callers serialise
11//! the same struct with extra contextual columns (entity_code,
12//! is_eliminated, eliminates_ic_pair_id) added at write time.
13
14use chrono::NaiveDate;
15use rust_decimal::Decimal;
16use std::collections::HashMap;
17use uuid::Uuid;
18
19use datasynth_config::JeNetworkMethod;
20use datasynth_core::models::JournalEntry;
21
22/// One row of the je_network output.
23///
24/// The 13 columns matching v5.8.0 CSV output are at the top; v5.10
25/// added the optional IC fields (Some only on group-context inputs).
26#[derive(Debug, Clone)]
27pub struct JeNetworkEdge {
28    pub edge_id: String,
29    pub document_id: Uuid,
30    pub posting_date: NaiveDate,
31    pub from_account: String,
32    pub to_account: String,
33    pub from_line_id: String,
34    pub to_line_id: String,
35    pub amount: Decimal,
36    pub confidence: f64,
37    pub predecessor_edge_id: String,
38    pub business_process: String,
39    pub is_fraud: bool,
40    pub is_anomaly: bool,
41    /// v5.10 — surfaces `JournalEntryHeader::ic_pair_id` when present.
42    /// `None` for non-IC postings (and on every edge from a single-entity run).
43    pub ic_pair_id: Option<String>,
44    /// v5.10 — surfaces `JournalEntryHeader::ic_partner_entity` when present.
45    pub ic_partner_entity: Option<String>,
46}
47
48/// Build the Method-A / Method-B/C edge list for a batch of JEs.
49///
50/// JEs are processed in the iteration order of the slice; predecessor
51/// edge ids are resolved against earlier JEs in the same batch (so the
52/// caller must pass JEs that share predecessor chains in the same call).
53///
54/// Method semantics — see Ivertowski et al. (2024) Methods A through E:
55/// * `JeNetworkMethod::A` — bijective on 2-line entries only; multi-line
56///   JEs are skipped. Confidence on every emitted edge = `1.0`.
57/// * `JeNetworkMethod::Cartesian` — full Cartesian debit × credit product
58///   with proportional amount allocation; confidence = `1 / (n × m)`.
59pub fn build_je_network_edges(jes: &[JournalEntry], method: JeNetworkMethod) -> Vec<JeNetworkEdge> {
60    let mut edges = Vec::with_capacity(jes.len() * 2);
61    let mut line_id_to_edge_id: HashMap<String, String> = HashMap::with_capacity(jes.len() * 2);
62
63    for je in jes {
64        let h = &je.header;
65
66        let line_ids: Vec<String> = je
67            .lines
68            .iter()
69            .map(|l| {
70                l.transaction_id.clone().unwrap_or_else(|| {
71                    datasynth_core::models::JournalEntryLine::derive_transaction_id(
72                        l.document_id,
73                        l.line_number,
74                    )
75                })
76            })
77            .collect();
78
79        let debits: Vec<usize> = je
80            .lines
81            .iter()
82            .enumerate()
83            .filter(|(_, l)| l.debit_amount > Decimal::ZERO)
84            .map(|(i, _)| i)
85            .collect();
86        let credits: Vec<usize> = je
87            .lines
88            .iter()
89            .enumerate()
90            .filter(|(_, l)| l.credit_amount > Decimal::ZERO)
91            .map(|(i, _)| i)
92            .collect();
93        if debits.is_empty() || credits.is_empty() {
94            continue;
95        }
96
97        if method == JeNetworkMethod::A && !(debits.len() == 1 && credits.len() == 1) {
98            continue;
99        }
100
101        let total_debit: Decimal = debits.iter().map(|i| je.lines[*i].debit_amount).sum();
102        let total_credit: Decimal = credits.iter().map(|i| je.lines[*i].credit_amount).sum();
103        if total_debit.is_zero() || total_credit.is_zero() {
104            continue;
105        }
106
107        let confidence: f64 = if debits.len() == 1 && credits.len() == 1 {
108            1.0
109        } else {
110            1.0 / (debits.len() * credits.len()) as f64
111        };
112
113        let bp = h
114            .business_process
115            .map(|bp| format!("{bp:?}"))
116            .unwrap_or_default();
117        let ic_pair_id_str = h.ic_pair_id.as_ref().map(|id| id.to_string());
118        let ic_partner = h.ic_partner_entity.clone();
119
120        for &di in &debits {
121            let debit_line = &je.lines[di];
122            let to_line_id = &line_ids[di];
123            for &ci in &credits {
124                let credit_line = &je.lines[ci];
125                let from_line_id = &line_ids[ci];
126
127                // Edge id = UUID v5 of (document_id, debit.line_number,
128                // credit.line_number). Stable across regenerations.
129                let mut input = Vec::with_capacity(16 + 8);
130                input.extend_from_slice(h.document_id.as_bytes());
131                input.extend_from_slice(&debit_line.line_number.to_le_bytes());
132                input.extend_from_slice(&credit_line.line_number.to_le_bytes());
133                let edge_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, &input).to_string();
134
135                // Proportional allocation matches
136                // TransactionGraphBuilder::add_journal_entry_debit_credit.
137                let proportion = (debit_line.debit_amount / total_debit)
138                    * (credit_line.credit_amount / total_credit);
139                let amount = debit_line.debit_amount * proportion;
140
141                let predecessor_edge_id: String = credit_line
142                    .predecessor_line_id
143                    .as_ref()
144                    .or(debit_line.predecessor_line_id.as_ref())
145                    .and_then(|tx_id| line_id_to_edge_id.get(tx_id).cloned())
146                    .unwrap_or_default();
147
148                edges.push(JeNetworkEdge {
149                    edge_id: edge_id.clone(),
150                    document_id: h.document_id,
151                    posting_date: h.posting_date,
152                    from_account: credit_line.gl_account.clone(),
153                    to_account: debit_line.gl_account.clone(),
154                    from_line_id: from_line_id.clone(),
155                    to_line_id: to_line_id.clone(),
156                    amount,
157                    confidence,
158                    predecessor_edge_id,
159                    business_process: bp.clone(),
160                    is_fraud: h.is_fraud,
161                    is_anomaly: h.is_anomaly,
162                    ic_pair_id: ic_pair_id_str.clone(),
163                    ic_partner_entity: ic_partner.clone(),
164                });
165
166                line_id_to_edge_id
167                    .entry(from_line_id.clone())
168                    .or_insert(edge_id);
169            }
170        }
171    }
172
173    edges
174}
175
176#[cfg(test)]
177mod tests {
178    use super::*;
179    use datasynth_core::models::{JournalEntry, JournalEntryHeader, JournalEntryLine};
180
181    fn dec(v: i64) -> Decimal {
182        Decimal::from(v)
183    }
184
185    fn header_for(doc: Uuid) -> JournalEntryHeader {
186        let mut h = JournalEntryHeader::new(
187            "C001".to_string(),
188            NaiveDate::from_ymd_opt(2026, 5, 9).expect("2026-05-09 is a valid date"),
189        );
190        h.document_id = doc;
191        h
192    }
193
194    fn make_line(doc: Uuid, n: u32, account: &str, debit: i64, credit: i64) -> JournalEntryLine {
195        JournalEntryLine {
196            document_id: doc,
197            line_number: n,
198            gl_account: account.into(),
199            debit_amount: dec(debit),
200            credit_amount: dec(credit),
201            ..Default::default()
202        }
203    }
204
205    fn make_two_line_je(debit_account: &str, credit_account: &str, amount: i64) -> JournalEntry {
206        let document_id = Uuid::new_v4();
207        let header = header_for(document_id);
208        let lines = smallvec::smallvec![
209            make_line(document_id, 1, debit_account, amount, 0),
210            make_line(document_id, 2, credit_account, 0, amount),
211        ];
212        JournalEntry { header, lines }
213    }
214
215    #[test]
216    fn method_a_emits_one_edge_per_two_line_je() {
217        let jes = vec![
218            make_two_line_je("1000", "2000", 1_000),
219            make_two_line_je("1000", "4000", 5_000),
220        ];
221        let edges = build_je_network_edges(&jes, JeNetworkMethod::A);
222        assert_eq!(edges.len(), 2, "one edge per 2-line JE");
223        for e in &edges {
224            assert_eq!(e.confidence, 1.0, "Method A confidence is exactly 1.0");
225            assert!(e.ic_pair_id.is_none());
226            assert!(e.ic_partner_entity.is_none());
227        }
228    }
229
230    #[test]
231    fn method_a_skips_multi_line_jes() {
232        let document_id = Uuid::new_v4();
233        let header = header_for(document_id);
234        let lines = smallvec::smallvec![
235            make_line(document_id, 1, "1000", 1_000, 0),
236            make_line(document_id, 2, "1010", 500, 0),
237            make_line(document_id, 3, "2000", 0, 1_500),
238        ];
239        let je = JournalEntry { header, lines };
240        let edges = build_je_network_edges(&[je], JeNetworkMethod::A);
241        assert_eq!(edges.len(), 0, "3-line JE skipped under Method A");
242    }
243
244    #[test]
245    fn cartesian_emits_n_times_m_edges_per_je() {
246        let document_id = Uuid::new_v4();
247        let header = header_for(document_id);
248        let lines = smallvec::smallvec![
249            make_line(document_id, 1, "D1", 100, 0),
250            make_line(document_id, 2, "D2", 50, 0),
251            make_line(document_id, 3, "C1", 0, 80),
252            make_line(document_id, 4, "C2", 0, 70),
253        ];
254        let je = JournalEntry { header, lines };
255        let edges = build_je_network_edges(&[je], JeNetworkMethod::Cartesian);
256        assert_eq!(
257            edges.len(),
258            4,
259            "2 debits × 2 credits = 4 edges under Cartesian"
260        );
261        for e in &edges {
262            assert!((e.confidence - 0.25).abs() < 1e-9, "1/(n*m) = 0.25");
263        }
264    }
265
266    #[test]
267    fn ic_fields_surface_when_present_on_header() {
268        let document_id = Uuid::new_v4();
269        let mut header = header_for(document_id);
270        header.ic_partner_entity = Some("ACME_EUR".to_string());
271
272        let lines = smallvec::smallvec![
273            make_line(document_id, 1, "1150", 1000, 0),
274            make_line(document_id, 2, "4500", 0, 1000),
275        ];
276        let je = JournalEntry { header, lines };
277        let edges = build_je_network_edges(&[je], JeNetworkMethod::A);
278        assert_eq!(edges.len(), 1);
279        assert_eq!(edges[0].ic_partner_entity, Some("ACME_EUR".to_string()));
280        // ic_pair_id requires construction via IcPairId — covered in group tests.
281    }
282}
datasynth_runtime/je_network.rs

datasynth_runtime/
je_network.rs
