Files
quant-os/tools/capability_probe.py

228 lines
7.1 KiB
Python

"""Report static artifacts and local optional-runtime availability."""
from __future__ import annotations
import argparse
import importlib
import importlib.metadata
import importlib.util
import json
import platform
import sys
from pathlib import Path
from typing import Any, Dict, Optional
CAPABILITY_MATRIX: Dict[str, Dict[str, Any]] = {
"quant_os_research": {
"backtest": True,
"hosted": False,
"live_orders": False,
"safe_default": True,
"market_scope": (
"synthetic research plus verified authorized-provider "
"PIT snapshot decisions/backtests"
),
},
"portable_local": {
"backtest": True,
"hosted": False,
"live_orders": False,
"safe_default": True,
"market_scope": "SH/SZ/BJ core math",
},
"joinquant_hosted": {
"backtest": True,
"hosted": True,
"live_orders": False,
"safe_default": True,
"market_scope": "configured SH/SZ wrapper universe",
},
"qmt_builtin": {
"backtest": True,
"hosted": True,
"live_orders": False,
"safe_default": True,
"market_scope": "configured SH/SZ wrapper universe; backtest-only",
},
"qmt_research": {
"backtest": True,
"hosted": False,
"live_orders": False,
"safe_default": True,
"market_scope": "QMT entitlement; runner is backtest/history only",
},
"qmt_shadow": {
"backtest": False,
"hosted": False,
"live_orders": False,
"safe_default": True,
"market_scope": (
"licensed XtTrader SH/SZ read-only account queries; "
"account-bound inert target diff only"
),
},
"xttrader": {
"backtest": False,
"hosted": False,
"live_orders": "explicit opt-in plus fresh structured policy guard",
"safe_default": True,
"market_scope": "SH/SZ; adapter rejects BJ explicitly",
},
"qlib_0_9_7": {
"backtest": True,
"hosted": False,
"live_orders": False,
"safe_default": True,
"market_scope": (
"provider-defined; portable signal research bridge, "
"not target/order parity"
),
},
"tushare_local_qlib": {
"backtest": True,
"hosted": False,
"live_orders": False,
"safe_default": True,
"market_scope": (
"completed checksummed local Tushare partitions converted into "
"an immutable Qlib provider; source completeness and adjustment "
"gates remain explicit"
),
},
}
def _module_probe(name: str, deep: bool) -> Dict[str, Any]:
try:
available = importlib.util.find_spec(name) is not None
except (ImportError, ModuleNotFoundError, ValueError):
available = False
result: Dict[str, Any] = {
"available": available,
"import_checked": bool(deep),
"import_ok": None,
"error": None,
}
top_level = name.split(".", 1)[0]
try:
result["version"] = importlib.metadata.version(
"pyqlib" if top_level == "qlib" else top_level
)
except importlib.metadata.PackageNotFoundError:
result["version"] = None
if available and deep:
try:
importlib.import_module(name)
result["import_ok"] = True
except Exception as exc:
result["import_ok"] = False
result["error"] = f"{type(exc).__name__}: {exc}"
return result
def probe_capabilities(
project_root: Optional[Path] = None,
*,
deep: bool = False,
) -> Dict[str, Any]:
root = (
project_root.expanduser().resolve()
if project_root
else Path(__file__).resolve().parents[1]
)
generated = root / "dist"
return {
"schema_version": 1,
"python": {
"version": platform.python_version(),
"implementation": platform.python_implementation(),
"executable": sys.executable,
"platform": platform.platform(),
},
"artifacts": {
"portable_core": (root / "src/quant60/portable_core.py").is_file(),
"joinquant_wrapper": (root / "platforms/joinquant_strategy.py").is_file(),
"qmt_builtin_wrapper": (
root / "platforms/qmt_builtin_strategy.py"
).is_file(),
"joinquant_bundle": (
generated / "joinquant_strategy.py"
).is_file(),
"qmt_builtin_bundle": (
generated / "qmt_builtin_strategy.py"
).is_file(),
"bundle_manifest": (
generated / "bundle_manifest.json"
).is_file(),
"colab_notebook": (
root / "notebooks/quant_os_colab.ipynb"
).is_file(),
"jqdata_snapshot_adapter": (
root / "adapters/jqdata_local.py"
).is_file(),
"snapshot_decision_pipeline": (
root / "src/quant60/snapshot_pipeline.py"
).is_file(),
"snapshot_backtest": (
root / "src/quant60/snapshot_backtest.py"
).is_file(),
"mock_parity_exporter": (
root / "tools/export_mock_parity.py"
).is_file(),
"qlib_momentum_and_alpha158_cli": (
root / "platforms/qlib_runner.py"
).is_file(),
"tushare_qlib_adapter": (
root / "adapters/tushare_local.py"
).is_file(),
"tushare_qlib_cli": (
root / "tools/tushare_qlib.py"
).is_file(),
"qmt_shadow_planner": (
root / "src/quant60/qmt_shadow.py"
).is_file(),
"qmt_shadow_cli": (
root / "tools/qmt_shadow_plan.py"
).is_file(),
},
"optional_runtimes": {
"jqdatasdk": _module_probe("jqdatasdk", deep),
"cvxpy": _module_probe("cvxpy", deep),
"pyarrow": _module_probe("pyarrow", deep),
"qlib": _module_probe("qlib", deep),
"xtquant": _module_probe("xtquant", deep),
"xtquant.qmttools": _module_probe("xtquant.qmttools", deep),
"xtquant.xttrader": _module_probe("xtquant.xttrader", deep),
},
"matrix": CAPABILITY_MATRIX,
"verification_boundary": (
"artifact presence and optional import probes only; no external "
"platform backtest or broker entitlement is claimed"
),
}
def _main(argv: Optional[list[str]] = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project-root", type=Path)
parser.add_argument(
"--deep",
action="store_true",
help="Import optional runtimes; still performs no broker mutation.",
)
args = parser.parse_args(argv)
print(
json.dumps(
probe_capabilities(args.project_root, deep=args.deep),
ensure_ascii=False,
indent=2,
sort_keys=True,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(_main())