#!/usr/bin/env python3 """Check a published fictional cashflow example using only two local public files. Python standard library only; no application imports, network, or input changes. Amounts are in 10,000 JPY. This is a fixed-example checker, not financial advice. """ import argparse import csv import hashlib import io import json import sys from decimal import Decimal, InvalidOperation, localcontext from pathlib import Path CSV_SHA256 = "f50e35c218f1131a6539fcd019fd019d1fadc42f7f3203af57ff00e800982d6d" JSON_SHA256 = "8c94c26126d72122d19f5f5894ad020577dbc81bf65b4331bbac78863cf72cdf" TOLERANCE = Decimal("0.0000001") # 0.001 JPY; absolute only, before display rounding MODEL = "housing-note-1.0 / housing-income-timeline-1.0" HEADER = ["区分", "項目・時点", "単位", "経過月", "共通の値", "予定の期間で時短を終える", "時短の終了が遅れる", "10年の終わりまで時短を続ける", "補足"] FIELDS = { "currentIncome": ("いまの世帯手取り", "万円 / 月"), "shortStartMonth": ("時短を始める時期", "か月後"), "cash": ("現在の現預金", "万円"), "investments": ("必要時に換金する資産", "万円"), "reserve": ("必ず残しておきたい資金", "万円"), "price": ("候補の住宅価格", "万円"), "downPayment": ("頭金", "万円"), "closingCosts": ("購入時の諸費用・引越し費用", "万円"), "interestRate": ("比較する住宅ローン金利", "% / 年"), "mortgageYears": ("ローン返済期間", "年"), "livingCosts": ("家賃を除く生活費", "万円 / 月"), "ownershipCosts": ("購入後の維持費", "万円 / 月"), } OPTIONS = { "shortIncome": ("時短中の世帯手取り", "万円 / 月"), "durationMonths": ("予定している時短の期間", "か月"), "returnIncome": ("時短を終えた後の世帯手取り", "万円 / 月"), "delayMonths": ("終了が遅れる場合の追加期間", "か月"), } CASES = ("planned", "delayed", "continued") def number(value): if value is None or isinstance(value, bool): raise ValueError("required numeric value is missing or boolean") try: result = Decimal(str(value)) except (InvalidOperation, ValueError): raise ValueError("invalid numeric value") from None if not result.is_finite(): raise ValueError("numeric value must be finite") return result def integer(value): value = number(value) if value != value.to_integral_value() or value < 0: raise ValueError("month/count must be a nonnegative integer") return int(value) def one(rows, section, label): found = [row for row in rows if row[0] == section and row[1] == label] if len(found) != 1: raise ValueError("required row is absent or duplicated: " + label) return found[0] def unique_case(items, case_id): found = [item for item in items if item["id"] == case_id] if len(found) != 1: raise ValueError("required JSON case is absent or duplicated: " + case_id) return found[0] def parse_csv(raw): rows = list(csv.reader(io.StringIO(raw.decode("utf-8-sig"), newline=""))) if not rows or rows[0] != HEADER or any(len(row) != 9 for row in rows): raise ValueError("unsupported CSV header or row width") if one(rows, "概要", "計算版")[4] != MODEL: raise ValueError("unsupported model version") if integer(one(rows, "概要", "比較期間")[4]) != 120: raise ValueError("this checker supports the published 120-month example only") inputs = {} for section, fields in [("反映済みの住宅・家計条件", FIELDS), ("反映済みの期間・手取り", OPTIONS)]: for key, (label, unit) in fields.items(): row = one(rows, section, label) if row[2] != unit: raise ValueError("unsupported unit: " + label) inputs[key] = number(row[4]) for key in ["shortStartMonth", "durationMonths", "delayMonths", "mortgageYears"]: integer(inputs[key]) if inputs["shortStartMonth"] > 120 or inputs["mortgageYears"] <= 0: raise ValueError("unsupported month or mortgage term") if any(value < 0 for value in inputs.values()) or inputs["downPayment"] > inputs["price"]: raise ValueError("unsupported negative value or down payment") ledger = [row for row in rows if row[0] == "月次残高"] if [integer(row[3]) for row in ledger] != list(range(121)): raise ValueError("ledger must contain each month 0 through 120 exactly once in order") if any(row[2] != "万円" for row in ledger): raise ValueError("ledger unit must be 10,000 JPY") return rows, inputs, ledger def direct_balances(inputs, end_month): """Closed form: no reuse of the application's sequential cash updates.""" principal = inputs["price"] - inputs["downPayment"] term = integer(inputs["mortgageYears"]) * 12 rate = inputs["interestRate"] / Decimal(1200) if rate == 0: mortgage = principal / term else: growth = (1 + rate) ** term mortgage = principal * rate * growth / (growth - 1) initial = inputs["cash"] + inputs["investments"] - inputs["downPayment"] - inputs["closingCosts"] start = integer(inputs["shortStartMonth"]) out = [] for month in range(121): before = min(month, start) reduced = max(0, min(month, end_month) - start) returned = max(0, month - end_month) total_income = (inputs["currentIncome"] * before + inputs["shortIncome"] * reduced + inputs["returnIncome"] * returned) out.append(initial + total_income - month * (inputs["livingCosts"] + inputs["ownershipCosts"]) - min(month, term) * mortgage) return mortgage, out def verify(csv_bytes, evidence_bytes): rows, inputs, ledger = parse_csv(csv_bytes) evidence = json.loads(evidence_bytes.decode("utf-8"), parse_float=Decimal) if evidence["edition"] != "housing-content-checks-v1" or evidence["schemaVersion"] != 1: raise ValueError("unsupported evidence edition") calc = evidence["calculation"] if number(calc["absoluteToleranceManYen"]) != TOLERANCE: raise ValueError("published tolerance differs from this fixed checker") failures = [] checks = [] def compare(label, expected, observed): error = abs(number(expected) - number(observed)) checks.append((label, error)) if error > TOLERANCE: failures.append({"check": label, "absoluteErrorManYen": str(error)}) def exact(label, expected, observed): if expected != observed: failures.append({"check": label, "expected": expected, "observed": observed}) csv_hash = hashlib.sha256(csv_bytes).hexdigest() json_hash = hashlib.sha256(evidence_bytes).hexdigest() exact("fixed CSV identity", CSV_SHA256, csv_hash) exact("fixed JSON identity", JSON_SHA256, json_hash) exact("JSON CSV reference", CSV_SHA256, calc["publicCsvSHA256"]) exact("JSON CSV path", "/examples/housing-income-32-30-months.csv", calc["publicCsv"]) # These 12 fields are common to the basic sample and this worked CSV. # JSON input.shortIncome=42 belongs to another example; never substitute it for CSV's 32. for key in FIELDS: compare("shared input: " + key, inputs[key], calc["input"][key]) summaries = unique_case(calc["workedResults"], "base")["cases"] summary_rows = {name: one(rows, "結果", name) for name in [ "終了後の手取りが始まる月", "10年間の最低資金", "最低になる時点", "残したい額を最初に下回る時点", "10年後の資金"]} start = integer(inputs["shortStartMonth"]) duration = integer(inputs["durationMonths"]) delay = integer(inputs["delayMonths"]) ends = (min(120, start + duration), min(120, start + duration + delay), 120) result_cases = [] for col, case_id, end in zip(range(5, 8), CASES, ends): mortgage, balances = direct_balances(inputs, end) if case_id == "planned": compare("monthly mortgage", mortgage, one(rows, "計算条件", "毎月のローン返済額")[4]) for month, balance in enumerate(balances): compare(case_id + " month " + str(month), balance, ledger[month][col]) minimum = min(balances) minimum_month = balances.index(minimum) # earliest exact minimum first_below = next((month for month, value in enumerate(balances) if value < inputs["reserve"]), None) saved = unique_case(summaries, case_id) for key, value, label in [("minimum", minimum, "10年間の最低資金"), ("final", balances[-1], "10年後の資金")]: compare(case_id + " CSV " + key, value, summary_rows[label][col]) compare(case_id + " JSON " + key, value, saved[key]) exact(case_id + " CSV minimumMonth", minimum_month, integer(summary_rows["最低になる時点"][col])) exact(case_id + " JSON minimumMonth", minimum_month, integer(saved["minimumMonth"])) below_text = summary_rows["残したい額を最初に下回る時点"][col] csv_below = None if below_text == "入力条件では下回らない" else integer(below_text) json_below = None if saved["firstBelow"] is None else integer(saved["firstBelow"]) exact(case_id + " CSV firstBelow", first_below, csv_below) exact(case_id + " JSON firstBelow", first_below, json_below) begins = summary_rows["終了後の手取りが始まる月"][col] csv_return = None if begins == "比較期間内にはなし" else integer(begins) exact(case_id + " CSV return month", None if end == 120 else end + 1, csv_return) result_cases.append({"id": case_id, "minimumManYen": str(minimum), "minimumMonth": minimum_month, "firstBelowReserveMonth": first_below, "finalManYen": str(balances[-1])}) monthly_checks = [error for name, error in checks if " month " in name] return { "checkerVersion": "public-housing-fixed-example-checker-v1", "status": "PASS_FOR_SUPPORTED_FIXED_EXAMPLE" if not failures else "FAIL", "inputs": {"csvSHA256": csv_hash, "evidenceJSONSHA256": json_hash, "pinnedCSVMatches": csv_hash == CSV_SHA256, "pinnedJSONMatches": json_hash == JSON_SHA256}, "method": "50-digit Decimal annuity and cumulative-income closed form; no application imports", "model": MODEL, "amountUnit": "10,000 JPY", "decimalPrecision": 50, "absoluteToleranceManYen": str(TOLERANCE), "relativeTolerance": 0, "monthlyValuesChecked": len(monthly_checks), "cases": result_cases, "maximumMonthlyAbsoluteErrorManYen": str(max(monthly_checks)), "numericComparisonsIncludingInputsAndSummaries": len(checks), "maximumNumericAbsoluteErrorManYen": str(max(error for _, error in checks)), "annualPayment": {"status": "NOT_CHECKED_MISSING_INPUTS", "monthlyValuesChecked": 0, "reason": "CSV and JSON do not specify monthly allocated expense, payment amount, first payment month or repeat interval. The published annual minimum is an output, not sufficient input."}, "notChecked": ["Other five worked examples and basic/stressed models without their complete public monthly ledgers", "Annual-payment adjustment; no missing schedule was guessed", "Current production behavior, browser UI, saving, PDF, tax, benefits, actual households or lending decisions"], "failures": failures, } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--csv", type=Path, required=True, help="downloaded fixed-example CSV") parser.add_argument("--evidence", type=Path, required=True, help="downloaded published evidence JSON") args = parser.parse_args() try: with localcontext() as ctx: ctx.prec = 50 result = verify(args.csv.read_bytes(), args.evidence.read_bytes()) except (OSError, ValueError, KeyError, TypeError, InvalidOperation, csv.Error): print(json.dumps({"status": "INVALID_INPUT", "message": "Required fixed files, fields, units or row structure are missing or invalid; no defaults were inserted."}, ensure_ascii=False)) return 2 print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result["status"] == "PASS_FOR_SUPPORTED_FIXED_EXAMPLE" else 1 if __name__ == "__main__": sys.exit(main())