#!/usr/bin/env python3 """Replay a constructed missing-versus-zero comparison using Python's stdlib. From the website root: python3 scripts/evidence/missing_data.py --check Standalone: python3 replay.py --source energydata_complete.csv --output . --check No network, model, product API, or third-party package is used by this script. """ from __future__ import annotations import argparse import csv import hashlib import io import json import sys import unittest from datetime import datetime, timedelta from pathlib import Path SOURCE_SHA256 = "2820bf712ad0275cb18b85a05250926100d8e65ebb9f4d2d016ca91ea152a25d" START = "2016-01-12 00:00:00" END = "2016-01-12 02:00:00" CHANGED = (4, 5, 6, 7) EXPECTED_VALUES = (40, 30, 40, 50, 310, 380, 380, 370, 120, 50, 40, 50) FIELDS = ["source_row", "timestamp", "Appliances_Wh", "status"] def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def require(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def summarize(values: list[int | None]) -> dict: """Null is unavailable; zero is available. Never infer a missing value.""" available = [value for value in values if value is not None] expected, count = len(values), len(available) total = sum(available) mean = total / count if count else None return { "expected_readings": expected, "available_readings": count, "missing_readings": expected - count, "coverage": { "numerator": count, "denominator": expected, "fraction": count / expected if expected else None, "percent": round(100 * count / expected, 6) if expected else None, }, "available_sum_Wh": total, "available_mean_Wh": { "numerator_Wh": total, "denominator_readings": count, "value": mean, }, "full_window_total_Wh": total if expected and count == expected else None, "full_window_mean_Wh": mean if expected and count == expected else None, } def load_window(source: Path) -> list[dict]: data = source.read_bytes() require(sha256(data) == SOURCE_SHA256, "Source SHA-256 does not match the published input.") original = list(csv.DictReader(io.StringIO(data.decode("utf-8")))) first = datetime.fromisoformat(original[0]["date"]) first_midnight = first.replace(hour=0, minute=0, second=0) if first != first_midnight: first_midnight += timedelta(days=1) require(first_midnight.isoformat(sep=" ") == START, "First complete source day changed.") rows = [ {"timestamp": row["date"], "value_Wh": int(row["Appliances"]), "source_row": index + 2, "status": "source_observation"} for index, row in enumerate(original) if START <= row["date"] < END ] expected_times = [first_midnight + timedelta(minutes=10 * index) for index in range(12)] require([row["timestamp"] for row in rows] == [time.isoformat(sep=" ") for time in expected_times], "Window must contain each expected timestamp once, in order.") require(tuple(row["value_Wh"] for row in rows) == EXPECTED_VALUES, "Independent known source-window values do not match.") require([row["source_row"] for row in rows] == list(range(44, 56)), "Source row mapping changed.") return rows def scenario(rows: list[dict], kind: str) -> dict: require(kind in {"original", "missing", "zero"}, "Unknown scenario.") samples = [dict(row) for row in rows] for index in CHANGED: if kind == "missing": samples[index].update(value_Wh=None, status="constructed_missing") elif kind == "zero": samples[index].update(value_Wh=0, status="constructed_zero") return { "label": {"original": "Original observations", "missing": "Four readings missing", "zero": "Four constructed zeroes"}[kind], "constructed": kind != "original", **summarize([row["value_Wh"] for row in samples]), "synthetic_zero_readings": len(CHANGED) if kind == "zero" else 0, "samples": samples, } def build_results(rows: list[dict]) -> dict: scenarios = {kind: scenario(rows, kind) for kind in ("original", "missing", "zero")} return { "schema": "reific.missing-data-versus-zero.v1", "title": "Missing sensor readings or a real drop?", "status": "deterministic_public_data_replay_with_constructed_scenarios", "verified_at": "2026-09-18", "agent_calls": 0, "source": { "dataset": "Appliances Energy Prediction", "citation": "Candanedo, L. (2017). Appliances Energy Prediction [Dataset]. UCI Machine Learning Repository.", "doi": "https://doi.org/10.24432/C5VC8G", "description_url": "https://archive.ics.uci.edu/dataset/374/appliances+energy+prediction", "archive_url": "https://archive.ics.uci.edu/static/public/374/appliances%2Benergy%2Bprediction.zip", "file": "energydata_complete.csv", "sha256": SOURCE_SHA256, "retrieved_at": "2026-09-15", "license": "CC BY 4.0", "license_url": "https://creativecommons.org/licenses/by/4.0/", "column": "Appliances", "unit": "Wh per 10-minute reading", "row_numbering": "1-based physical CSV lines; the header is line 1", "changes": "Select 12 rows and the timestamp/Appliances columns, normalize CSV serialization, then construct two separately labeled scenarios. The source file is unchanged.", }, "window": { "selection_rule": "First two hours of the first complete source day; illustrative window, not a representative sample.", "start_inclusive": START, "end_exclusive": END, "first_reading": rows[0]["timestamp"], "last_reading": rows[-1]["timestamp"], "cadence_seconds": 600, "expected_readings": 12, "timezone": "Not supplied; timestamps are retained as provided.", "interval_label_convention": "The source does not state whether timestamps label interval starts or ends.", }, "construction": { "rule": "Change only the middle third of the 12 timestamped values: blank in the missing scenario, numeric zero in the zero scenario.", "changed_indices_zero_based": list(CHANGED), "changed_timestamps": [rows[index]["timestamp"] for index in CHANGED], "original_values_Wh": [rows[index]["value_Wh"] for index in CHANGED], "original_changed_sum_Wh": 1440, "unchanged_source_readings": 8, "is_real_outage": False, }, "scenarios": scenarios, "naive_zero_fill": { "description": "Imputing zero for each missing value reproduces the numeric zero scenario, but does not turn missing readings into observations.", "source_available_readings": 8, "imputed_readings": 4, "actual_observation_coverage_fraction": 8 / 12, "sum_Wh": 420, "denominator_readings": 12, "mean_Wh": 35.0, "difference_from_available_mean_Wh": -17.5, "relative_change_from_available_mean_percent": -100 / 3, "relative_change_formula": "(35 - 52.5) / 52.5 * 100", "is_observed_full_window_mean": False, }, "limits": [ "The missing readings and zeroes are constructed by Reific. Neither altered scenario is an observed equipment or telemetry outage.", "This is one illustrative window in one public household dataset, not a frequency estimate or a representative sample of operational failures.", "The omitted readings are the high values in this window; missingness is not random. The 52.5 Wh available-only mean does not estimate the original full-window mean of 155 Wh without further assumptions.", "Coverage is available values divided by 12 expected timestamps, under the assumed known 10-minute cadence. It is not an uptime, device-health, or value-validity metric.", "Even a reported numeric zero in a real system would require telemetry and device context before concluding a physical outage; these zeroes are synthetic.", "Energy is measured in Wh, not W. Means are Wh per reading; timestamps do not specify timezone or interval-start/end convention.", "No model, detector, agent, product performance, customer outcome, or causal attribution is evaluated. Ordinary code can reproduce these calculations.", ], } def csv_bytes(samples: list[dict]) -> bytes: output = io.StringIO(newline="") writer = csv.DictWriter(output, FIELDS, lineterminator="\n") writer.writeheader() for row in samples: writer.writerow({"source_row": row["source_row"], "timestamp": row["timestamp"], "Appliances_Wh": row["value_Wh"], "status": row["status"]}) return output.getvalue().encode("utf-8") def artifacts(results: dict) -> dict[str, bytes]: return { "source-window.csv": csv_bytes(results["scenarios"]["original"]["samples"]), "missing.csv": csv_bytes(results["scenarios"]["missing"]["samples"]), "zero.csv": csv_bytes(results["scenarios"]["zero"]["samples"]), "results.json": (json.dumps(results, indent=2, ensure_ascii=False, allow_nan=False) + "\n").encode(), "replay.py": Path(__file__).read_bytes(), } class SanityChecks(unittest.TestCase): """Independent small fixtures and hand-checked invariants, not model scores.""" def test_zero_is_an_observation_and_null_is_not(self) -> None: result = summarize([10, 0, None, 30]) self.assertEqual((result["available_readings"], result["missing_readings"]), (3, 1)) self.assertEqual(result["available_sum_Wh"], 40) self.assertEqual(result["available_mean_Wh"]["denominator_readings"], 3) self.assertAlmostEqual(result["available_mean_Wh"]["value"], 40 / 3) self.assertEqual(result["coverage"]["fraction"], 0.75) self.assertIsNone(result["full_window_total_Wh"]) def test_no_observations_does_not_create_a_mean(self) -> None: for values in ([None, None], []): result = summarize(values) self.assertIsNone(result["available_mean_Wh"]["value"]) self.assertIsNone(result["full_window_total_Wh"]) self.assertIsNone(result["full_window_mean_Wh"]) def test_complete_numeric_zeroes_have_full_coverage(self) -> None: result = summarize([0, 0, 0]) self.assertEqual(result["coverage"]["fraction"], 1) self.assertEqual(result["full_window_mean_Wh"], 0) self.assertEqual(result["full_window_total_Wh"], 0) def test_known_window_sums_and_denominators(self) -> None: rows = [{"value_Wh": value, "timestamp": str(index), "source_row": index + 44, "status": "source_observation"} for index, value in enumerate(EXPECTED_VALUES)] original, missing, zero = (scenario(rows, kind) for kind in ("original", "missing", "zero")) self.assertEqual(original["available_sum_Wh"], 1860) self.assertEqual(original["full_window_mean_Wh"], 155) self.assertEqual(missing["available_sum_Wh"], 420) self.assertEqual(missing["available_mean_Wh"], {"numerator_Wh": 420, "denominator_readings": 8, "value": 52.5}) self.assertIsNone(missing["full_window_mean_Wh"]) self.assertEqual(zero["available_mean_Wh"], {"numerator_Wh": 420, "denominator_readings": 12, "value": 35}) self.assertEqual(original["available_sum_Wh"] - missing["available_sum_Wh"], 1440) def test_only_selected_values_change_and_source_is_untouched(self) -> None: rows = [{"value_Wh": value, "timestamp": str(index), "source_row": index + 44, "status": "source_observation"} for index, value in enumerate(EXPECTED_VALUES)] before = json.dumps(rows) for kind in ("missing", "zero"): altered = scenario(rows, kind)["samples"] self.assertEqual(len(altered), 12) for index, (source, changed) in enumerate(zip(rows, altered)): self.assertEqual(source["timestamp"], changed["timestamp"]) self.assertEqual(source["source_row"], changed["source_row"]) if index not in CHANGED: self.assertEqual(source, changed) else: self.assertEqual(changed["value_Wh"], None if kind == "missing" else 0) self.assertEqual(json.dumps(rows), before) def test_csv_preserves_blank_versus_numeric_zero(self) -> None: samples = [{"source_row": 2, "timestamp": "x", "value_Wh": None, "status": "constructed_missing"}, {"source_row": 3, "timestamp": "y", "value_Wh": 0, "status": "constructed_zero"}] parsed = list(csv.DictReader(io.StringIO(csv_bytes(samples).decode()))) self.assertEqual([row["Appliances_Wh"] for row in parsed], ["", "0"]) def main() -> int: here = Path(__file__).resolve().parent repository_evidence = here.parents[1] / "public/evidence" evidence = here.parent if here.name == "missing-data-versus-zero" else repository_evidence parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", type=Path, default=evidence / "source/energydata_complete.csv") parser.add_argument("--output", type=Path, default=evidence / "missing-data-versus-zero") parser.add_argument("--check", action="store_true", help="Verify exact published data/script bytes without writing.") parser.add_argument("--self-test", action="store_true", help="Run independent fixtures only; no source file required.") args = parser.parse_args() checks = unittest.TextTestRunner(verbosity=1).run(unittest.defaultTestLoader.loadTestsFromTestCase(SanityChecks)) if not checks.wasSuccessful(): return 1 if args.self_test: return 0 try: generated = artifacts(build_results(load_window(args.source))) if args.check: for name, expected in generated.items(): require((args.output / name).read_bytes() == expected, f"Artifact bytes differ: {name}") print(f"Verified source SHA-256, known observations, and {len(generated)} deterministic data/script artifacts.") else: args.output.mkdir(parents=True, exist_ok=True) for name, content in generated.items(): (args.output / name).write_bytes(content) print(f"Wrote {len(generated)} deterministic data/script artifacts to {args.output}.") return 0 except (OSError, ValueError) as error: print(f"Replay failed: {error}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())