Files
quant-os/tests/test_public_status.py
T

561 lines
22 KiB
Python

import importlib.util
import hashlib
import json
import os
import re
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
PROJECT = Path(__file__).resolve().parents[1]
BUILDER_PATH = PROJECT / "tools" / "build_public_status.py"
OUTPUT_PATH = PROJECT / "status" / "status-data.js"
INDEX_PATH = PROJECT / "status" / "index.html"
PREFIX = "window.QUANT_OS_PUBLIC_STATUS = "
def _load_builder():
spec = importlib.util.spec_from_file_location("build_public_status", BUILDER_PATH)
if spec is None or spec.loader is None:
raise RuntimeError("cannot import public status builder")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _read_payload(path=OUTPUT_PATH):
content = path.read_text(encoding="utf-8")
lines = content.splitlines()
if not lines or not lines[0].startswith("// Generated by"):
raise AssertionError("generated header is missing")
assignment = "\n".join(lines[1:])
if not assignment.startswith(PREFIX) or not assignment.endswith(";"):
raise AssertionError("unexpected JavaScript wrapper")
return json.loads(assignment[len(PREFIX) : -1])
class PublicStatusTests(unittest.TestCase):
def test_committed_status_data_has_no_drift(self):
result = subprocess.run(
[sys.executable, str(BUILDER_PATH), "--check"],
cwd=PROJECT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
payload = _read_payload()
self.assertEqual(payload["claim"]["status"], "NOT_BASELINE_60")
self.assertFalse(payload["claim"]["releaseVerified"])
self.assertFalse(payload["claim"]["liveReady"])
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"])
self.assertEqual(payload["summary"]["testsRecorded"], 325)
self.assertEqual(payload["summary"]["optionalSkipsRecorded"], 1)
def test_public_payload_is_deliberately_non_secret_and_non_local(self):
payload = _read_payload()
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True)
self.assertNotRegex(
encoded,
re.compile(r"/(?:Users|home|root|private|var|tmp)/|[A-Za-z]:[\\/]"),
)
self.assertNotIn("RELEASE_VERIFIED", encoded)
self.assertNotIn("LIVE_READY", encoded)
lowered = encoded.lower()
for marker in (
"password",
"passwd",
"api_key",
"private_key",
"userdata_path",
):
self.assertNotIn(marker, lowered)
for evidence in payload["evidence"]:
self.assertFalse(Path(evidence["path"]).is_absolute())
self.assertNotIn("..", Path(evidence["path"]).parts)
def test_public_tushare_claims_cannot_regress_to_early_mirror(self):
payload = _read_payload()
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True)
scorecard = (PROJECT / "gate_scorecard.json").read_text(
encoding="utf-8"
)
stale_claims = (
"93 个",
"8.2%",
"1993-10-29",
"CSI500-2018-2025/de9657cd",
"de5b96253ff765fa4e4be9736ddc2fad528f72c7721964313956d44b02969173",
"3c550c0ba986f7eb736b421ac318fdc2a58eacaa65cfbdeaf6b3f920b5288304",
"aed22a7b83e4defd45c4874e568d6aeafdabe43987cea301e0c1d09543dccb7c",
)
for stale_claim in stale_claims:
self.assertNotIn(stale_claim, encoded)
self.assertNotIn(stale_claim, scorecard)
def test_static_page_has_fail_closed_fallback_and_no_local_paths(self):
page = INDEX_PATH.read_text(encoding="utf-8")
self.assertIn('src="./status-data.js"', page)
self.assertGreaterEqual(page.count("STATUS DATA UNAVAILABLE"), 2)
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(
r"(?:(?<=^)|(?<=[\s\"'=:(]))"
r"(?:/(?:Users|home|root|private|var|tmp)/|[A-Za-z]:[\\/])"
),
)
self.assertNotIn("quant-os/artifacts", page)
self.assertNotIn("quant-os/data", page)
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"]}
self.assertEqual(
set(models),
{"portable-momentum-v1", "ridge-challenger", "alpha158-lightgbm"},
)
self.assertFalse(models["portable-momentum-v1"]["releaseVerified"])
self.assertFalse(models["ridge-challenger"]["investmentValueClaim"])
self.assertFalse(models["alpha158-lightgbm"]["investmentValueClaim"])
self.assertIn("synthetic", models["ridge-challenger"]["role"])
self.assertIn("synthetic", models["alpha158-lightgbm"]["limitations"][0])
platforms = {platform["name"]: platform for platform in payload["platforms"]}
self.assertEqual(platforms["JoinQuant hosted bundle"]["status"], "real_runtime")
self.assertIn(
"3",
platforms["JoinQuant hosted bundle"]["claim"],
)
self.assertIn(
"execution",
platforms["JoinQuant hosted bundle"]["limitations"][0],
)
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()
candidate = payload["candidateBoundary"]
self.assertEqual(candidate["decisionMode"], "target_package")
self.assertEqual(candidate["localEvidenceClass"], "synthetic_research")
self.assertTrue(candidate["realJoinQuantObserved"])
self.assertFalse(candidate["realQmtObserved"])
self.assertEqual(
candidate["realPlatformObservedLayers"],
["execution"],
)
self.assertFalse(candidate["allFiveLayersRealPlatformObserved"])
run = candidate["joinQuantTargetPackageRun"]
self.assertEqual(
run["runId"],
"2457a7c39a276e09e0fabf99e28978e1",
)
self.assertEqual(run["evidenceClass"], "synthetic-research")
self.assertEqual(run["observedLayers"], ["execution"])
self.assertEqual((run["orders"], run["fills"]), (3, 3))
self.assertFalse(run["allFiveLayersObserved"])
self.assertFalse(run["qmtPeerObserved"])
self.assertFalse(run["g9Passed"])
self.assertFalse(run["baseline60Passed"])
self.assertFalse(run["investmentValueClaim"])
self.assertFalse(candidate["joinQuantTargetPackageCountsAsBaseline"])
self.assertFalse(candidate["joinQuantTargetPackageCountsAsG9"])
self.assertEqual(
candidate["historicalJoinQuantMode"], "portable_momentum_smoke"
)
self.assertFalse(candidate["historicalJoinQuantCountsAsBaseline"])
self.assertTrue(payload["summary"]["candidateRealPlatformObserved"])
self.assertEqual(
payload["summary"]["candidateRealPlatformObservationScope"],
"target_package_execution_consumer_only",
)
self.assertFalse(
payload["summary"]["allFiveLayersRealPlatformObserved"]
)
reachability = candidate["reachability"]
self.assertEqual(
reachability["releaseId"], "quant60-research-candidate-v1"
)
self.assertTrue(
reachability["summary"]["all_layers_implemented"]
)
self.assertTrue(
reachability["summary"]["all_layers_local_entrypoint_reachable"]
)
self.assertFalse(
reachability["summary"]["all_layers_real_platform_observed"]
)
self.assertEqual(
reachability["realPlatformsObserved"],
["joinquant"],
)
self.assertEqual(
reachability["realPlatformObservedLayers"],
["execution"],
)
evidence = {item["id"]: item for item in payload["evidence"]}
self.assertEqual(
evidence["release-reachability"]["path"],
"releases/quant60-research-candidate-v1/reachability.json",
)
self.assertEqual(
evidence["joinquant-target-package-run"]["path"],
"evidence/joinquant/TP-20240311-795dcfba/"
"2457a7c39a276e09e0fabf99e28978e1/run.json",
)
self.assertEqual(
evidence["joinquant-target-package-evidence"]["path"],
"docs/JOINQUANT_TARGET_PACKAGE_EVIDENCE_2026-07-26.md",
)
self.assertEqual(
evidence["tushare-csi500-research-run"]["path"],
"evidence/tushare/CSI500-2018-2025/5bf19d2d/run.json",
)
gaps = payload["deliveryGaps"]
self.assertGreaterEqual(len(gaps["internal"]), 3)
self.assertGreaterEqual(len(gaps["externalEvidence"]), 5)
internal = " ".join(gaps["internal"])
self.assertIn("provider-data", internal)
self.assertIn("factor_risk.py", internal)
self.assertIn("CVXPY", internal)
self.assertIn("TWAP/POV", internal)
self.assertIn("synthetic-research", internal)
self.assertIn("TargetPackageV1", internal)
self.assertIn("real JoinQuant", internal)
self.assertIn("real QMT", internal)
def test_public_builder_rejects_wrong_reachability_evidence_class(self):
module = _load_builder()
path = (
PROJECT
/ "releases"
/ "quant60-research-candidate-v1"
/ "reachability.json"
)
manifest = json.loads(path.read_text(encoding="utf-8"))
artifact = manifest["layers"]["alpha"]["jq_entrypoint_reachable"][
"artifacts"
][0]
artifact["kind"] = "source"
body = {
key: value
for key, value in manifest.items()
if key != "manifest_sha256"
}
manifest["manifest_sha256"] = hashlib.sha256(
json.dumps(
body,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
).hexdigest()
with self.assertRaises(module.PublicStatusError):
module._validate_reachability_manifest(PROJECT, manifest)
def test_public_builder_rejects_upstream_real_platform_overclaim(self):
module = _load_builder()
path = (
PROJECT
/ "releases"
/ "quant60-research-candidate-v1"
/ "reachability.json"
)
manifest = json.loads(path.read_text(encoding="utf-8"))
artifact = dict(
manifest["layers"]["execution"]["real_platform_observed"][
"artifacts"
][0]
)
artifact["artifact_id"] = "alpha-real-platform-overclaim"
manifest["layers"]["alpha"]["real_platform_observed"] = {
"status": True,
"artifacts": [artifact],
}
body = {
key: value
for key, value in manifest.items()
if key != "manifest_sha256"
}
manifest["manifest_sha256"] = hashlib.sha256(
json.dumps(
body,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
).hexdigest()
with self.assertRaisesRegex(
module.PublicStatusError,
"only JoinQuant execution",
):
module._validate_reachability_manifest(PROJECT, manifest)
def test_public_builder_rejects_target_package_performance_claim(self):
module = _load_builder()
path = (
PROJECT
/ "evidence"
/ "joinquant"
/ "TP-20240311-795dcfba"
/ "2457a7c39a276e09e0fabf99e28978e1"
/ "run.json"
)
run = json.loads(path.read_text(encoding="utf-8"))
run["performance_claim"] = True
with self.assertRaisesRegex(
module.PublicStatusError,
"evidence classes are unsafe",
):
module._validate_joinquant_target_package_run(PROJECT, run)
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-31")
self.assertEqual(
freshness["joinquant-hosted-market-window"]["marketDataThrough"],
"2024-12-31",
)
target_package = freshness[
"joinquant-target-package-execution-smoke"
]
self.assertEqual(target_package["marketDataFrom"], "2024-03-11")
self.assertEqual(target_package["marketDataThrough"], "2024-03-13")
self.assertEqual(
target_package["observedAt"],
"2026-07-26T08:50:15Z",
)
self.assertFalse(target_package["productionReady"])
self.assertFalse(target_package["investmentValueClaim"])
self.assertIn("execution", target_package["semantics"])
self.assertIn("G9", target_package["semantics"])
tushare = freshness["tushare-daily-market-data"]
self.assertEqual(tushare["status"], "research_only")
self.assertEqual(tushare["completedPartitions"], 96)
self.assertEqual(tushare["plannedPartitions"], 96)
self.assertEqual(tushare["completionRatio"], 1.0)
self.assertEqual(tushare["completedRows"], 8_813_885)
self.assertEqual(tushare["marketDataFrom"], "2018-01-02")
self.assertEqual(tushare["marketDataThrough"], "2025-12-31")
self.assertEqual(
(tushare["verifiedSnapshotJobs"], tushare["snapshotJobs"]),
(221, 221),
)
self.assertEqual(tushare["providerFiles"], 10_011)
self.assertEqual(tushare["providerBytes"], 71_707_194)
self.assertEqual(tushare["calendarSessions"], 1_942)
self.assertEqual(tushare["instrumentCount"], 1_111)
self.assertEqual(tushare["selectedDailyRows"], 1_975_455)
self.assertEqual(tushare["observedMembershipSnapshots"], 97)
self.assertEqual(tushare["effectiveMembershipSnapshots"], 96)
self.assertFalse(tushare["sameSessionMembershipUse"])
self.assertTrue(tushare["sourceStable"])
self.assertTrue(tushare["stateDbChangedDuringVerification"])
self.assertTrue(tushare["converterSourceMatchesCurrent"])
self.assertEqual(
tushare["providerDataVersion"],
"5bf19d2da064357ad1802bca4bfa0c0505ed63fe2b24785ec6c56ecff1724963",
)
self.assertEqual(
tushare["providerManifestSha256"],
"27fb6fedb6a4b114eae2aba44505fb6c4d32c7a9ae81e58c724dfadb8dd10ed0",
)
self.assertEqual(tushare["sourceJobCount"], 221)
self.assertEqual(
tushare["lineageMatchedFields"],
[
"id",
"path",
"row_count",
"byte_count",
"sha256",
"request_sha256",
],
)
self.assertEqual(tushare["lineageMismatchCount"], 0)
self.assertEqual(
tushare["runSha256"],
"803035a95ee9cf6f90af20cdf7e7024149b97e0113b585209a80f80048be87d3",
)
self.assertEqual(tushare["signalRows"], 176_615)
self.assertEqual(tushare["portfolioDays"], 1_698)
self.assertFalse(tushare["productionReady"])
self.assertFalse(tushare["investmentValueClaim"])
self.assertEqual(tushare["gateCredit"], [])
source = (PROJECT / "docs" / "TUSHARE_LOCAL_DATA.md").read_text(
encoding="utf-8"
)
self.assertRegex(source, r"`trade_cal`.*2026")
self.assertIn("221/221", tushare["semantics"])
self.assertIn(
"tushare-csi500-research-run",
tushare["evidenceLinks"],
)
self.assertEqual(freshness["jqdata-real-snapshot"]["value"], "N/A")
self.assertEqual(freshness["jqdata-real-snapshot"]["status"], "blocked")
self.assertIsNone(
freshness["qmt-broker-real-runtime"]["lastRealRuntimeAt"]
)
self.assertEqual(
freshness["qmt-broker-real-runtime"]["status"], "blocked"
)
def test_explicit_timestamp_build_is_deterministic(self):
with tempfile.TemporaryDirectory() as temp_dir:
first = Path(temp_dir) / "first.js"
second = Path(temp_dir) / "second.js"
command = [
sys.executable,
str(BUILDER_PATH),
"--generated-at",
"2026-07-26T08:09:10+08:00",
]
subprocess.run(
[*command, "--output", str(first)],
cwd=PROJECT,
check=True,
capture_output=True,
text=True,
)
subprocess.run(
[*command, "--output", str(second)],
cwd=PROJECT,
check=True,
capture_output=True,
text=True,
)
self.assertEqual(first.read_bytes(), second.read_bytes())
payload = _read_payload(first)
self.assertEqual(payload["generatedAt"], "2026-07-26T00:09:10Z")
self.assertEqual(payload["generatedAtSource"], "argument")
def test_source_date_epoch_controls_timestamp(self):
with tempfile.TemporaryDirectory() as temp_dir:
output = Path(temp_dir) / "epoch.js"
environment = dict(os.environ)
environment["SOURCE_DATE_EPOCH"] = "0"
result = subprocess.run(
[
sys.executable,
str(BUILDER_PATH),
"--output",
str(output),
],
cwd=PROJECT,
env=environment,
check=False,
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
payload = _read_payload(output)
self.assertEqual(payload["generatedAt"], "1970-01-01T00:00:00Z")
self.assertEqual(payload["generatedAtSource"], "source_date_epoch")
def test_builder_rejects_unearned_positive_claim(self):
module = _load_builder()
scorecard = {
"claim_policy": {
"baseline_threshold": 60,
"current_claim": "BASELINE_60_VERIFIED",
},
"current_delivery": {"score": 59},
"gates": [{"id": "G1", "status": "not_passed"}],
}
with self.assertRaises(module.PublicStatusError):
module._validate_claim(scorecard)
if __name__ == "__main__":
unittest.main()