#!/usr/bin/env python3 """Build the deterministic, deliberately non-secret Quant OS public status data.""" from __future__ import annotations import argparse import hashlib import json import os import re import sys from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Mapping, Sequence, Tuple PROJECT = Path(__file__).resolve().parents[1] DEFAULT_OUTPUT = PROJECT / "status" / "status-data.js" GLOBAL_NAME = "QUANT_OS_PUBLIC_STATUS" SCHEMA_VERSION = 1 SOURCE_PATHS: Tuple[str, ...] = ( "gate_scorecard.json", "dist/bundle_manifest.json", "configs/baseline.json", "README.md", "docs/ARCHITECTURE.md", "docs/AUDIT_2026-07-25.md", "docs/PLATFORM_MATRIX.md", "docs/CROSS_ENGINE_CONSISTENCY.md", "docs/JOINQUANT_HOSTED_EVIDENCE_2026-07-26.md", "docs/TUSHARE_LOCAL_DATA.md", "runbooks/PLATFORM_DEPLOYMENT.md", "tools/build_public_status.py", ) ALLOWED_PROGRESS_STATUSES = frozenset( {"implemented", "tested", "real_runtime", "blocked"} ) FORBIDDEN_PUBLIC_CLAIMS = frozenset({"RELEASE_VERIFIED", "LIVE_READY"}) SENSITIVE_KEY_PARTS = ( "password", "passwd", "secret", "token", "cookie", "credential", "account_id", "userdata_path", "api_key", "private_key", ) ABSOLUTE_PATH_PATTERN = re.compile( r"(?:(?<=^)|(?<=[\s\"'=:(]))" r"(?:/(?:Users|home|root|private|var|tmp)/|[A-Za-z]:[\\/])" ) class PublicStatusError(ValueError): """Raised when a source cannot support a safe public claim.""" def _sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _canonical_json(value: Any) -> bytes: return ( json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) + "\n" ).encode("utf-8") def _load_json(path: Path) -> Dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise PublicStatusError(f"cannot load required JSON source {path.name}: {exc}") if not isinstance(value, dict): raise PublicStatusError(f"{path.name} must contain one JSON object") return value def _safe_repo_path(value: Any, *, label: str) -> str: path = str(value) candidate = Path(path) if ( not path or candidate.is_absolute() or ".." in candidate.parts or "\\" in path or ABSOLUTE_PATH_PATTERN.search(path) ): raise PublicStatusError(f"{label} is not a safe repository-relative path") return candidate.as_posix() def _strip_markdown(value: str) -> str: value = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", value) value = re.sub(r"`([^`]*)`", r"\1", value) value = re.sub(r"\*\*([^*]+)\*\*", r"\1", value) return " ".join(value.split()) def _parse_markdown_table(markdown: str, heading: str) -> List[List[str]]: lines = markdown.splitlines() try: header_index = next( index for index, line in enumerate(lines) if line.strip() == heading ) except StopIteration as exc: raise PublicStatusError(f"required Markdown table is missing: {heading}") from exc if header_index + 1 >= len(lines): raise PublicStatusError(f"Markdown table has no separator: {heading}") separator = lines[header_index + 1].strip() if not separator.startswith("|") or "---" not in separator: raise PublicStatusError(f"Markdown table has an invalid separator: {heading}") rows: List[List[str]] = [] for line in lines[header_index + 2 :]: stripped = line.strip() if not stripped.startswith("|"): break cells = [_strip_markdown(cell.strip()) for cell in stripped.strip("|").split("|")] rows.append(cells) if not rows: raise PublicStatusError(f"Markdown table has no data rows: {heading}") return rows def _slug(value: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") if not slug: raise PublicStatusError(f"cannot derive public id from {value!r}") return slug def _resolve_generated_at( *, explicit: str | None, source_date_epoch: str | None, audit_as_of: str ) -> Tuple[str, str]: if explicit: raw = explicit source = "argument" elif source_date_epoch is not None: try: epoch = int(source_date_epoch) except ValueError as exc: raise PublicStatusError("SOURCE_DATE_EPOCH must be an integer") from exc if epoch < 0: raise PublicStatusError("SOURCE_DATE_EPOCH must not be negative") return ( datetime.fromtimestamp(epoch, timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z"), "source_date_epoch", ) else: raw = f"{audit_as_of}T00:00:00Z" source = "audit_as_of" try: parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError as exc: raise PublicStatusError( "--generated-at must be an ISO-8601 timestamp with a timezone" ) from exc if parsed.tzinfo is None: raise PublicStatusError("--generated-at must include a timezone") normalized = ( parsed.astimezone(timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) return normalized, source def _verify_file_hash(project: Path, relative_path: str, expected: str) -> None: path = project / _safe_repo_path(relative_path, label="bundle path") if not path.is_file(): raise PublicStatusError(f"bundle source is missing: {relative_path}") actual = _sha256_bytes(path.read_bytes()) if actual != expected: raise PublicStatusError( f"bundle manifest drift for {relative_path}: expected {expected}, got {actual}" ) def _validate_bundle_manifest( project: Path, manifest: Mapping[str, Any] ) -> List[Dict[str, Any]]: baseline_path = _safe_repo_path( manifest.get("baseline_config"), label="baseline_config" ) baseline_sha = str(manifest.get("baseline_config_sha256", "")) _verify_file_hash(project, baseline_path, baseline_sha) core_sha = str(manifest.get("portable_core_sha256", "")) artifacts = manifest.get("artifacts") if not isinstance(artifacts, dict) or not artifacts: raise PublicStatusError("bundle manifest artifacts must be a non-empty object") public_artifacts: List[Dict[str, Any]] = [] for artifact_id, raw in sorted(artifacts.items()): if not isinstance(raw, dict): raise PublicStatusError(f"bundle artifact {artifact_id} must be an object") core = _safe_repo_path(raw.get("core"), label=f"{artifact_id}.core") wrapper = _safe_repo_path(raw.get("wrapper"), label=f"{artifact_id}.wrapper") output = _safe_repo_path(raw.get("output"), label=f"{artifact_id}.output") core_artifact_sha = str(raw.get("core_sha256", "")) wrapper_sha = str(raw.get("wrapper_sha256", "")) output_sha = str(raw.get("output_sha256", "")) config_sha = str(raw.get("effective_config_sha256", "")) for path, digest in ( (core, core_artifact_sha), (wrapper, wrapper_sha), (output, output_sha), ): _verify_file_hash(project, path, digest) if core_artifact_sha != core_sha: raise PublicStatusError( f"{artifact_id} portable core hash differs from bundle manifest" ) public_artifacts.append( { "id": str(artifact_id), "core": core, "coreSha256": core_artifact_sha, "effectiveConfigSha256": config_sha, "output": output, "outputSha256": output_sha, "wrapper": wrapper, "wrapperSha256": wrapper_sha, "integrity": "manifest_hashes_match_repository", } ) return public_artifacts def _extract_test_summary(facts: Sequence[Any]) -> Tuple[int | None, int | None]: for fact in facts: text = str(fact) match = re.search( r"completed\s+(\d+)\s+tests.*?(\d+)\s+optional-runtime skips", text, flags=re.IGNORECASE, ) if match: return int(match.group(1)), int(match.group(2)) return None, None def _platform_status(name: str) -> str: mapping = { "Local synthetic execution": "tested", "Local synthetic research": "tested", "JQData ingestion": "tested", "Snapshot local backtest": "tested", "Snapshot decision": "tested", "Colab": "tested", "JoinQuant local harness": "tested", "JoinQuant hosted bundle": "real_runtime", "QMT local harness": "tested", "QMT built-in bundle": "implemented", "qmttools runner": "implemented", "XtTrader read-only shadow": "tested", "Qlib native momentum": "real_runtime", "Qlib Alpha158/LightGBM": "real_runtime", "Tushare local → Qlib": "real_runtime", "Mock parity exporter": "tested", } try: return mapping[name] except KeyError as exc: raise PublicStatusError( f"PLATFORM_MATRIX has an unmapped capability row: {name}" ) from exc def _platform_evidence_links(name: str) -> List[str]: links = ["platform-matrix"] if "JoinQuant" in name: links.append("joinquant-hosted-evidence") if "Qlib" in name or "Tushare" in name: links.append("tushare-data") if "QMT" in name or "XtTrader" in name: links.append("platform-deployment") return list(dict.fromkeys(links)) def _build_platforms(platform_markdown: str) -> List[Dict[str, Any]]: rows = _parse_markdown_table( platform_markdown, "| Path | Intended use | Runtime / prerequisite | Current verified fact | Missing release evidence |", ) platforms: List[Dict[str, Any]] = [] for row in rows: if len(row) != 5: raise PublicStatusError( "PLATFORM_MATRIX capability rows must have exactly five columns" ) name, intended_use, prerequisite, verified_fact, missing = row status = _platform_status(name) if status not in ALLOWED_PROGRESS_STATUSES: raise PublicStatusError(f"unsafe platform status for {name}: {status}") platforms.append( { "id": _slug(name.replace("→", " to ")), "name": name, "status": status, "claim": verified_fact, "intendedUse": intended_use, "prerequisite": prerequisite, "limitations": [missing], "evidenceLinks": _platform_evidence_links(name), "releaseVerified": False, "liveReady": False, } ) return platforms def _build_evidence(project: Path) -> List[Dict[str, Any]]: records = ( ("scorecard", "Gate scorecard", "gate_scorecard.json", "machine_record"), ( "bundle-manifest", "Portable bundle manifest", "dist/bundle_manifest.json", "machine_record", ), ("readme", "Quant OS verification guide", "README.md", "documentation"), ( "architecture", "Architecture and trust boundaries", "docs/ARCHITECTURE.md", "documentation", ), ( "audit", "Baseline 60 reassessment", "docs/AUDIT_2026-07-25.md", "audit", ), ( "platform-matrix", "Platform capability matrix", "docs/PLATFORM_MATRIX.md", "audit", ), ( "cross-engine", "Cross-engine consistency contract", "docs/CROSS_ENGINE_CONSISTENCY.md", "documentation", ), ( "joinquant-hosted-evidence", "JoinQuant hosted smoke boundary", "docs/JOINQUANT_HOSTED_EVIDENCE_2026-07-26.md", "runtime_evidence", ), ( "tushare-data", "Tushare inventory and Qlib bridge record", "docs/TUSHARE_LOCAL_DATA.md", "runtime_evidence", ), ( "platform-deployment", "Platform deployment runbook", "runbooks/PLATFORM_DEPLOYMENT.md", "runbook", ), ) result: List[Dict[str, Any]] = [] for evidence_id, label, raw_path, kind in records: path = _safe_repo_path(raw_path, label=f"evidence {evidence_id}") if not (project / path).is_file(): raise PublicStatusError(f"required public evidence is missing: {path}") result.append( { "id": evidence_id, "label": label, "path": path, "kind": kind, } ) return result def _build_layers() -> List[Dict[str, Any]]: return [ { "id": "data", "name": "数据与时点", "status": "tested", "claim": "不可变快照、PIT 字段合同和 Tushare→Qlib 技术链路已实现并有本地合同/运行证据。", "capabilities": [ "JQData raw/factor/limits/paused/PIT membership/ST snapshot", "Tushare completed-Parquet inventory and immutable Qlib provider", "semantic manifest and data-version verification", ], "limitations": [ "尚无授权 JQData 真实快照与许可 lineage；Tushare 镜像仅部分完成且缺生产关键字段。" ], "evidenceLinks": ["architecture", "tushare-data", "scorecard"], "releaseVerified": False, }, { "id": "research", "name": "研究与模型", "status": "real_runtime", "claim": "synthetic research pipeline 已通过测试；Qlib 0.9.7 的 momentum 与 Alpha158/LightGBM fixture 已在真实本地运行时完成。", "capabilities": [ "PIT features, purged walk-forward and deterministic Ridge", "Qlib native momentum", "Qlib Alpha158, LightGBM and Recorder workflow", ], "limitations": [ "fixture 仅两只 synthetic 股票；没有授权长样本 OOS、冻结模型包或投资价值证据。" ], "evidenceLinks": ["architecture", "platform-matrix", "scorecard"], "releaseVerified": False, }, { "id": "signal-portfolio", "name": "信号、组合与事前风险", "status": "tested", "claim": "Signal→Target→Order Delta 合同、成本、风险与约束目标在本地测试路径可重放。", "capabilities": [ "portable momentum scoring and capped target weights", "lot-rounded target quantities and T+1 sellable deltas", "factor risk, dated cost and constrained target research", ], "limitations": [ "风险、冲击和容量尚未用授权真实数据校准；G4/G5 均未通过。" ], "evidenceLinks": ["architecture", "scorecard"], "releaseVerified": False, }, { "id": "backtest", "name": "回测与跨引擎", "status": "real_runtime", "claim": "本地 synthetic、Qlib fixture 与一次真实 JoinQuant hosted smoke 已运行；QMT 仍只有 bundle/harness。", "capabilities": [ "local event and snapshot backtests", "JoinQuant hosted single-file bundle", "QMT built-in/qmttools backtest-only bundle", "Qlib research runners", ], "limitations": [ "没有冻结同输入的本地/聚宽/QMT 逐层对比；真实 QMT 尚未运行，G9 未通过。" ], "evidenceLinks": [ "platform-matrix", "joinquant-hosted-evidence", "cross-engine", "scorecard", ], "releaseVerified": False, }, { "id": "platforms", "name": "平台适配", "status": "real_runtime", "claim": "Local、Colab、JoinQuant、Qlib 有可执行入口；JoinQuant/Qlib 有受限真实运行时 smoke。", "capabilities": [ "local and Colab zero-account smoke", "JQData snapshot entrypoint", "JoinQuant upload bundle", "QMT built-in and qmttools entrypoints", ], "limitations": [ "JQData 真实账号、QMT 客户端/数据权限及平台完整导出仍是外部前置条件。" ], "evidenceLinks": ["platform-matrix", "platform-deployment"], "releaseVerified": False, }, { "id": "broker-oms", "name": "券商、订单与影子盘", "status": "blocked", "claim": "XtTrader operator 入口严格只读；HMAC 账户绑定和 semantic replay 已有 fake-broker 合同测试。", "capabilities": [ "read-only broker observation and broker snapshot", "account-bound target-diff shadow plan", "authenticated evidence envelope and exact replay", ], "limitations": [ "无真实券商 attestation、回调/重启证明和 20 个交易日影子记录；没有实盘下单入口。" ], "evidenceLinks": ["architecture", "platform-matrix", "scorecard"], "releaseVerified": False, }, { "id": "operations", "name": "证据、对账与合规", "status": "blocked", "claim": "版本化 schema、manifest、hash-chain、runbook 与 fail-closed 验证器已实现。", "capabilities": [ "versioned exchange schemas and manifest verification", "ledger replay and reconciliation contracts", "security, deployment and incident runbooks", ], "limitations": [ "缺连续真实对账、恢复演练、kill switch 演练及券商程序化交易书面确认。" ], "evidenceLinks": ["architecture", "scorecard", "platform-deployment"], "releaseVerified": False, }, ] def _build_models() -> List[Dict[str, Any]]: return [ { "id": "portable-momentum-v1", "name": "portable-momentum-v1", "role": "跨引擎冻结计算基线", "status": "real_runtime", "claim": "共享 symbol normalization、momentum、target weight/quantity 与 order-delta 计算；已进入 JoinQuant hosted smoke 和本地 Qlib momentum 路径。", "limitations": [ "尚无同输入 L1–L4 真实跨引擎通过证据，不是已验证的生产 champion。" ], "evidenceLinks": [ "architecture", "bundle-manifest", "joinquant-hosted-evidence", ], "investmentValueClaim": False, "releaseVerified": False, }, { "id": "ridge-challenger", "name": "Deterministic Ridge", "role": "synthetic 研究候选", "status": "tested", "claim": "purged walk-forward research pipeline 中可确定性训练和评估。", "limitations": [ "只在 synthetic 研究路径验证；无真实 OOS、versioned model bundle 或跨平台推理验证。" ], "evidenceLinks": ["architecture", "platform-matrix"], "investmentValueClaim": False, "releaseVerified": False, }, { "id": "alpha158-lightgbm", "name": "Alpha158 + LightGBM", "role": "Qlib 研究候选", "status": "real_runtime", "claim": "pyqlib 0.9.7 的 fit、Recorder 和 portfolio workflow 已在真实本地运行时完成。", "limitations": [ "数据仅为两只 synthetic 股票 fixture；表现没有投资意义，尚无冻结部署模型。" ], "evidenceLinks": ["architecture", "platform-matrix"], "investmentValueClaim": False, "releaseVerified": False, }, ] def _build_next_actions(scorecard: Mapping[str, Any]) -> List[Dict[str, Any]]: missing = scorecard.get("current_delivery", {}).get( "external_evidence_missing", [] ) if not isinstance(missing, list) or len(missing) < 5: raise PublicStatusError( "scorecard must list the external evidence blocking release" ) return [ { "priority": 1, "title": "生成授权 JQData 不可变真实快照", "why": str(missing[0]), "gateIds": ["G1", "G3", "G7"], }, { "priority": 2, "title": "冻结聚宽同输入并完成逐层本地比较", "why": str(missing[1]), "gateIds": ["G3", "G9"], }, { "priority": 3, "title": "取得真实 QMT built-in 与 qmttools 导出", "why": str(missing[2]), "gateIds": ["G3", "G6", "G9"], }, { "priority": 4, "title": "完成 QMT 券商合同探针与至少 20 日影子盘", "why": str(missing[3]), "gateIds": ["G6", "G8", "G9"], }, { "priority": 5, "title": "完成连续对账、恢复演练与券商合规确认", "why": str(missing[4]), "gateIds": ["G8", "G10"], }, ] def _build_freshness( *, scorecard: Mapping[str, Any], joinquant_markdown: str, tushare_markdown: str, ) -> List[Dict[str, Any]]: period_match = re.search( r"(?m)^-\s+Period:\s+" r"(\d{4}-\d{2}-\d{2})\s+through\s+(\d{4}-\d{2}-\d{2})\s*$", joinquant_markdown, ) if period_match is None: raise PublicStatusError( "JoinQuant evidence must declare its hosted market-data period" ) joinquant_start, joinquant_end = period_match.groups() daily_match = re.search( r"(?m)^\|\s*`daily`\s*\|\s*([\d,]+)\s*\|\s*([\d,]+)\s*\|\s*" r"(\d{4}-\d{2}-\d{2})[—-](\d{4}-\d{2}-\d{2})", tushare_markdown, ) plan_match = re.search( r"`daily`\s*原计划覆盖.*?共\s*(\d+)\s*个月；当前完成\s*" r"(\d+)\s*个月，约\s*([\d.]+)%", tushare_markdown, flags=re.DOTALL, ) ready_match = re.search( r"(?m)^production_ready\s*=\s*(true|false)\s*$", tushare_markdown, ) if daily_match is None or plan_match is None or ready_match is None: raise PublicStatusError( "Tushare evidence must declare daily coverage and production readiness" ) ( daily_partitions_text, daily_rows_text, tushare_start, tushare_end, ) = daily_match.groups() planned_months_text, completed_months_text, percentage_text = plan_match.groups() completed_partitions = int(daily_partitions_text.replace(",", "")) completed_months = int(completed_months_text) planned_months = int(planned_months_text) if completed_partitions != completed_months: raise PublicStatusError( "Tushare daily partition count differs from completed-month statement" ) calculated_ratio = completed_months / planned_months recorded_percentage = float(percentage_text) if abs(calculated_ratio * 100.0 - recorded_percentage) > 0.1: raise PublicStatusError( "Tushare recorded coverage percentage does not match month counts" ) production_ready = ready_match.group(1) == "true" if production_ready: raise PublicStatusError( "partial Tushare source must not be publicized as production ready" ) audit_as_of = str(scorecard.get("audit_as_of")) return [ { "id": "project-audit", "label": "项目证据审计", "status": "recorded", "asOf": audit_as_of, "value": audit_as_of, "semantics": "证据盘点日期；不是部署时间，也不是行情截止日。", "evidenceLinks": ["scorecard", "audit"], }, { "id": "joinquant-hosted-market-window", "label": "JoinQuant hosted smoke 行情区间", "status": "runtime_smoke", "marketDataFrom": joinquant_start, "marketDataThrough": joinquant_end, "value": f"{joinquant_start} — {joinquant_end}", "productionReady": False, "semantics": "单次真实 hosted runtime smoke 的回测区间；不代表同输入跨引擎通过。", "evidenceLinks": ["joinquant-hosted-evidence", "scorecard"], }, { "id": "tushare-daily-market-data", "label": "Tushare daily 行情覆盖", "status": "partial", "marketDataFrom": tushare_start, "marketDataThrough": tushare_end, "completedPartitions": completed_partitions, "plannedPartitions": planned_months, "completedRows": int(daily_rows_text.replace(",", "")), "completionRatio": round(calculated_ratio, 6), "completionLabel": ( f"{completed_partitions}/{planned_months}（约 {recorded_percentage:g}%）" ), "value": ( f"{completed_partitions}/{planned_months}（约 {recorded_percentage:g}%），" f"行情截至 {tushare_end}" ), "productionReady": production_ready, "semantics": "marketDataThrough 仅取自 daily 行情分区；绝不使用覆盖更晚的 trade_cal 日期。", "evidenceLinks": ["tushare-data", "scorecard"], }, { "id": "jqdata-real-snapshot", "label": "JQData 授权真实快照", "status": "blocked", "marketDataThrough": None, "value": "N/A", "productionReady": False, "semantics": "只有 fake-provider 合同测试；尚无授权账号生成并保存的真实快照。", "evidenceLinks": ["platform-matrix", "scorecard"], }, { "id": "qmt-broker-real-runtime", "label": "QMT / broker 真实运行", "status": "blocked", "lastRealRuntimeAt": None, "value": "N/A", "productionReady": False, "semantics": "QMT 只有 bundle/harness，XtTrader 只有 fake-broker 只读合同；尚无真实运行或 20 日影子证据。", "evidenceLinks": ["platform-matrix", "platform-deployment", "scorecard"], }, ] def _validate_claim(scorecard: Mapping[str, Any]) -> None: policy = scorecard.get("claim_policy") if not isinstance(policy, dict): raise PublicStatusError("scorecard claim_policy must be an object") claim = str(policy.get("current_claim", "")) if claim in FORBIDDEN_PUBLIC_CLAIMS: raise PublicStatusError(f"unsafe public release claim: {claim}") gates = scorecard.get("gates") if not isinstance(gates, list) or not gates: raise PublicStatusError("scorecard gates must be a non-empty list") passed = sum(gate.get("status") == "passed" for gate in gates if isinstance(gate, dict)) threshold = int(policy.get("baseline_threshold", 60)) delivery = scorecard.get("current_delivery", {}) score = delivery.get("score") if isinstance(delivery, dict) else None all_passed = passed == len(gates) if claim != "NOT_BASELINE_60": if not all_passed or not isinstance(score, (int, float)) or score < threshold: raise PublicStatusError( "a baseline claim requires every gate passed and a threshold score" ) if claim == "NOT_BASELINE_60" and all_passed: raise PublicStatusError( "scorecard is internally inconsistent: every gate passed but claim is negative" ) def _walk_public_values(value: Any, path: str = "$") -> Iterable[Tuple[str, Any]]: yield path, value if isinstance(value, dict): for key, child in value.items(): yield from _walk_public_values(child, f"{path}.{key}") elif isinstance(value, list): for index, child in enumerate(value): yield from _walk_public_values(child, f"{path}[{index}]") def _validate_public_payload(payload: Mapping[str, Any]) -> None: encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True) absolute_match = ABSOLUTE_PATH_PATTERN.search(encoded) if absolute_match: raise PublicStatusError( f"public payload contains an absolute local path near {absolute_match.group(0)!r}" ) for claim in FORBIDDEN_PUBLIC_CLAIMS: if claim in encoded: raise PublicStatusError(f"public payload contains forbidden claim {claim}") for path, value in _walk_public_values(payload): if isinstance(value, dict): for key in value: lowered = str(key).lower() if any(part in lowered for part in SENSITIVE_KEY_PARTS): raise PublicStatusError( f"public payload contains a sensitive field at {path}.{key}" ) def build_public_status( project: Path, *, generated_at: str | None = None, source_date_epoch: str | None = None, ) -> Dict[str, Any]: project = project.resolve() scorecard = _load_json(project / "gate_scorecard.json") manifest = _load_json(project / "dist" / "bundle_manifest.json") baseline = _load_json(project / "configs" / "baseline.json") platform_markdown = (project / "docs" / "PLATFORM_MATRIX.md").read_text( encoding="utf-8" ) joinquant_markdown = ( project / "docs" / "JOINQUANT_HOSTED_EVIDENCE_2026-07-26.md" ).read_text(encoding="utf-8") tushare_markdown = (project / "docs" / "TUSHARE_LOCAL_DATA.md").read_text( encoding="utf-8" ) _validate_claim(scorecard) audit_as_of = str(scorecard.get("audit_as_of", "")) try: datetime.strptime(audit_as_of, "%Y-%m-%d") except ValueError as exc: raise PublicStatusError("scorecard audit_as_of must be YYYY-MM-DD") from exc generated_at_value, generated_at_source = _resolve_generated_at( explicit=generated_at, source_date_epoch=source_date_epoch, audit_as_of=audit_as_of, ) gates_raw = scorecard["gates"] gates: List[Dict[str, Any]] = [] for raw_gate in gates_raw: if not isinstance(raw_gate, dict): raise PublicStatusError("every gate must be an object") evidence = raw_gate.get("evidence", []) required = raw_gate.get("required_evidence", []) if not isinstance(evidence, list) or not isinstance(required, list): raise PublicStatusError("gate evidence fields must be arrays") status = str(raw_gate.get("status", "")) if status not in {"not_passed", "passed"}: raise PublicStatusError(f"unsupported gate status: {status}") if status == "passed" and not evidence: raise PublicStatusError( f"{raw_gate.get('id')} cannot be publicized as passed without evidence" ) gates.append( { "id": str(raw_gate.get("id")), "name": str(raw_gate.get("name")), "status": status, "claim": "已通过并有 scorecard 证据。" if status == "passed" else "未通过；不得计分。", "acceptance": str(raw_gate.get("acceptance")), "requiredEvidence": [str(item) for item in required], "evidence": [str(item) for item in evidence], "limitations": [str(item) for item in required] if status != "passed" else [], "evidenceLinks": ["scorecard", "audit"], } ) passed = sum(gate["status"] == "passed" for gate in gates) facts = scorecard.get("current_delivery", {}).get( "verified_supporting_facts", [] ) if not isinstance(facts, list): raise PublicStatusError("verified_supporting_facts must be an array") test_count, optional_skips = _extract_test_summary(facts) artifacts = _validate_bundle_manifest(project, manifest) source_records: List[Dict[str, str]] = [] for relative in SOURCE_PATHS: safe_path = _safe_repo_path(relative, label="provenance source") source_path = project / safe_path if not source_path.is_file(): raise PublicStatusError(f"required provenance source is missing: {safe_path}") source_records.append( {"path": safe_path, "sha256": _sha256_bytes(source_path.read_bytes())} ) source_set_sha = _sha256_bytes(_canonical_json(source_records)) claim_policy = scorecard["claim_policy"] delivery = scorecard["current_delivery"] payload: Dict[str, Any] = { "schemaVersion": SCHEMA_VERSION, "generatedAt": generated_at_value, "generatedAtSource": generated_at_source, "auditAsOf": audit_as_of, "system": str(scorecard.get("system", "Quant OS")), "claim": { "status": str(claim_policy["current_claim"]), "baselineThreshold": int(claim_policy["baseline_threshold"]), "allGatesRequired": bool(claim_policy["all_gates_required"]), "gatesPassed": passed, "gatesTotal": len(gates), "score": delivery.get("score"), "scoreStatus": str(delivery.get("status")), "reason": str(delivery.get("reason")), "releaseVerified": False, "liveReady": False, }, "summary": { "testsRecorded": test_count, "optionalSkipsRecorded": optional_skips, "testEvidenceQualifier": "Scorecard-recorded run; tests do not prove real-platform or broker gates.", "realRuntimeSmokes": [ "JoinQuant hosted daily backtest (runtime/path only)", "Qlib native momentum fixture", "Qlib Alpha158 + LightGBM fixture", "Tushare completed-partition Qlib momentum smoke", ], "gateCoverageLabel": f"{passed}/{len(gates)} passed", "productionReady": False, "investmentValueClaim": False, }, "layers": _build_layers(), "models": _build_models(), "platforms": _build_platforms(platform_markdown), "freshness": _build_freshness( scorecard=scorecard, joinquant_markdown=joinquant_markdown, tushare_markdown=tushare_markdown, ), "gates": gates, "bundles": { "schemaVersion": int(manifest.get("schema_version")), "baselineConfig": _safe_repo_path( manifest.get("baseline_config"), label="baseline_config" ), "baselineConfigSha256": str(manifest.get("baseline_config_sha256")), "portableCoreSha256": str(manifest.get("portable_core_sha256")), "artifacts": artifacts, "baselineParameters": { "benchmark": str(baseline["universe"]["index_symbol"]), "universeMode": str(baseline["universe"]["mode"]), "lookback": int(baseline["strategy"]["lookback"]), "topN": int(baseline["strategy"]["top_n"]), "maxWeight": float(baseline["strategy"]["max_weight"]), "grossTarget": float(baseline["strategy"]["gross_target"]), "cashBuffer": float(baseline["strategy"]["cash_buffer"]), "participationRate": float( baseline["execution"]["participation_rate"] ), "slippageBps": float(baseline["execution"]["slippage_bps"]), "rebalanceSchedule": str( baseline["strategy"]["rebalance_schedule"] ), }, }, "evidence": _build_evidence(project), "nextActions": _build_next_actions(scorecard), "provenance": { "scorecard": "gate_scorecard.json", "bundleManifest": "dist/bundle_manifest.json", "platformMatrix": "docs/PLATFORM_MATRIX.md", "baselineConfig": "configs/baseline.json", "sourceSetSha256": source_set_sha, "sources": source_records, "payloadSha256Semantics": "SHA-256 of canonical payload before payloadSha256 is added.", }, } payload_sha = _sha256_bytes(_canonical_json(payload)) payload["provenance"]["payloadSha256"] = payload_sha _validate_public_payload(payload) return payload def render_javascript(payload: Mapping[str, Any]) -> bytes: json_text = json.dumps( payload, ensure_ascii=False, indent=2, sort_keys=True, ) json_text = ( json_text.replace("<", "\\u003c") .replace(">", "\\u003e") .replace("&", "\\u0026") .replace("\u2028", "\\u2028") .replace("\u2029", "\\u2029") ) return ( "// Generated by tools/build_public_status.py; do not edit by hand.\n" f"window.{GLOBAL_NAME} = {json_text};\n" ).encode("utf-8") def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Build the deterministic public Quant OS status data." ) parser.add_argument( "--generated-at", help=( "ISO-8601 build timestamp. Defaults to SOURCE_DATE_EPOCH, then the " "scorecard audit date for reproducible local builds." ), ) parser.add_argument( "--output", type=Path, default=DEFAULT_OUTPUT, help="Output JavaScript path.", ) parser.add_argument( "--check", action="store_true", help="Fail if the output is missing or differs from the generated bytes.", ) return parser def main(argv: Sequence[str] | None = None) -> int: args = _build_parser().parse_args(argv) try: payload = build_public_status( PROJECT, generated_at=args.generated_at, source_date_epoch=os.environ.get("SOURCE_DATE_EPOCH"), ) expected = render_javascript(payload) output = args.output.resolve() if args.check: if not output.is_file(): print(f"public status data is missing: {output}", file=sys.stderr) return 1 actual = output.read_bytes() if actual != expected: print( "public status data drift detected; run " "python3 tools/build_public_status.py", file=sys.stderr, ) return 1 print( json.dumps( { "ok": True, "mode": "check", "output": output.name, "payload_sha256": payload["provenance"]["payloadSha256"], }, sort_keys=True, ) ) return 0 output.parent.mkdir(parents=True, exist_ok=True) output.write_bytes(expected) print( json.dumps( { "ok": True, "mode": "write", "output": output.name, "payload_sha256": payload["provenance"]["payloadSha256"], }, sort_keys=True, ) ) return 0 except (OSError, KeyError, TypeError, PublicStatusError) as exc: print(f"public status build failed: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())