import copy import hashlib import json import sys import tempfile import unittest from pathlib import Path PROJECT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT / "src")) from quant_os.architecture import validate_architecture from quant_os.evidence.maturity import ( MaturityStandardError, _final_assessment_digest, _pre_canary_projection_digest, evaluate_assessment, load_json_object, validate_standard, ) PROFILE = PROJECT / "profiles/production-80/standard.json" ASSESSMENT = PROJECT / "profiles/production-80/current-assessment.json" class QuantOsMaturityTests(unittest.TestCase): def setUp(self): self.standard = load_json_object(PROFILE) self.assessment = load_json_object(ASSESSMENT) @staticmethod def _frozen_subject(): return { "candidate_id": "candidate-test-v1", "release_sha256": "1" * 64, "model_sha256": "2" * 64, "dataset_sha256": "3" * 64, "account_scope_id": "broker-account-pseudonym", "material_change_epoch": 0, } @staticmethod def _trusted_test_verifier(claim, control, source): trusted = ( claim.get("attestation") == {"method": "unit-test-trust-boundary"} and source.is_file() and claim.get("control_id") == control["id"] ) return { "trusted": trusted, "trust_environment": "test", "verifier_id": "unit-test-verifier", "issuer_id": "unit-test-issuer", "key_fingerprint_sha256": "9" * 64, "policy_version": "unit-test-policy-v1", } def _write_claim( self, root, *, control_id, tier, subject, evidence_type="control_acceptance_evidence", assessment_sha256=None, observed_as_of="2026-06-30", extra=None, ): claims = root / "claims" claims.mkdir(exist_ok=True) claim = { "schema_version": "quant-os-evidence-claim/1.0", "claim_id": f"claim-{control_id.lower()}", "control_id": control_id, "tier": tier, "subject": subject, "standard_id": self.standard["standard_id"], "standard_canonical_sha256": self.assessment[ "standard_canonical_sha256" ], "evidence_type": evidence_type, "authority_class": "unit_test_only", "observed_as_of": observed_as_of, "source_hashes": ["4" * 64], "attestation": {"method": "unit-test-trust-boundary"}, } if assessment_sha256 is not None: claim["assessment_sha256"] = assessment_sha256 if extra is not None: claim.update(extra) path = claims / f"{control_id.lower()}.json" path.write_text( json.dumps(claim, sort_keys=True), encoding="utf-8", ) return { "path": path.relative_to(root).as_posix(), "tier": tier, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } def _prepare_qualified_assessment( self, root, passing_controls, *, canary_days=10, attestation_kind="qualification", ): assessment = copy.deepcopy(self.assessment) subject = self._frozen_subject() assessment["subject"] = subject baseline_profile = load_json_object( PROJECT / "profiles/baseline-60/profile.json" ) baseline_root = root / "profiles" / "baseline-60" baseline_root.mkdir(parents=True) (baseline_root / "profile.json").write_text( json.dumps(baseline_profile, sort_keys=True), encoding="utf-8", ) baseline_verification = self._write_claim( root, control_id="BASELINE_60", tier="E3", subject=subject, evidence_type="baseline60_qualification_attestation", extra={ "baseline_profile_id": "quant-os-baseline-60", "baseline_profile_canonical_sha256": self.assessment[ "baseline_60" ]["profile_canonical_sha256"], "gates_passed": 10, "gates_total": 10, "score": 60, }, ) assessment["baseline_60"] = { "all_gates_passed": True, "gates_passed": 10, "gates_total": 10, "profile_id": "quant-os-baseline-60", "profile_canonical_sha256": self.assessment["baseline_60"][ "profile_canonical_sha256" ], "verification": baseline_verification, } assessment["observations"] = { "real_qmt_observed": True, "broker_programmatic_confirmation": True, "shadow_trading_days": 60, "canary_trading_days": canary_days, "open_p0": 0, "open_p1": 0, } for control in assessment["controls"]: if ( control["id"] in passing_controls and control["id"] != "D6-CANARY" ): standard_control = next( item for domain in self.standard["domains"] for item in domain["controls"] if item["id"] == control["id"] ) control["status"] = "passed" control["evidence"] = [ self._write_claim( root, control_id=control["id"], tier=standard_control["minimum_evidence_tier"], subject=subject, ) ] if attestation_kind not in {"qualification", "pre_canary"}: raise AssertionError(f"unknown attestation kind {attestation_kind}") assessment["pre_canary_attestation"] = self._write_claim( root, control_id="PRE_CANARY_AUTHORIZATION", tier="E3", subject=subject, evidence_type="pre_canary_authorization_attestation", assessment_sha256=_pre_canary_projection_digest(assessment), observed_as_of="2026-07-01", extra={ "authorization_id": ( "claim-pre_canary_authorization" ), "authorized_at": "2026-07-01", "expires_on": "2026-08-31", "account_scope_id": subject["account_scope_id"], "strategy_ids": [subject["candidate_id"]], "maximum_capital_cny": 100000, "maximum_gross_notional_cny": 80000, "maximum_symbols": 2, "symbol_allowlist": [ "000001.XSHE", "600000.XSHG", ], "flow_limits": { "maximum_submission_and_cancel_requests_per_second": 2, "maximum_submission_and_cancel_requests_per_day": 50, }, "stop_conditions": [ "any_risk_data_rule_or_reconciliation_breach", "authorization_expired", "operator_kill_switch", ], "no_automatic_scale_up": True, }, ) if "D6-CANARY" in passing_controls: canary_control = next( control for control in assessment["controls"] if control["id"] == "D6-CANARY" ) canary_control["status"] = "passed" canary_control["evidence"] = [ self._write_claim( root, control_id="D6-CANARY", tier="E4", subject=subject, evidence_type="controlled_live_canary_observation", observed_as_of="2026-07-15", extra={ "pre_canary_authorization_claim_id": ( "claim-pre_canary_authorization" ), "pre_canary_authorization_sha256": assessment[ "pre_canary_attestation" ]["sha256"], "canary_started_on": "2026-07-02", "canary_ended_on": "2026-07-15", "trading_days_observed": canary_days, "maximum_capital_cny_observed": 75000, "maximum_gross_notional_cny_observed": 60000, "symbols_observed": [ "000001.XSHE", "600000.XSHG", ], "maximum_submission_and_cancel_requests_per_second_observed": 2, "maximum_submission_and_cancel_requests_per_day_observed": 40, "automatic_scale_up_observed": False, "breach_events": [], "stop_events": [], }, ) ] if attestation_kind == "qualification": assessment["qualification_attestation"] = self._write_claim( root, control_id="QUALIFICATION", tier="E4", subject=subject, evidence_type="production80_promotion_attestation", assessment_sha256=_final_assessment_digest(assessment), observed_as_of="2026-07-16", ) return assessment def test_profile_is_exactly_weighted_and_current_claim_is_blocked(self): summary = validate_standard(self.standard) self.assertEqual(summary["maximum_score"], 100) self.assertEqual(summary["domain_weight_total"], 100) self.assertEqual(summary["qualification_threshold"], 80) self.assertEqual(summary["domain_count"], 7) self.assertEqual(summary["hard_gate_count"], 21) self.assertEqual(summary["hard_gate_weight_total"], 80) result = evaluate_assessment( self.standard, self.assessment, repository_root=PROJECT, ) self.assertEqual(result["raw_score"], 0) self.assertEqual(result["effective_score"], 0) self.assertFalse(result["baseline_60_passed"]) self.assertFalse(result["qualified"]) self.assertEqual( result["qualification_state"], "BLOCKED_FROM_LIMITED_LIVE", ) self.assertTrue(result["hard_gate_failures"]) self.assertEqual( set(result["domain_floor_failures"]), {f"D{index}" for index in range(1, 8)}, ) def test_a_pass_without_hashed_evidence_is_rejected(self): assessment = copy.deepcopy(self.assessment) assessment["controls"][0]["status"] = "passed" with self.assertRaisesRegex( MaturityStandardError, "without immutable evidence", ): evaluate_assessment( self.standard, assessment, repository_root=PROJECT, ) def test_all_controls_form_a_test_only_structural_path(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) passing_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] } assessment = self._prepare_qualified_assessment( root, passing_controls, ) result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertEqual(result["effective_score"], 100) self.assertFalse(result["qualified"]) self.assertTrue(result["structurally_eligible"]) self.assertTrue(result["final_attestation_verified"]) self.assertFalse(result["trusted_attestation_verified"]) self.assertTrue(result["test_only"]) self.assertTrue(result["canary_authorization_chain_verified"]) self.assertEqual( result["qualification_state"], "TEST_ONLY_STRUCTURAL_PASS", ) def test_exactly_eighty_points_is_a_structural_path_only(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertEqual(result["raw_score"], 80) self.assertEqual(result["effective_score"], 80) self.assertFalse(result["qualified"]) self.assertEqual( result["qualification_state"], "TEST_ONLY_STRUCTURAL_PASS", ) def test_structural_score_cannot_self_attest_without_trusted_verifier(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) with self.assertRaisesRegex( MaturityStandardError, "trusted evidence verifier", ): evaluate_assessment( self.standard, assessment, repository_root=root, ) def test_pre_canary_requires_a_separate_trusted_authorization(self): pre_canary_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] and control["id"] != "D6-CANARY" } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, pre_canary_controls, canary_days=0, attestation_kind="pre_canary", ) result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertEqual(result["raw_score"], 77) self.assertFalse(result["qualified"]) self.assertFalse(result["pre_canary_authorized"]) self.assertTrue(result["pre_canary_attestation_verified"]) self.assertEqual( result["pre_canary_authorization"]["claim_id"], "claim-pre_canary_authorization", ) self.assertEqual( result["qualification_state"], "TEST_ONLY_PRE_CANARY_PASS", ) def test_pre_canary_authorization_survives_running_day_updates(self): pre_canary_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] and control["id"] != "D6-CANARY" } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, pre_canary_controls, canary_days=0, attestation_kind="pre_canary", ) assessment["observations"]["canary_trading_days"] = 5 result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertTrue(result["pre_canary_attestation_verified"]) self.assertEqual( result["qualification_state"], "TEST_ONLY_PRE_CANARY_PASS", ) def test_expired_pre_canary_authorization_cannot_open_or_continue_canary(self): pre_canary_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] and control["id"] != "D6-CANARY" } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, pre_canary_controls, canary_days=0, attestation_kind="pre_canary", ) assessment["assessed_as_of"] = "2026-09-01" result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertFalse( result["pre_canary_authorization_valid_as_of_assessment"] ) self.assertEqual( result["pre_canary_authorization"]["validity_status"], "expired", ) self.assertFalse(result["canary_running"]) self.assertEqual( result["qualification_state"], "TEST_ONLY_PRE_CANARY_EXPIRED", ) def test_completed_canary_chain_survives_authorization_expiry(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) assessment["assessed_as_of"] = "2026-09-01" final_link = assessment["qualification_attestation"] final_path = root / final_link["path"] final_claim = json.loads(final_path.read_text(encoding="utf-8")) final_claim["assessment_sha256"] = _final_assessment_digest( assessment ) final_path.write_text( json.dumps(final_claim, sort_keys=True), encoding="utf-8", ) final_link["sha256"] = hashlib.sha256( final_path.read_bytes() ).hexdigest() result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertFalse( result["pre_canary_authorization_valid_as_of_assessment"] ) self.assertTrue(result["canary_authorization_chain_verified"]) self.assertTrue(result["structurally_eligible"]) self.assertEqual( result["qualification_state"], "TEST_ONLY_STRUCTURAL_PASS", ) def test_canary_observation_must_stay_within_authorized_hard_caps(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } mutations = ( ( "capital", {"maximum_capital_cny_observed": 100001}, "authorized capital caps", ), ( "gross_notional", {"maximum_gross_notional_cny_observed": 80001}, "authorized capital caps", ), ( "symbol", {"symbols_observed": ["300001.XSHE"]}, "authorized symbol caps", ), ( "flow", { "maximum_submission_and_cancel_requests_per_second_observed": 3 }, "authorized flow caps", ), ( "automatic_scale_up", {"automatic_scale_up_observed": True}, "automatic scale-up", ), ) for name, mutation, expected in mutations: with self.subTest(name=name), tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) canary = next( control for control in assessment["controls"] if control["id"] == "D6-CANARY" ) evidence = canary["evidence"][0] path = root / evidence["path"] claim = json.loads(path.read_text(encoding="utf-8")) claim.update(mutation) path.write_text( json.dumps(claim, sort_keys=True), encoding="utf-8", ) evidence["sha256"] = hashlib.sha256( path.read_bytes() ).hexdigest() with self.assertRaisesRegex( MaturityStandardError, expected, ): evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=( self._trusted_test_verifier ), ) def test_canary_breach_must_link_to_fail_closed_stop_event(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) canary = next( control for control in assessment["controls"] if control["id"] == "D6-CANARY" ) evidence = canary["evidence"][0] path = root / evidence["path"] claim = json.loads(path.read_text(encoding="utf-8")) claim["breach_events"] = [ { "breach_id": "breach-1", "condition": ( "any_risk_data_rule_or_reconciliation_breach" ), "observed_on": "2026-07-08", "stop_event_id": "missing-stop", } ] path.write_text( json.dumps(claim, sort_keys=True), encoding="utf-8", ) evidence["sha256"] = hashlib.sha256( path.read_bytes() ).hexdigest() with self.assertRaisesRegex( MaturityStandardError, "not linked to an authorized stop", ): evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) def test_canary_breach_with_fail_closed_stop_event_is_accepted(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) canary = next( control for control in assessment["controls"] if control["id"] == "D6-CANARY" ) evidence = canary["evidence"][0] path = root / evidence["path"] claim = json.loads(path.read_text(encoding="utf-8")) claim["breach_events"] = [ { "breach_id": "breach-1", "condition": ( "any_risk_data_rule_or_reconciliation_breach" ), "observed_on": "2026-07-08", "stop_event_id": "stop-1", } ] claim["stop_events"] = [ { "stop_event_id": "stop-1", "condition": ( "any_risk_data_rule_or_reconciliation_breach" ), "triggered_on": "2026-07-08", "trading_stopped": True, "automatic_resume": False, } ] path.write_text( json.dumps(claim, sort_keys=True), encoding="utf-8", ) evidence["sha256"] = hashlib.sha256( path.read_bytes() ).hexdigest() final_link = assessment["qualification_attestation"] final_path = root / final_link["path"] final_claim = json.loads(final_path.read_text(encoding="utf-8")) final_claim["assessment_sha256"] = _final_assessment_digest( assessment ) final_path.write_text( json.dumps(final_claim, sort_keys=True), encoding="utf-8", ) final_link["sha256"] = hashlib.sha256( final_path.read_bytes() ).hexdigest() result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertTrue(result["canary_authorization_chain_verified"]) self.assertEqual( result["canary_observation"]["breach_event_count"], 1, ) self.assertEqual( result["canary_observation"]["stop_event_count"], 1, ) self.assertEqual( result["qualification_state"], "TEST_ONLY_STRUCTURAL_PASS", ) def test_final_path_cannot_skip_pre_canary_history(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) assessment["pre_canary_attestation"] = None result = evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) self.assertFalse(result["qualified"]) self.assertFalse(result["canary_authorization_chain_verified"]) self.assertEqual( result["qualification_state"], "PENDING_PRE_CANARY_HISTORY", ) def test_canary_claim_must_reference_pre_authorization_hash(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) canary = next( control for control in assessment["controls"] if control["id"] == "D6-CANARY" ) evidence = canary["evidence"][0] path = root / evidence["path"] claim = json.loads(path.read_text(encoding="utf-8")) claim["pre_canary_authorization_sha256"] = "8" * 64 path.write_text(json.dumps(claim, sort_keys=True), encoding="utf-8") evidence["sha256"] = hashlib.sha256(path.read_bytes()).hexdigest() with self.assertRaisesRegex( MaturityStandardError, "does not reference the verified authorization", ): evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=self._trusted_test_verifier, ) def test_standard_hash_drift_rejects_old_assessment(self): standard = copy.deepcopy(self.standard) standard["domains"][0]["controls"][0]["acceptance"] += " changed" with self.assertRaisesRegex( MaturityStandardError, "canonical hash does not match", ): evaluate_assessment( standard, self.assessment, repository_root=PROJECT, ) def test_boolean_verifier_cannot_issue_a_qualification(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) with self.assertRaisesRegex( MaturityStandardError, "attestation is not trusted", ): evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=lambda *_args: True, ) def test_repository_cannot_accept_a_fake_production_registry(self): hard_controls = { control["id"] for domain in self.standard["domains"] for control in domain["controls"] if control["hard_gate"] } def fake_production_verifier(*_args): return { "trusted": True, "trust_environment": "production", "verifier_id": "forged-local-verifier", "issuer_id": "forged-local-issuer", "key_fingerprint_sha256": "7" * 64, "policy_version": "forged-v1", } with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) assessment = self._prepare_qualified_assessment( root, hard_controls, ) with self.assertRaisesRegex( MaturityStandardError, "production verifier registry is not implemented", ): evaluate_assessment( self.standard, assessment, repository_root=root, trusted_evidence_verifier=fake_production_verifier, ) def test_hard_gates_cannot_make_the_published_threshold_unattainable(self): standard = copy.deepcopy(self.standard) soft_control = next( control for domain in standard["domains"] for control in domain["controls"] if control["id"] == "D1-REDUNDANCY" ) soft_control["hard_gate"] = True with self.assertRaisesRegex( MaturityStandardError, "hard-gate weights must exactly equal", ): validate_standard(standard) def test_hash_drift_is_rejected(self): assessment = copy.deepcopy(self.assessment) assessment["controls"][0]["status"] = "passed" assessment["controls"][0]["evidence"] = [ { "path": "profiles/production-80/standard.json", "tier": "E4", "sha256": "0" * 64, } ] with self.assertRaisesRegex(MaturityStandardError, "hash drift"): evaluate_assessment( self.standard, assessment, repository_root=PROJECT, ) def test_architecture_is_valid_and_forbidden_import_is_detected(self): current = validate_architecture(PROJECT / "src/quant_os") self.assertTrue(current["ok"], current["violations"]) self.assertEqual(current["scope"], "new_namespace_only") self.assertEqual(current["package_root"], "src/quant_os") self.assertEqual( current["migration_state"], "transitioning_from_quant60_legacy", ) with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) for layer in current["layers"]: (root / layer).mkdir() (root / layer / "__init__.py").write_text("", encoding="utf-8") (root / "research" / "bad.py").write_text( "from quant_os.execution import submit\n", encoding="utf-8", ) (root / "research" / "bad_relative.py").write_text( "from ..execution import submit\n", encoding="utf-8", ) (root / "research" / "bad_from_root.py").write_text( "from quant_os import execution\n", encoding="utf-8", ) (root / "control" / "bad_vendor.py").write_text( "from xtquant import xttrader\n", encoding="utf-8", ) (root / "research" / "bad_legacy.py").write_text( "from quant60.experiment import run\n" "from adapters.jqdata_local import load\n" "from platforms.qlib_runner import main\n", encoding="utf-8", ) (root / "architecture.py").write_text( "from .research import train\n", encoding="utf-8", ) (root / "bad_root.py").write_text("", encoding="utf-8") (root / "unregistered").mkdir() (root / "unregistered" / "__init__.py").write_text( "", encoding="utf-8", ) invalid = validate_architecture(root) self.assertFalse(invalid["ok"]) reasons = [item["reason"] for item in invalid["violations"]] self.assertEqual(reasons.count("forbidden_dependency"), 4) self.assertIn("vendor_sdk_outside_adapter", reasons) self.assertIn("legacy_dependency_escape", reasons) self.assertIn("unregistered_layer_package", reasons) self.assertIn("unregistered_root_module", reasons) if __name__ == "__main__": unittest.main()