"""Calendar and boundary audit for Financial ML from Zero, part 4. Python 3.10+, standard library only. This is a splitting demonstration, not mortgage data, a model trainer, or a performance experiment. python walk_forward.py python walk_forward.py --check python walk_forward.py --out fold_membership.csv The example assumes month-end availability and a full 12-month horizon for EVERY label. Real reporting/ingestion delays belong in label_available_at. None means the outcome is unresolved, never negative. """ from __future__ import annotations import argparse import calendar import csv from dataclasses import dataclass, replace from datetime import date from pathlib import Path import unittest @dataclass(frozen=True) class Observation: key: str decision_time: date label_end: date label_available_at: date | None @dataclass(frozen=True) class Fold: year: int train: tuple[Observation, ...] calibration: tuple[Observation, ...] policy: tuple[Observation, ...] evaluation: tuple[Observation, ...] train_excluded: int def eligible_before(rows, boundary): return tuple( row for row in rows if row.decision_time < boundary and row.label_end < boundary and row.label_available_at is not None and row.label_available_at < boundary ) def validate_rows(rows): keys = set() for row in rows: if row.key in keys: raise ValueError(f"Duplicate observation key: {row.key}") keys.add(row.key) if row.label_end <= row.decision_time: raise ValueError(f"Non-forward label interval: {row.key}") if (row.label_available_at is not None and row.label_available_at < row.label_end): raise ValueError(f"Availability precedes full horizon: {row.key}") def in_window(rows, start, stop): # Half-open decision windows: start is included, stop is excluded. return tuple(row for row in rows if start <= row.decision_time < stop) def require_ready(rows, boundary, name): if not rows: raise ValueError(f"Empty {name} window") ready = eligible_before(rows, boundary) if len(ready) != len(rows): raise ValueError(f"{name}: {len(rows) - len(ready)} outcomes unavailable " f"before {boundary}") def make_fold(rows, year): rows = tuple(sorted(rows, key=lambda row: (row.decision_time, row.key))) validate_rows(rows) calibration_start = date(year - 4, 1, 1) policy_start = date(year - 2, 1, 1) evaluation_start = date(year, 1, 1) candidates = tuple(row for row in rows if row.decision_time < calibration_start) train = eligible_before(candidates, calibration_start) if not train: raise ValueError("Empty training window after purging") calibration = in_window(rows, calibration_start, date(year - 3, 1, 1)) policy = in_window(rows, policy_start, date(year - 1, 1, 1)) evaluation = in_window(rows, evaluation_start, date(year + 1, 1, 1)) require_ready(calibration, policy_start, "calibration") require_ready(policy, evaluation_start, "policy") if not evaluation: raise ValueError("Empty evaluation window") # Evaluation membership uses decision dates, not eventual outcomes. # Outcome availability is checked separately when reporting scores. return Fold(year, train, calibration, policy, evaluation, len(candidates) - len(train)) def month_end(year, month): return date(year, month, calendar.monthrange(year, month)[1]) def example_calendar(): rows = [] for year in range(2005, 2016): for month in range(1, 13): decision = month_end(year, month) if decision < date(2005, 6, 1): continue label_end = month_end(year + 1, month) rows.append(Observation(decision.isoformat(), decision, label_end, label_end)) return tuple(rows) class BoundaryChecks(unittest.TestCase): def test_strict_boundary_and_reporting_delay(self): boundary = date(2011, 1, 1) mature = Observation("mature", date(2009, 12, 31), date(2010, 12, 31), date(2010, 12, 31)) delayed = replace(mature, key="delayed", label_available_at=date(2011, 1, 15)) equal_end = Observation("equal-end", date(2010, 1, 1), boundary, boundary) equal_available = replace(mature, key="equal-available", label_available_at=boundary) unresolved = replace(mature, key="unresolved", label_available_at=None) overlaps = Observation("overlaps", date(2010, 1, 31), date(2011, 1, 31), date(2011, 1, 31)) self.assertEqual(eligible_before( [mature, delayed, equal_end, equal_available, unresolved, overlaps], boundary), (mature,)) def test_calendar_counts_and_boundaries(self): rows = example_calendar() for year, count in [(2013, 31), (2014, 43), (2015, 55)]: fold = make_fold(rows, year) self.assertEqual(len(fold.train), count) self.assertEqual(fold.train_excluded, 12) for name in ("calibration", "policy", "evaluation"): self.assertEqual(len(getattr(fold, name)), 12) for name, boundary in [("train", date(year - 4, 1, 1)), ("calibration", date(year - 2, 1, 1)), ("policy", date(year, 1, 1))]: require_ready(getattr(fold, name), boundary, name) keys = [row.key for name in ("train", "calibration", "policy", "evaluation") for row in getattr(fold, name)] self.assertEqual(len(keys), len(set(keys))) def test_evaluation_outcomes_do_not_select_evaluation_rows(self): rows = example_calendar() changed = tuple(replace(row, label_available_at=None) if row.decision_time.year == 2013 else row for row in rows) before = make_fold(rows, 2013) after = make_fold(changed, 2013) self.assertEqual(before.train, after.train) self.assertEqual(before.calibration, after.calibration) self.assertEqual(before.policy, after.policy) self.assertEqual([r.key for r in before.evaluation], [r.key for r in after.evaluation]) with self.assertRaisesRegex(ValueError, "outcomes unavailable"): require_ready(after.evaluation, date(2015, 1, 1), "evaluation") def test_incomplete_calibration_and_policy_fail(self): for year, name in [(2009, "calibration"), (2011, "policy")]: rows = tuple(replace(row, label_available_at=None) if row.decision_time.year == year else row for row in example_calendar()) with self.assertRaisesRegex(ValueError, name): make_fold(rows, 2013) def test_evaluation_report_waits_for_last_outcome(self): fold = make_fold(example_calendar(), 2013) with self.assertRaisesRegex(ValueError, "outcomes unavailable"): require_ready(fold.evaluation, date(2014, 12, 31), "evaluation") require_ready(fold.evaluation, date(2015, 1, 1), "evaluation") def test_panel_uses_dates_not_row_counts(self): rows = tuple(replace(row, key=f"{loan}:{row.key}") for row in example_calendar() for loan in ("a", "b", "c")) fold = make_fold(rows, 2013) self.assertEqual(len(fold.train), 31 * 3) self.assertEqual(fold.train_excluded, 12 * 3) self.assertEqual(len(fold.evaluation), 12 * 3) def test_bad_metadata_and_empty_windows_fail(self): rows = example_calendar() with self.assertRaisesRegex(ValueError, "Duplicate"): make_fold(rows + rows[:1], 2013) with self.assertRaisesRegex(ValueError, "precedes"): make_fold((replace(rows[0], label_available_at=rows[0].decision_time),), 2013) with self.assertRaisesRegex(ValueError, "Non-forward"): make_fold((replace(rows[0], label_end=rows[0].decision_time),), 2013) with self.assertRaisesRegex(ValueError, "Empty training"): make_fold((), 2013) with self.assertRaisesRegex(ValueError, "Empty evaluation"): make_fold(tuple(r for r in rows if r.decision_time.year < 2013), 2013) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--check", action="store_true", help="Run boundary checks") parser.add_argument("--out", type=Path, help="Write membership to a NEW CSV") args = parser.parse_args() if args.check: suite = unittest.defaultTestLoader.loadTestsFromTestCase(BoundaryChecks) result = unittest.TextTestRunner(verbosity=2).run(suite) if not result.wasSuccessful(): raise SystemExit(1) return rows = example_calendar() folds = [make_fold(rows, year) for year in (2013, 2014, 2015)] print("Illustrative calendar rows, not loans or measured model results.") print("eval_year train excluded calibration policy evaluation report_before") for fold in folds: report_before = date(fold.year + 2, 1, 1) require_ready(fold.evaluation, report_before, "evaluation") print(f"{fold.year:<11}{len(fold.train):<7}{fold.train_excluded:<10}" f"{len(fold.calibration):<13}{len(fold.policy):<8}" f"{len(fold.evaluation):<12}{report_before}") if args.out: with args.out.open("x", encoding="utf-8", newline="") as stream: writer = csv.writer(stream) writer.writerow(["fold", "role", "key", "decision_time", "label_end", "label_available_at"]) for fold in folds: for role in ("train", "calibration", "policy", "evaluation"): for row in getattr(fold, role): writer.writerow([fold.year, role, row.key, row.decision_time, row.label_end, row.label_available_at]) print(f"Wrote {args.out}") if __name__ == "__main__": main()