boat/quant-os
Public Access
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat: establish Quant OS production-80 architecture

Browse Source
This commit is contained in:
zebzhang
2026-07-30 22:59:25 +08:00
parent 919c64c679
commit 26cc814f3f
67 changed files with 7005 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_project_scaffold.py
+338 -6
View File
@@ -1,6 +1,9 @@
import json
import os
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
@@ -23,26 +26,355 @@ class ProjectScaffoldTests(unittest.TestCase):
)
payload = json.loads(result.stdout.strip().splitlines()[-1])
self.assertTrue(payload["ok"])
self.assertGreaterEqual(payload["code_cells"], 5)
self.assertGreaterEqual(payload["code_cells"], 4)
self.assertFalse(payload["executed"])
optional_result = subprocess.run(
[
sys.executable,
"tools/run_colab_notebook.py",
"notebooks/quant_os_colab.ipynb",
"--include-optional",
],
cwd=PROJECT,
check=True,
capture_output=True,
text=True,
)
optional_payload = json.loads(
optional_result.stdout.strip().splitlines()[-1]
)
self.assertTrue(optional_payload["optional_included"])
self.assertGreater(
optional_payload["code_cells"],
payload["code_cells"],
)
def test_secret_files_are_ignored(self):
ignore = (PROJECT / ".gitignore").read_text(encoding="utf-8")
for entry in (".env", "secrets/", "credentials/", "userdata_mini/"):
for entry in (
".env",
"secrets/",
"credentials/",
"userdata_mini/",
"data/",
"build/",
):
self.assertIn(entry, ignore)
def test_tracked_secret_scan_passes(self):
result = subprocess.run(
[sys.executable, "tools/check_tracked_secrets.py"],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("tracked secret scan: OK", result.stdout)
def test_claude_project_contract_exists(self):
content = (PROJECT / "CLAUDE.md").read_text(encoding="utf-8")
self.assertIn("executable source of truth for Quant OS", content)
self.assertIn("make local", content)
def test_gitea_workflow_is_at_monorepo_root(self):
workflow = PROJECT.parent / ".gitea/workflows/quant-os-ci.yml"
def test_gitea_workflow_is_at_independent_repo_root(self):
workflow = PROJECT / ".gitea/workflows/ci.yml"
self.assertTrue(workflow.is_file())
content = workflow.read_text(encoding="utf-8")
self.assertIn("working-directory: quant-os", content)
self.assertIn("bash scripts/validate_all.sh", content)
self.assertFalse((PROJECT / ".gitea/workflows/ci.yml").exists())
self.assertNotIn("working-directory: quant-os", content)
self.assertNotIn('"quant-os/**"', content)
self.assertIn(r"\.env(\.[^/]+)?", content)
self.assertIn(r"\.env\.example", content)
def test_long_lived_product_namespace_is_installable(self):
pyproject = (PROJECT / "pyproject.toml").read_text(encoding="utf-8")
self.assertIn('name = "quant-os"', pyproject)
self.assertIn('quant-os = "quant_os.cli:main"', pyproject)
self.assertIn('quant60 = "quant60.cli:main"', pyproject)
self.assertTrue((PROJECT / "src/quant_os/__init__.py").is_file())
def test_cli_can_resolve_an_explicit_checkout_from_elsewhere(self):
environment = {**os.environ, "PYTHONPATH": str(PROJECT / "src")}
with tempfile.TemporaryDirectory() as temporary:
result = subprocess.run(
[
sys.executable,
"-m",
"quant_os",
"--project-root",
str(PROJECT),
"doctor",
],
cwd=temporary,
env=environment,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
payload = json.loads(result.stdout)
self.assertTrue(payload["ok"])
self.assertEqual(
payload["architecture"]["scope"],
"new_namespace_only",
)
self.assertGreater(
payload["migration"]["legacy_python_files_outside_checker"],
0,
)
def test_nginx_source_of_truth_has_atomic_release_routes(self):
config = (
PROJECT / "ops/nginx/quant-os-static.conf"
).read_text(encoding="utf-8")
self.assertIn("location ^~ /quant-os/status/", config)
self.assertIn("root /srv/www;", config)
self.assertIn(
"https://$host/quant-os/status/",
config,
)
self.assertNotIn("/quants-strategies/", config)
self.assertNotIn(":8443", config)
temporary = (
PROJECT
/ "ops/nginx/quant-os-legacy-redirects-302.conf"
).read_text(encoding="utf-8")
stable = (
PROJECT
/ "ops/nginx/quant-os-legacy-redirects-308.conf"
).read_text(encoding="utf-8")
self.assertIn(
"location = /quants-strategies/quant-os/status",
temporary,
)
self.assertIn("return 302 ", temporary)
self.assertNotIn("return 308 ", temporary)
self.assertIn("return 308 ", stable)
def test_static_status_deployer_is_atomic_and_non_overwriting(self):
sha = "a" * 40
with tempfile.TemporaryDirectory() as temporary:
web_root = Path(temporary) / "web"
result = subprocess.run(
[
"bash",
"ops/deploy_static_status.sh",
sha,
"status",
str(web_root),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(
(web_root / "status").readlink(),
Path("releases") / sha,
)
release = web_root / "releases" / sha
self.assertEqual(
{path.name for path in release.iterdir()},
{"index.html", "status-data.js"},
)
self.assertFalse(release.stat().st_mode & stat.S_IWUSR)
self.assertFalse(
release.joinpath("index.html").stat().st_mode & stat.S_IWUSR
)
self.assertFalse(web_root.joinpath(f".incoming-{sha}").exists())
self.assertFalse(web_root.joinpath(".status-deploy.lock").exists())
second_sha = "b" * 40
switched = subprocess.run(
[
"bash",
"ops/deploy_static_status.sh",
second_sha,
"status",
str(web_root),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(switched.returncode, 0, switched.stderr)
self.assertEqual(
(web_root / "status").readlink(),
Path("releases") / second_sha,
)
second = subprocess.run(
[
"bash",
"ops/deploy_static_status.sh",
second_sha,
"status",
str(web_root),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertNotEqual(second.returncode, 0)
self.assertIn("refusing to overwrite", second.stderr)
def test_static_status_deployer_rejects_non_regular_or_extra_source(self):
cases = ("symlink", "extra-directory")
for case in cases:
with self.subTest(case=case):
temporary_context = tempfile.TemporaryDirectory()
self.addCleanup(temporary_context.cleanup)
temporary = temporary_context.name
root = Path(temporary)
source = root / "source"
source.mkdir()
source.joinpath("status-data.js").write_text(
"window.STATUS = {};\n",
encoding="utf-8",
)
if case == "symlink":
source.joinpath("index.html").symlink_to(
PROJECT / "status/index.html"
)
else:
source.joinpath("index.html").write_text(
"<!doctype html>\n",
encoding="utf-8",
)
source.joinpath("extra").mkdir()
web_root = root / "web"
result = subprocess.run(
[
"bash",
"ops/deploy_static_status.sh",
"c" * 40,
str(source),
str(web_root),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertNotEqual(result.returncode, 0)
self.assertFalse(web_root.joinpath("releases").exists())
def test_static_status_deploy_and_activation_share_concurrency_lock(self):
sha = "d" * 40
with tempfile.TemporaryDirectory() as temporary:
web_root = Path(temporary) / "web"
web_root.mkdir()
web_root.joinpath(".status-deploy.lock").mkdir()
deploy = subprocess.run(
[
"bash",
"ops/deploy_static_status.sh",
sha,
"status",
str(web_root),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertNotEqual(deploy.returncode, 0)
self.assertIn("another status deployment", deploy.stderr)
release = web_root / "releases" / sha
release.mkdir(parents=True)
for name in ("index.html", "status-data.js"):
release.joinpath(name).write_text(name, encoding="utf-8")
activate = subprocess.run(
[
"bash",
"ops/activate_static_status.sh",
sha,
str(web_root),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertNotEqual(activate.returncode, 0)
self.assertIn("another status deployment", activate.stderr)
self.assertFalse(web_root.joinpath("status").exists())
def test_static_status_activation_rejects_extra_release_entry(self):
sha = "e" * 40
with tempfile.TemporaryDirectory() as temporary:
web_root = Path(temporary) / "web"
release = web_root / "releases" / sha
release.mkdir(parents=True)
for name in ("index.html", "status-data.js"):
release.joinpath(name).write_text(name, encoding="utf-8")
release.joinpath("extra").mkdir()
result = subprocess.run(
[
"bash",
"ops/activate_static_status.sh",
sha,
str(web_root),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertNotEqual(result.returncode, 0)
self.assertIn("exactly index.html and status-data.js", result.stderr)
self.assertFalse(web_root.joinpath("status").exists())
self.assertFalse(web_root.joinpath(".status-deploy.lock").exists())
def test_local_asset_inventory_is_deterministic_and_path_safe(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / "data").mkdir()
(root / "artifacts").mkdir()
(root / "data" / "one.bin").write_bytes(b"one")
(root / "artifacts" / "two.json").write_text(
"{}\n",
encoding="utf-8",
)
first = root / "first.json"
result = subprocess.run(
[
sys.executable,
str(PROJECT / "tools/inventory_local_assets.py"),
str(root),
"--output",
str(first),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
payload = json.loads(first.read_text(encoding="utf-8"))
self.assertEqual(
[entry["path"] for entry in payload["entries"]],
["data/one.bin", "artifacts/two.json"],
)
encoded = json.dumps(payload)
self.assertNotIn(str(root), encoded)
compare = subprocess.run(
[
sys.executable,
str(PROJECT / "tools/inventory_local_assets.py"),
str(root),
"--compare",
str(first),
],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(compare.returncode, 0, compare.stderr)
if __name__ == "__main__":
tests/test_public_status.py
+75 -2
View File
@@ -54,6 +54,28 @@ class PublicStatusTests(unittest.TestCase):
self.assertEqual(payload["claim"]["gatesPassed"], 0)
self.assertEqual(payload["claim"]["gatesTotal"], 10)
self.assertTrue(all(gate["status"] == "not_passed" for gate in payload["gates"]))
self.assertEqual(payload["schemaVersion"], 2)
self.assertFalse(payload["maturity80"]["qualified"])
self.assertFalse(payload["maturity80"]["baseline60Passed"])
self.assertEqual(payload["maturity80"]["effectiveScore"], 0)
self.assertFalse(payload["maturity80"]["preCanaryAuthorized"])
self.assertFalse(
payload["maturity80"]["preCanaryAuthorizationVerified"]
)
self.assertFalse(
payload["maturity80"]["canaryAuthorizationChainVerified"]
)
self.assertFalse(
payload["maturity80"]["trustedAttestationVerified"]
)
self.assertEqual(payload["marketScope"]["included"], ["SSE", "SZSE"])
self.assertEqual(
payload["marketScope"]["excludedFailClosed"],
["BSE"],
)
self.assertFalse(payload["marketScope"]["bseTradingEnabled"])
self.assertIn("北交所", payload["marketScope"]["statement"])
self.assertIn("fail closed", payload["marketScope"]["statement"])
def test_public_payload_is_deliberately_non_secret_and_non_local(self):
payload = _read_payload()
@@ -84,8 +106,23 @@ class PublicStatusTests(unittest.TestCase):
self.assertIn("NOT_BASELINE_60", page)
self.assertIn("0 / 10", page)
self.assertIn("validatePublicData", page)
self.assertIn("current canary state lacks authorization", page)
self.assertNotIn("all gates must remain", page)
self.assertIn('id="maturity80"', page)
self.assertIn("renderMaturity80", page)
self.assertIn('id="market-scope-copy"', page)
self.assertIn("个人沪深 A 股", page)
self.assertIn("北交所", page)
self.assertIn("platform.limitations", page)
self.assertNotIn("platform.missingReleaseEvidence", page)
self.assertIn(
"https://git.gomars.fun/boat/quant-os/src/branch/master/",
page,
)
self.assertNotIn(
"quants-strategies/src/branch/master/quant-os",
page,
)
self.assertNotRegex(
page,
re.compile(
@@ -98,6 +135,37 @@ class PublicStatusTests(unittest.TestCase):
for tag in re.findall(r"<a\\b[^>]*target=\"_blank\"[^>]*>", page):
self.assertIn('rel="noopener noreferrer"', tag)
def test_production80_is_machine_evaluated_and_fail_closed(self):
payload = _read_payload()
maturity = payload["maturity80"]
self.assertEqual(maturity["standardId"], "quant-os-production-80-v1")
self.assertEqual(maturity["maximumScore"], 100)
self.assertEqual(maturity["qualificationThreshold"], 80)
self.assertEqual((maturity["rawScore"], maturity["effectiveScore"]), (0, 0))
self.assertEqual(
maturity["state"],
"BLOCKED_FROM_LIMITED_LIVE",
)
self.assertFalse(maturity["qualified"])
self.assertEqual(maturity["controlsPassed"], 0)
self.assertEqual(maturity["controlsTotal"], 25)
self.assertGreater(maturity["hardGatesTotal"], 0)
self.assertEqual(len(maturity["domains"]), 7)
self.assertTrue(
all(not domain["floorPassed"] for domain in maturity["domains"])
)
self.assertTrue(
all(value is False for value in maturity["prerequisites"].values())
)
self.assertEqual(
payload["provenance"]["production80Standard"],
"profiles/production-80/standard.json",
)
self.assertEqual(
payload["provenance"]["production80Assessment"],
"profiles/production-80/current-assessment.json",
)
def test_models_and_platforms_preserve_evidence_boundaries(self):
payload = _read_payload()
models = {model["id"]: model for model in payload["models"]}
@@ -124,6 +192,11 @@ class PublicStatusTests(unittest.TestCase):
self.assertEqual(platforms["QMT built-in bundle"]["status"], "implemented")
self.assertFalse(platforms["XtTrader read-only shadow"]["liveReady"])
self.assertIn("fake", platforms["XtTrader read-only shadow"]["claim"].lower())
shadow_limit = platforms["XtTrader read-only shadow"]["limitations"][0]
self.assertIn("Baseline 60", shadow_limit)
self.assertIn("20", shadow_limit)
self.assertIn("Production 80", shadow_limit)
self.assertIn("60", shadow_limit)
def test_candidate_records_only_real_joinquant_execution_observation(self):
payload = _read_payload()
@@ -301,7 +374,7 @@ class PublicStatusTests(unittest.TestCase):
def test_freshness_uses_daily_bars_not_the_later_trade_calendar(self):
payload = _read_payload()
freshness = {item["id"]: item for item in payload["freshness"]}
self.assertEqual(freshness["project-audit"]["value"], "2026-07-26")
self.assertEqual(freshness["project-audit"]["value"], "2026-07-30")
self.assertEqual(
freshness["joinquant-hosted-market-window"]["marketDataThrough"],
"2024-12-31",
@@ -399,7 +472,7 @@ class PublicStatusTests(unittest.TestCase):
scorecard = {
"claim_policy": {
"baseline_threshold": 60,
"current_claim": "BASELINE_60",
"current_claim": "BASELINE_60_VERIFIED",
},
"current_delivery": {"score": 59},
"gates": [{"id": "G1", "status": "not_passed"}],
tests/test_quant_os_maturity.py
+954
View File
@@ -0,0 +1,954 @@
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()