"""Stdlib-only tests for safe wrapper argv and observed-outcome audits. Run with a Python installation: python -m unittest discover -s ML -p test_legacy_entrypoints.py These tests never train, connect to a broker, or install packages. """ import contextlib import csv import io import json from pathlib import Path import tempfile import unittest from unittest.mock import patch import snapshot_dataset import snapshot_export_onnx import snapshot_train class LegacyEntrypointTests(unittest.TestCase): def test_training_command_is_argument_vector_with_symbols(self): command = snapshot_train.build_command("USDCHF, EURUSD", "artifacts/universal") self.assertEqual(command[0], "powershell.exe") self.assertIn("Build-MarketModel.ps1", command[5]) self.assertEqual(command[-2:], ["-Symbols", "USDCHF,EURUSD"]) def test_empty_duplicate_and_control_character_symbols_rejected(self): for value in ("", "USDCHF,,EURUSD", "USDCHF,usdchf", "USDCHF,\nEURUSD"): with self.subTest(value=value), self.assertRaises(ValueError): snapshot_train.build_command(value, "output") def test_dry_run_does_not_launch_training(self): output = io.StringIO() with patch.object(snapshot_train.subprocess, "run") as run, contextlib.redirect_stdout(output): self.assertEqual(snapshot_train.main(["--dry-run", "--symbols", "USDCHF"]), 0) run.assert_not_called() self.assertIn("-Symbols", json.loads(output.getvalue())) def test_status_training_argument_is_not_supported(self): with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as error: snapshot_train.main(["--label_mode", "status"]) self.assertEqual(error.exception.code, 2) def test_old_label_and_feature_apis_fail_explicitly(self): for function in (snapshot_dataset.label_from_status, snapshot_dataset.labels_from_future_prices, snapshot_dataset.load_snapshot_dataframe, snapshot_dataset.build_feature_matrix): with self.subTest(function=function.__name__), self.assertRaisesRegex(RuntimeError, "retired"): function(None) def test_export_uses_single_native_exporter(self): command = snapshot_export_onnx.build_command("input directory", "output directory") self.assertIn("Export-MarketModel.ps1", command[5]) self.assertEqual(command[6], "-InputDirectory") self.assertEqual(command[8], "-OutputDirectory") def test_legacy_pickle_override_is_rejected(self): with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as error: snapshot_export_onnx.main(["--artifacts", "old", "--scaler", "scaler.pkl"]) self.assertEqual(error.exception.code, 2) def test_unicode_snapshots_are_inspected_without_labels(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "feature_batch_test.txt" path.write_text("timestamp:2025.01.01 00:00:00\nsymbol:USDCHF\nstatus:executed\n", encoding="utf-16") report = snapshot_dataset.inspect_snapshots(directory) self.assertEqual(report["records"], 1) self.assertEqual(report["symbols"], ["USDCHF"]) self.assertFalse(report["training_supported"]) self.assertNotIn("label", snapshot_dataset.parse_snapshot_file(path)) def test_duplicate_snapshot_fields_are_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "feature_batch_bad.txt" path.write_text("symbol:USDCHF\nsymbol:EURUSD\n", encoding="utf-8") with self.assertRaisesRegex(ValueError, "duplicate"): snapshot_dataset.parse_snapshot_file(path) def _write_market_fixture(self, path, mutation=None): features = list(snapshot_dataset.MARKET_BASE_FEATURES + snapshot_dataset.ACTIVE_FEATURES) fields = ["decision_time", "target_end", "split", "symbol", "direction", "label", "net_log_return", "entry_spread_points", "exit_spread_points", *features] rows = [] for split, month in (("train", "01"), ("validation", "07"), ("test", "10")): for direction in (0, 1): rows.append(dict(decision_time=f"2025.{month}.02 00:00", target_end=f"2025.{month}.02 04:00", split=split, symbol="USDCHF", direction=direction, label=direction, net_log_return=0.001 if direction else -0.001, entry_spread_points=7, exit_spread_points=8, **{feature: 0.0 for feature in features})) for row in rows: row["mm_direction"] = 1.0 if row["direction"] == 0 else -1.0 row["active_symbol_flag"] = 1.0 if mutation: mutation(rows) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows(rows) def test_market_audit_accepts_observed_paired_purged_rows(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "dataset.csv" self._write_market_fixture(path) report = snapshot_dataset.audit_market_dataset(path) self.assertTrue(report["audit_passed"]) self.assertEqual(report["rows"], 6) def test_unknown_labels_are_rejected_never_filled(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "dataset.csv" self._write_market_fixture(path, lambda rows: rows[0].update(label="")) with self.assertRaisesRegex(ValueError, "Unknown/malformed"): snapshot_dataset.audit_market_dataset(path) def test_split_boundaries_inferred_from_sibling_model_metadata(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "dataset.csv" self._write_market_fixture(path) content = path.read_text(encoding="utf-8") content = content.replace("2025.01.02", "2026.01.02").replace("2025.07.02", "2026.07.02").replace("2025.10.02", "2026.09.02") path.write_text(content, encoding="utf-8") metadata = {"training_end": "2026.07.01", "validation_end": "2026.09.01", "test_end": "2026.10.01"} (Path(directory) / "model_data.json").write_text(json.dumps(metadata), encoding="utf-8") report = snapshot_dataset.audit_market_dataset(path) self.assertTrue(report["audit_passed"]) self.assertEqual(report["split_boundaries"], metadata) def test_explicit_boundaries_override_supplied_metadata_file(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "dataset.csv" self._write_market_fixture(path) metadata = Path(directory) / "alternate-model.json" metadata.write_text(json.dumps({"training_end": "2026.07.01", "validation_end": "2026.09.01", "test_end": "2026.10.01"}), encoding="utf-8") report = snapshot_dataset.audit_market_dataset(path, "2025.07.01", "2025.10.01", "2026.01.01", model_data=metadata) self.assertTrue(report["audit_passed"]) self.assertEqual(report["split_boundaries"]["training_end"], "2025.07.01") def test_partial_metadata_does_not_silently_use_legacy_dates(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "dataset.csv" self._write_market_fixture(path) (Path(directory) / "model_data.json").write_text('{"training_end":"2026.07.01"}', encoding="utf-8") with self.assertRaisesRegex(ValueError, "validation_end"): snapshot_dataset.audit_market_dataset(path) def test_horizon_crossing_split_boundary_is_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "dataset.csv" self._write_market_fixture(path, lambda rows: rows[0].update(target_end="2025.07.01 00:00")) with self.assertRaisesRegex(ValueError, "split boundary"): snapshot_dataset.audit_market_dataset(path) def test_duplicate_symbol_time_direction_is_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "dataset.csv" self._write_market_fixture(path, lambda rows: rows[1].update(rows[0])) with self.assertRaisesRegex(ValueError, "Duplicate"): snapshot_dataset.audit_market_dataset(path) if __name__ == "__main__": unittest.main()