"""Synthetic accounting examples. No market data, network access, or broker orders. Python 3.10+. See README.md for assumptions, limitations, and commands. Copyright 2026 Vesper Applied LLC. Distributed under the included MIT license. """ import argparse import csv from datetime import date from decimal import Decimal, InvalidOperation, localcontext from fractions import Fraction import io import json from pathlib import Path import sys FIELDS = ("case", "date", "shares", "close", "share_value", "receivable", "cash", "fees", "equity", "index_level", "order_status") CASE_KEYS = {"name", "initial_shares", "initial_cash", "previous_close", "fee_rate", "days"} DAY_KEYS = {"date", "open", "close", "split", "dividend", "cash_available", "action", "notional"} def number(value, label, positive=False): """Require finite, nonnegative decimal strings, then keep exact fractions.""" try: if not isinstance(value, str): raise ValueError(f"{label} must be a decimal string") decimal = Decimal(value) if not decimal.is_finite(): raise ValueError(f"{label} must be finite") result = Fraction(decimal) except (InvalidOperation, TypeError) as error: raise ValueError(f"Invalid number for {label}") from error if result < 0 or (positive and result == 0): raise ValueError(f"{label} must be {'positive' if positive else 'nonnegative'}") return result def event_date(value): if not isinstance(value, str): raise ValueError("Dates must use YYYY-MM-DD strings") parsed = date.fromisoformat(value) if parsed.isoformat() != value: raise ValueError("Dates must use YYYY-MM-DD") return parsed def run_case(case): """Return close-of-session records for one isolated synthetic security.""" if not isinstance(case, dict) or set(case) != CASE_KEYS: raise ValueError("Scenario fields must match the documented format") if not isinstance(case["name"], str) or not case["name"].strip(): raise ValueError("A scenario needs a name") if not isinstance(case["days"], list) or not case["days"]: raise ValueError("A scenario needs at least one session") shares = number(case["initial_shares"], "initial_shares") cash = number(case["initial_cash"], "initial_cash") previous_close = number(case["previous_close"], "previous_close", positive=True) fee_rate = number(case["fee_rate"], "fee_rate") if fee_rate >= 1: raise ValueError("fee_rate must be less than 1") claims = [] # Fixed cash amounts paired with explicitly declared availability dates. previous_date = None index_level = Fraction(100) records = [] for session in case["days"]: if not isinstance(session, dict) or set(session) - DAY_KEYS or not {"date", "open", "close"} <= set(session): raise ValueError("Unknown or missing session fields") current_date = event_date(session["date"]) if previous_date is not None and current_date <= previous_date: raise ValueError("Session dates must be strictly increasing") open_price = number(session["open"], "open", positive=True) close = number(session["close"], "close", positive=True) split = number(session.get("split", "1"), "split", positive=True) dividend = number(session.get("dividend", "0"), "dividend") available = event_date(session["cash_available"]) if "cash_available" in session else None if dividend and (available is None or available < current_date): raise ValueError("A dividend requires cash_available on or after its ex-date") if not dividend and available is not None: raise ValueError("cash_available belongs on a nonzero dividend event") action = session.get("action", "hold") if action not in ("hold", "sell_all", "buy_cash", "buy_notional"): raise ValueError("Unknown action") if (action == "buy_notional") != ("notional" in session): raise ValueError("Only buy_notional requires a notional amount") requested = number(session["notional"], "notional", positive=True) if action == "buy_notional" else None # Before the open: split carried shares, establish entitlement, deliver cash. shares *= split if dividend and shares: claims.append((available, shares * dividend)) cash += sum((amount for due, amount in claims if due <= current_date), Fraction(0)) claims = [(due, amount) for due, amount in claims if due > current_date] fees = Fraction(0) status = "no_order" if action == "sell_all" and shares: proceeds = shares * open_price fees = proceeds * fee_rate cash += proceeds - fees shares = Fraction(0) status = "filled" elif action in ("buy_cash", "buy_notional"): notional = cash / (1 + fee_rate) if action == "buy_cash" else requested cost = notional * (1 + fee_rate) if cost > cash: status = "rejected_insufficient_cash" elif notional: fees = notional * fee_rate cash -= cost shares += notional / open_price status = "filled" # The benchmark notionally reinvests on the ex-date close, regardless of cash delivery. index_level *= split * (close + dividend) / previous_close receivable = sum((amount for _, amount in claims), Fraction(0)) share_value = shares * close records.append(dict(case=case["name"], date=current_date.isoformat(), shares=shares, close=close, share_value=share_value, receivable=receivable, cash=cash, fees=fees, equity=cash + share_value + receivable, index_level=index_level, order_status=status)) previous_close, previous_date = close, current_date return records def csv_text(records): """Round only the displayed output, to four decimal places.""" output = io.StringIO(newline="") writer = csv.DictWriter(output, fieldnames=FIELDS, lineterminator="\n") writer.writeheader() for row in records: formatted = {} for key, value in row.items(): if isinstance(value, Fraction): with localcontext() as context: context.prec = max(50, len(str(abs(value.numerator))) + len(str(value.denominator)) + 10) formatted[key] = format(Decimal(value.numerator) / Decimal(value.denominator), ".4f") else: formatted[key] = value writer.writerow(formatted) return output.getvalue() def main(): here = Path(__file__).resolve().parent parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", type=Path, default=here / "scenarios.json") parser.add_argument("--check", action="store_true", help="Compare output with the supplied expected.csv") args = parser.parse_args() try: cases = json.loads(args.input.read_text(encoding="utf-8")) if not isinstance(cases, list) or not cases: raise ValueError("Input must be a nonempty list of scenarios") records = [row for case in cases for row in run_case(case)] output = csv_text(records) if args.check: if output != (here / "expected.csv").read_text(encoding="utf-8"): parser.exit(1, "Output differs from expected.csv. Modified inputs need their own reviewed expectations.\n") print(f"PASS: {len(cases)} synthetic scenarios, {len(records)} rows match expected.csv.") else: sys.stdout.write(output) except (OSError, ValueError, KeyError, TypeError) as error: parser.error(str(error)) if __name__ == "__main__": main()