pomelo_audit/
lib.rs

1//! Read-only data-quality audit of a pomelo data-layout tree (#133).
2//!
3//! Given a synced `prices/` / `fundamentals/` / `panels/` / `tracked/` tree,
4//! [`run_data_audit`] answers *"is this clean enough to trust a backtest?"* —
5//! turning "high-quality data" from a claim into a measurement. It also
6//! doubles as the verification tool for #131 (filing-date lag) and #132
7//! (snapshot-factor coverage).
8//!
9//! Storage-agnostic (#149): `run_data_audit` takes any [`ObjectSource`] +
10//! [`ObjectLister`], so it audits a local tree or an S3/R2 tree identically —
11//! `yuzu-cli data-audit --data s3://…` builds a `pomelo_s3::OutStore` and
12//! passes it straight in. Discovery (which symbols / fundamentals files /
13//! membership panels exist) goes through `ObjectLister::list`; reads go
14//! through `ObjectSource::get`. Over S3 this means shallow checks (coverage,
15//! membership) cost a handful of `ListObjectsV2` calls, while deep checks
16//! (gaps / jumps / NaN / lookahead) GET every object — auditing a full R2 tree
17//! deeply is therefore comparable in cost to downloading it (see
18//! `docs/fmp-data-source.md`).
19//!
20//! No network beyond the source's own reads, no engine run. It reuses the
21//! [`pomelo_data`] loaders and returns a serializable [`DataAuditReport`] of
22//! per-check `OK` / `WARN` / `FAIL` verdicts, so any front end —
23//! `yuzu-cli data-audit`, a nightly job, or a backend service — can call the
24//! same logic. The CLI is a thin shim that renders/emits the report and maps a
25//! `FAIL` to a non-zero exit.
26//!
27//! ## Module layout
28//!
29//! - [`report`] — the [`Status`] / [`Check`] / [`DataAuditReport`] shapes,
30//!   [`render_table`], and the shared formatting/date helpers.
31//! - [`checks`] — the individual `check_*` data-quality checks.
32//! - [`scan`] — tree discovery and the single-pass fundamentals scan.
33//! - this module — `run_data_audit` orchestration + the public re-exports.
34
35mod checks;
36mod report;
37mod scan;
38
39pub use report::{render_table, Check, DataAuditReport, Status};
40
41use pomelo_data::{load_panel, Field, ObjectLister, ObjectSource, PRICES_DIR};
42
43use checks::{
44    check_adjustment, check_calendar_gaps, check_coverage, check_index_membership,
45    check_nan_density, check_pit_lag, check_survivorship,
46};
47use scan::{list_price_symbols, scan_fundamentals};
48
49/// Run every check over the data-layout tree served by `src`, windowed to
50/// `[from, to]`. `data_dir` is a display label only (a local path or an
51/// `s3://…` URL) — it never drives I/O; all discovery goes through `src`.
52/// Fail-soft: a missing directory or file downgrades a check, never panics.
53pub fn run_data_audit<S: ObjectSource + ObjectLister + Sync>(
54    src: &S,
55    data_dir: &str,
56    from: i32,
57    to: i32,
58) -> Result<DataAuditReport, String> {
59    let symbols = list_price_symbols(src);
60
61    // One adj-close Panel (union calendar × symbols, NaN where absent) backs the
62    // coverage / gaps / delist / jump checks.
63    let closes = if symbols.is_empty() {
64        None
65    } else {
66        Some(
67            load_panel(src, &symbols, Field::AdjClose, from, to, PRICES_DIR)
68                .map_err(|e| format!("loading price panel: {e}"))?,
69        )
70    };
71
72    // Fundamentals are parsed once (per-field coverage + filing-event days).
73    let fund = scan_fundamentals(src, from, to);
74
75    let checks = vec![
76        check_coverage(src, &symbols, closes.as_ref()),
77        check_calendar_gaps(&symbols, closes.as_ref()),
78        check_adjustment(&symbols, closes.as_ref()),
79        check_survivorship(&symbols, closes.as_ref()),
80        check_nan_density(src, &symbols, &fund, from, to),
81        check_pit_lag(&fund),
82        check_index_membership(src, &symbols, from, to),
83    ];
84
85    let overall = checks.iter().map(|c| c.status).max().unwrap_or(Status::Ok);
86    Ok(DataAuditReport {
87        data_dir: data_dir.to_string(),
88        from,
89        to,
90        symbol_count: symbols.len(),
91        overall,
92        checks,
93    })
94}
95
96#[cfg(test)]
97mod tests {
98    use super::*;
99    use crate::report::status_str;
100    use pomelo_data::csv_io::{write_series, OhlcvRow};
101    use pomelo_data::fundamentals::{write_fundamentals, FundamentalRow};
102    use pomelo_data::LocalSource;
103    use pomelo_data::FUNDAMENTAL_FIELDS;
104    use std::fs;
105    use std::path::{Path, PathBuf};
106
107    /// Run the audit over a local tree, matching the CLI's local-path path.
108    fn run_local(dir: &Path, from: i32, to: i32) -> DataAuditReport {
109        let src = LocalSource::new(dir);
110        run_data_audit(&src, &dir.display().to_string(), from, to).unwrap()
111    }
112
113    fn tmp(tag: &str) -> PathBuf {
114        let d = std::env::temp_dir().join(format!("pomelo_audit_ut_{tag}"));
115        let _ = fs::remove_dir_all(&d);
116        fs::create_dir_all(&d).unwrap();
117        d
118    }
119
120    fn bar(day: i32, close: f64) -> OhlcvRow {
121        OhlcvRow {
122            day,
123            adj_open: close,
124            adj_high: close,
125            adj_low: close,
126            adj_close: close,
127            volume: 0.0,
128        }
129    }
130
131    fn write_prices(dir: &Path, sym: &str, bars: &[(i32, f64)]) {
132        let p = dir.join("prices");
133        fs::create_dir_all(&p).unwrap();
134        let rows: Vec<OhlcvRow> = bars.iter().map(|&(d, c)| bar(d, c)).collect();
135        fs::write(
136            p.join(format!("{sym}.csv.gz")),
137            write_series(&rows).unwrap(),
138        )
139        .unwrap();
140    }
141
142    fn find<'a>(r: &'a DataAuditReport, name: &str) -> &'a Check {
143        r.checks.iter().find(|c| c.name == name).unwrap()
144    }
145
146    /// A tree that trips one specific check each — the full WARN sweep.
147    fn build_rich_tree(tag: &str) -> PathBuf {
148        let dir = tmp(tag);
149        let d = [20240102, 20240103, 20240104, 20240105];
150        write_prices(
151            &dir,
152            "GOOD",
153            &[(d[0], 100.0), (d[1], 101.0), (d[2], 102.0), (d[3], 103.0)],
154        );
155        write_prices(&dir, "GAP", &[(d[0], 50.0), (d[1], 51.0), (d[3], 52.0)]); // 2024-01-04 missing
156        write_prices(
157            &dir,
158            "SPLIT",
159            &[(d[0], 100.0), (d[1], 100.0), (d[2], 200.0), (d[3], 201.0)],
160        );
161        write_prices(&dir, "DEAD", &[(d[0], 100.0), (d[1], 101.0)]); // ends early
162
163        // Fundamentals for GOOD: `pe` populated; a filing on 2023-12-31 (a month-end).
164        let fdir = dir.join("fundamentals");
165        fs::create_dir_all(&fdir).unwrap();
166        let mut vals = vec![f64::NAN; FUNDAMENTAL_FIELDS.len()];
167        vals[0] = 15.0; // pe
168        let rows = vec![
169            FundamentalRow {
170                day: 20231229,
171                values: vals.clone(),
172                report_event: 0.0,
173            },
174            FundamentalRow {
175                day: 20231231,
176                values: vals.clone(),
177                report_event: 1.0,
178            },
179        ];
180        fs::write(fdir.join("GOOD.csv.gz"), write_fundamentals(&rows).unwrap()).unwrap();
181
182        // An all-NaN snapshot-factor panel (plain .csv — the loader probes .csv).
183        let pdir = dir.join("panels");
184        fs::create_dir_all(&pdir).unwrap();
185        fs::write(
186            pdir.join("piotroski_score.csv"),
187            "day,GOOD\n2024-01-02,\n2024-01-03,\n",
188        )
189        .unwrap();
190
191        // Universe map with an extra name that has no price file.
192        let tdir = dir.join("tracked");
193        fs::create_dir_all(&tdir).unwrap();
194        fs::write(
195            tdir.join("universe.csv"),
196            "symbol,sector,market_cap\nGOOD,Tech,1e12\nGAP,Tech,1e11\nSPLIT,Tech,1e11\nDEAD,Tech,1e10\nMISSING,Tech,1e9\n",
197        )
198        .unwrap();
199        dir
200    }
201
202    #[test]
203    fn rich_tree_flags_each_check() {
204        let dir = build_rich_tree("rich");
205        let r = run_local(&dir, 20000101, 99991231);
206        assert_eq!(r.symbol_count, 4);
207        assert_eq!(r.overall, Status::Warn);
208
209        let cov = find(&r, "coverage");
210        assert_eq!(cov.status, Status::Warn);
211        assert!(cov.details["in_universe_missing_prices"]
212            .as_array()
213            .unwrap()
214            .iter()
215            .any(|v| v == "MISSING"));
216
217        assert_eq!(find(&r, "calendar_gaps").status, Status::Warn);
218        assert_eq!(find(&r, "calendar_gaps").details["total_holes"], 1);
219
220        assert_eq!(find(&r, "adjustment").status, Status::Warn);
221        assert_eq!(find(&r, "adjustment").details["flagged"], 1);
222
223        assert_eq!(find(&r, "survivorship").status, Status::Ok);
224        assert_eq!(find(&r, "survivorship").details["ended_early"], 1);
225
226        assert_eq!(find(&r, "nan_density").status, Status::Warn);
227        assert!(find(&r, "nan_density").details["all_nan_factor_panels"]
228            .as_array()
229            .unwrap()
230            .iter()
231            .any(|v| v == "piotroski_score"));
232
233        let pit = find(&r, "pit_lag");
234        assert_eq!(pit.status, Status::Warn);
235        assert_eq!(pit.details["report_events"], 1);
236        assert_eq!(pit.details["on_month_end"], 1);
237
238        assert_eq!(find(&r, "index_membership").status, Status::Ok);
239
240        // The rendered table shows the overall WARN and the flagged checks.
241        let table = render_table(&r);
242        assert!(table.contains("WARN"));
243        assert!(table.contains("adjustment"));
244    }
245
246    #[test]
247    fn empty_tree_fails_and_takes_no_price_arms() {
248        let d = tmp("empty");
249        let r = run_local(&d, 20000101, 99991231);
250        assert_eq!(r.overall, Status::Fail);
251        assert_eq!(r.symbol_count, 0);
252        assert_eq!(find(&r, "coverage").status, Status::Fail);
253        for name in ["calendar_gaps", "adjustment", "survivorship"] {
254            assert_eq!(find(&r, name).status, Status::Ok);
255        }
256        let table = render_table(&r);
257        assert!(table.contains("FAIL"));
258        assert!(table.contains("coverage"));
259        assert_eq!(status_str(Status::Ok), "OK");
260        assert_eq!(status_str(Status::Warn), "WARN");
261        assert_eq!(status_str(Status::Fail), "FAIL");
262    }
263
264    #[test]
265    fn prices_without_universe_or_fundamentals_are_ok() {
266        let d = tmp("bare");
267        write_prices(
268            &d,
269            "AAA",
270            &[(20240102, 10.0), (20240103, 10.0), (20240104, 10.0)],
271        );
272        let r = run_local(&d, 20000101, 99991231);
273        assert_eq!(r.overall, Status::Ok);
274        let cov = find(&r, "coverage");
275        assert_eq!(cov.status, Status::Ok);
276        assert!(cov.summary.contains("no tracked/universe"));
277        assert!(cov.details["date_range"]["first_day"] == 20240102);
278        assert_eq!(find(&r, "survivorship").status, Status::Ok);
279        assert_eq!(find(&r, "nan_density").status, Status::Ok);
280        assert!(find(&r, "nan_density")
281            .summary
282            .contains("no fundamentals or factor panels"));
283        assert_eq!(find(&r, "pit_lag").status, Status::Ok);
284    }
285
286    #[test]
287    fn index_membership_panel_is_summarized() {
288        let d = tmp("index");
289        write_prices(&d, "AAA", &[(20240102, 10.0), (20240103, 10.0)]);
290        write_prices(&d, "BBB", &[(20240102, 10.0), (20240103, 10.0)]);
291        let pan = d.join("panels");
292        fs::create_dir_all(&pan).unwrap();
293        fs::write(
294            pan.join("in_sp500.csv"),
295            "day,AAA,BBB\n2024-01-02,1,\n2024-01-03,1,1\n",
296        )
297        .unwrap();
298        let r = run_local(&d, 20000101, 99991231);
299        let idx = find(&r, "index_membership");
300        assert_eq!(idx.status, Status::Ok);
301        let p0 = &idx.details["panels"][0];
302        assert_eq!(p0["panel"], "in_sp500");
303        assert_eq!(p0["min_members"], 1);
304        assert_eq!(p0["max_members"], 2);
305        assert_eq!(p0["last_members"], 2);
306    }
307
308    #[test]
309    fn edge_arms_single_point_zero_close_and_empty_membership() {
310        let d = tmp("edge");
311        write_prices(&d, "ONE", &[(20240102, 10.0)]);
312        write_prices(&d, "ZERO", &[(20240102, 0.0), (20240103, 10.0)]);
313        let pan = d.join("panels");
314        fs::create_dir_all(&pan).unwrap();
315        fs::write(pan.join("in_empty.csv"), "day,ONE,ZERO\n2024-01-02,,\n").unwrap();
316
317        let r = run_local(&d, 20000101, 99991231);
318        // The 0 → 10 step is skipped by the c0 <= 0 guard, so nothing is flagged.
319        assert_eq!(find(&r, "adjustment").details["flagged"], 0);
320        assert_eq!(find(&r, "survivorship").status, Status::Ok);
321        let idx = find(&r, "index_membership");
322        assert_eq!(idx.status, Status::Warn);
323        assert_eq!(idx.details["panels"][0]["max_members"], 0);
324    }
325
326    /// #149: `run_data_audit` is storage-agnostic — this drives it against a
327    /// stub S3 endpoint (routing `ListObjectsV2` + `GET` requests by hand,
328    /// mirroring `pomelo-s3`'s own stub-server tests) instead of `LocalSource`,
329    /// proving discovery (via `ObjectLister::list`) and reads (via
330    /// `ObjectSource::get`) both work over S3/R2.
331    #[test]
332    fn run_data_audit_over_a_stub_s3_source() {
333        use pomelo_s3::S3Source;
334        use std::io::{Read, Write};
335        use std::net::TcpListener;
336
337        fn list_xml(keys: &[&str]) -> String {
338            let contents: String = keys
339                .iter()
340                .map(|k| {
341                    format!(
342                        "<Contents><Key>{k}</Key><LastModified>2020-01-01T00:00:00.000Z</LastModified>\
343                         <ETag>\"e\"</ETag><Size>1</Size><StorageClass>STANDARD</StorageClass></Contents>"
344                    )
345                })
346                .collect();
347            format!(
348                "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
349                 <ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">\
350                 <Name>bucket</Name><KeyCount>{}</KeyCount><MaxKeys>1000</MaxKeys>\
351                 <IsTruncated>false</IsTruncated>{contents}<EncodingType>url</EncodingType>\
352                 </ListBucketResult>",
353                keys.len()
354            )
355        }
356        fn http_ok_bytes(body: &[u8]) -> Vec<u8> {
357            let mut resp = format!(
358                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
359                body.len()
360            )
361            .into_bytes();
362            resp.extend_from_slice(body);
363            resp
364        }
365        const NOT_FOUND: &[u8] =
366            b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
367
368        let aapl = write_series(&[bar(20240102, 10.0), bar(20240103, 11.0)]).unwrap();
369        let msft = write_series(&[bar(20240102, 20.0), bar(20240103, 21.0)]).unwrap();
370
371        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
372        let addr = listener.local_addr().unwrap();
373        std::thread::spawn(move || {
374            for stream in listener.incoming() {
375                let Ok(mut sock) = stream else { break };
376                let mut buf = [0u8; 4096];
377                let read = sock.read(&mut buf).unwrap_or(0);
378                let req = String::from_utf8_lossy(&buf[..read]).to_string();
379                let first_line = req.lines().next().unwrap_or("");
380                let resp = if first_line.contains("prefix=prices") {
381                    http_ok_bytes(
382                        list_xml(&["prices/AAPL.csv.gz", "prices/MSFT.csv.gz"]).as_bytes(),
383                    )
384                } else if first_line.contains("prices/AAPL.csv.gz") {
385                    http_ok_bytes(&aapl)
386                } else if first_line.contains("prices/MSFT.csv.gz") {
387                    http_ok_bytes(&msft)
388                } else if first_line.contains("prefix=fundamentals")
389                    || first_line.contains("prefix=panels")
390                {
391                    http_ok_bytes(list_xml(&[]).as_bytes())
392                } else {
393                    // Every other lookup (universe map, snapshot-factor panels) is
394                    // "not present" — a bare synced tree, same as the local
395                    // `prices_without_universe_or_fundamentals_are_ok` case.
396                    NOT_FOUND.to_vec()
397                };
398                let _ = sock.write_all(&resp);
399            }
400        });
401
402        let src = S3Source::new(
403            &format!("http://{addr}"),
404            "bucket",
405            "ak",
406            "sk",
407            None,
408            "auto",
409        )
410        .unwrap();
411        let r = run_data_audit(&src, "s3://bucket", 20000101, 99991231).unwrap();
412        assert_eq!(r.data_dir, "s3://bucket");
413        assert_eq!(r.symbol_count, 2);
414        assert_eq!(find(&r, "coverage").status, Status::Ok);
415        assert_eq!(find(&r, "coverage").details["symbols_with_prices"], 2);
416        assert_eq!(find(&r, "nan_density").status, Status::Ok);
417        assert_eq!(find(&r, "index_membership").status, Status::Ok);
418    }
419}
pomelo_audit/lib.rs

pomelo_audit/
lib.rs
