diff --git a/docs/adaptive_allocation_control_plane.zh-CN.md b/docs/adaptive_allocation_control_plane.zh-CN.md index 8adb30d..39c7bc9 100644 --- a/docs/adaptive_allocation_control_plane.zh-CN.md +++ b/docs/adaptive_allocation_control_plane.zh-CN.md @@ -14,6 +14,19 @@ P0 提供统一、只读的 Shadow 决策记录,不提供交易授权。它解 输出为 `qsl.selection_decision.v1`,完整保存候选、拒绝原因、平台选择、策略分数和 输入摘要。输出固定为 `authority=shadow_only`、`no_order=true` 且所有建议权重为零。 +## 通用接入边界 + +任何策略仓库或平台仓库都可以向 `quant-adaptive-selection` 提交 +`qsl.selection_input.v1` JSON,并得到可保存、可回放的决策工件: + +```bash +quant-adaptive-selection --input selection-input.json --output selection-decision.json +``` + +输入必须提供带时区的平台健康快照、版本化市场上下文、不可变候选 release、插件风险 +缩放和冻结策略。命令不接受 broker 凭据、运行时目标或下单参数;输出文件仅是 JSON +工件,不会修改平台或调度器。 + ## 固定边界 - 不读取新闻叙事并直接交易;因子必须来自版本化的数据链。 diff --git a/pyproject.toml b/pyproject.toml index d9d79ec..90945aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,7 @@ dev = [ [project.scripts] quant-lifecycle = "quant_platform_kit.strategy_lifecycle.cli:main" quant-strategy-spec = "quant_platform_kit.strategy_spec.cli:main" +quant-adaptive-selection = "quant_platform_kit.adaptive_allocation.cli:main" [tool.setuptools] package-dir = { "" = "src" } diff --git a/src/quant_platform_kit/adaptive_allocation/__init__.py b/src/quant_platform_kit/adaptive_allocation/__init__.py index 1f8ed06..b65baac 100644 --- a/src/quant_platform_kit/adaptive_allocation/__init__.py +++ b/src/quant_platform_kit/adaptive_allocation/__init__.py @@ -14,6 +14,11 @@ StrategyCandidate, ) from quant_platform_kit.adaptive_allocation.selector import select_shadow +from quant_platform_kit.adaptive_allocation.io import ( + SELECTION_INPUT_SCHEMA, + build_shadow_selection, + load_shadow_selection_input, +) __all__ = [ "AdaptiveSelectionPolicy", @@ -22,5 +27,8 @@ "PluginRiskAdjustment", "SelectionDecision", "StrategyCandidate", + "SELECTION_INPUT_SCHEMA", + "build_shadow_selection", + "load_shadow_selection_input", "select_shadow", ] diff --git a/src/quant_platform_kit/adaptive_allocation/cli.py b/src/quant_platform_kit/adaptive_allocation/cli.py new file mode 100644 index 0000000..46191e5 --- /dev/null +++ b/src/quant_platform_kit/adaptive_allocation/cli.py @@ -0,0 +1,27 @@ +"""Command-line entry point for writing a Shadow-only selection record.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Sequence + +from quant_platform_kit.adaptive_allocation.io import build_shadow_selection, load_shadow_selection_input + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Generate a no-order adaptive Shadow selection record") + parser.add_argument("--input", required=True, help="versioned selection-input JSON file") + parser.add_argument("--output", required=True, help="destination JSON artifact path") + args = parser.parse_args(argv) + + decision = build_shadow_selection(load_shadow_selection_input(args.input)) + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(decision.to_dict(), indent=2, sort_keys=True) + "\n", encoding="utf-8") + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/quant_platform_kit/adaptive_allocation/io.py b/src/quant_platform_kit/adaptive_allocation/io.py new file mode 100644 index 0000000..ab902d2 --- /dev/null +++ b/src/quant_platform_kit/adaptive_allocation/io.py @@ -0,0 +1,196 @@ +"""Strict JSON input/output boundary for Shadow-only adaptive selection.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from datetime import date, datetime +from pathlib import Path +from typing import Any + +from quant_platform_kit.adaptive_allocation.contracts import ( + PLATFORM_HEALTH_SCHEMA, + MARKET_CONTEXT_SCHEMA, + AdaptiveSelectionPolicy, + MarketContextSnapshot, + PlatformHealthSnapshot, + PluginRiskAdjustment, + SelectionDecision, + StrategyCandidate, +) +from quant_platform_kit.adaptive_allocation.selector import select_shadow + + +SELECTION_INPUT_SCHEMA = "qsl.selection_input.v1" + + +def _mapping(value: object, field_name: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise ValueError(f"{field_name} must be an object") + return value + + +def _sequence(value: object, field_name: str) -> Sequence[object]: + if isinstance(value, (str, bytes, bytearray)) or not isinstance(value, Sequence): + raise ValueError(f"{field_name} must be an array") + return value + + +def _required(mapping: Mapping[str, Any], field_name: str) -> Any: + if field_name not in mapping: + raise ValueError(f"{field_name} is required") + return mapping[field_name] + + +def _bool(value: object, field_name: str) -> bool: + if not isinstance(value, bool): + raise ValueError(f"{field_name} must be boolean") + return value + + +def _number(value: object, field_name: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{field_name} must be numeric") + return float(value) + + +def _integer(value: object, field_name: str) -> int: + normalized = _number(value, field_name) + if not normalized.is_integer(): + raise ValueError(f"{field_name} must be an integer") + return int(normalized) + + +def _string(value: object, field_name: str) -> str: + if not isinstance(value, str): + raise ValueError(f"{field_name} must be a string") + return value + + +def _date(value: object, field_name: str) -> date: + try: + return date.fromisoformat(_string(value, field_name)) + except ValueError as exc: + raise ValueError(f"{field_name} must be an ISO date") from exc + + +def _datetime(value: object, field_name: str) -> datetime: + try: + parsed = datetime.fromisoformat(_string(value, field_name).replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"{field_name} must be an ISO datetime") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise ValueError(f"{field_name} must include a timezone") + return parsed + + +def _numeric_mapping(value: object, field_name: str) -> dict[str, float]: + return { + _string(key, f"{field_name} key"): _number(item, f"{field_name}.{key}") + for key, item in _mapping(value, field_name).items() + } + + +def _strings(value: object, field_name: str) -> tuple[str, ...]: + return tuple(_string(item, f"{field_name} item") for item in _sequence(value, field_name)) + + +def _parse_market_context(payload: object) -> MarketContextSnapshot: + item = _mapping(payload, "market_context") + if item.get("schema") != MARKET_CONTEXT_SCHEMA: + raise ValueError("market_context schema is unsupported") + return MarketContextSnapshot( + as_of=_date(_required(item, "as_of"), "market_context.as_of"), + domain=_string(_required(item, "domain"), "market_context.domain"), + data_version=_string(_required(item, "data_version"), "market_context.data_version"), + data_freshness_days=_integer(_required(item, "data_freshness_days"), "market_context.data_freshness_days"), + regime=_string(_required(item, "regime"), "market_context.regime"), + regime_confidence=_number(_required(item, "regime_confidence"), "market_context.regime_confidence"), + factors=_numeric_mapping(item.get("factors", {}), "market_context.factors"), + ) + + +def _parse_candidate(payload: object) -> StrategyCandidate: + item = _mapping(payload, "candidate") + return StrategyCandidate( + strategy_profile=_string(_required(item, "strategy_profile"), "candidate.strategy_profile"), + release_digest=_string(_required(item, "release_digest"), "candidate.release_digest"), + lifecycle_stage=_string(_required(item, "lifecycle_stage"), "candidate.lifecycle_stage"), + approved_for_shadow=_bool(_required(item, "approved_for_shadow"), "candidate.approved_for_shadow"), + base_score=_number(_required(item, "base_score"), "candidate.base_score"), + estimated_volatility=_number(_required(item, "estimated_volatility"), "candidate.estimated_volatility"), + factor_exposures=_numeric_mapping(item.get("factor_exposures", {}), "candidate.factor_exposures"), + required_plugins=_strings(item.get("required_plugins", []), "candidate.required_plugins"), + allowed_platform_ids=_strings(item.get("allowed_platform_ids", []), "candidate.allowed_platform_ids"), + ) + + +def _parse_platform_health(payload: object) -> PlatformHealthSnapshot: + item = _mapping(payload, "platform_health item") + if item.get("schema") != PLATFORM_HEALTH_SCHEMA: + raise ValueError("platform_health schema is unsupported") + return PlatformHealthSnapshot( + platform_id=_string(_required(item, "platform_id"), "platform_health.platform_id"), + observed_at=_datetime(_required(item, "observed_at"), "platform_health.observed_at"), + healthy=_bool(_required(item, "healthy"), "platform_health.healthy"), + shadow_capable=_bool(_required(item, "shadow_capable"), "platform_health.shadow_capable"), + reconciliation_ok=_bool(_required(item, "reconciliation_ok"), "platform_health.reconciliation_ok"), + capacity_score=_number(_required(item, "capacity_score"), "platform_health.capacity_score"), + expected_cost_bps=_number(_required(item, "expected_cost_bps"), "platform_health.expected_cost_bps"), + ) + + +def _parse_plugin_adjustment(payload: object) -> PluginRiskAdjustment: + item = _mapping(payload, "plugin_adjustment") + return PluginRiskAdjustment( + plugin_id=_string(_required(item, "plugin_id"), "plugin_adjustment.plugin_id"), + risk_multiplier=_number(_required(item, "risk_multiplier"), "plugin_adjustment.risk_multiplier"), + approved=_bool(item.get("approved", True), "plugin_adjustment.approved"), + ) + + +def _parse_policy(payload: object) -> AdaptiveSelectionPolicy: + item = _mapping(payload, "policy") + return AdaptiveSelectionPolicy( + policy_id=_string(_required(item, "policy_id"), "policy.policy_id"), + max_data_freshness_days=_integer(_required(item, "max_data_freshness_days"), "policy.max_data_freshness_days"), + minimum_regime_confidence=_number(_required(item, "minimum_regime_confidence"), "policy.minimum_regime_confidence"), + minimum_score=_number(_required(item, "minimum_score"), "policy.minimum_score"), + volatility_penalty=_number(_required(item, "volatility_penalty"), "policy.volatility_penalty"), + cost_penalty=_number(_required(item, "cost_penalty"), "policy.cost_penalty"), + max_recommendations=_integer(item.get("max_recommendations", 1), "policy.max_recommendations"), + fail_closed_on_unknown_regime=_bool(item.get("fail_closed_on_unknown_regime", True), "policy.fail_closed_on_unknown_regime"), + ) + + +def build_shadow_selection(payload: Mapping[str, object]) -> SelectionDecision: + """Validate one versioned input bundle and produce a no-order decision record.""" + if payload.get("schema") != SELECTION_INPUT_SCHEMA: + raise ValueError(f"schema must equal {SELECTION_INPUT_SCHEMA}") + return select_shadow( + decision_id=_string(_required(payload, "decision_id"), "decision_id"), + created_at=_datetime(_required(payload, "created_at"), "created_at"), + market_context=_parse_market_context(_required(payload, "market_context")), + candidates=[_parse_candidate(item) for item in _sequence(_required(payload, "candidates"), "candidates")], + platform_health=[ + _parse_platform_health(item) + for item in _sequence(_required(payload, "platform_health"), "platform_health") + ], + plugin_adjustments=[ + _parse_plugin_adjustment(item) + for item in _sequence(payload.get("plugin_adjustments", []), "plugin_adjustments") + ], + policy=_parse_policy(_required(payload, "policy")), + ) + + +def load_shadow_selection_input(path: str | Path) -> Mapping[str, object]: + """Read a JSON input bundle without accepting executable configuration.""" + try: + payload = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError(f"cannot read selection input: {path}") from exc + return _mapping(payload, "selection input") + + +__all__ = ["SELECTION_INPUT_SCHEMA", "build_shadow_selection", "load_shadow_selection_input"] diff --git a/tests/test_adaptive_allocation_io.py b/tests/test_adaptive_allocation_io.py new file mode 100644 index 0000000..501e95e --- /dev/null +++ b/tests/test_adaptive_allocation_io.py @@ -0,0 +1,108 @@ +import json + +import pytest + +from quant_platform_kit.adaptive_allocation import ( + SELECTION_INPUT_SCHEMA, + build_shadow_selection, + load_shadow_selection_input, +) +from quant_platform_kit.adaptive_allocation.cli import main + + +def _payload(**overrides): + values = { + "schema": SELECTION_INPUT_SCHEMA, + "decision_id": "shadow-us-equity-001", + "created_at": "2026-08-29T00:00:00Z", + "market_context": { + "schema": "qsl.market_context_snapshot.v1", + "as_of": "2026-08-28", + "domain": "us_equity", + "data_version": "trusted-snapshot-sha", + "data_freshness_days": 0, + "regime": "normal", + "regime_confidence": 0.9, + "factors": {"momentum": 0.1}, + }, + "candidates": [ + { + "strategy_profile": "candidate_a", + "release_digest": "sha256:candidate-a", + "lifecycle_stage": "shadow_active", + "approved_for_shadow": True, + "base_score": 0.4, + "estimated_volatility": 0.2, + "factor_exposures": {"momentum": 0.5}, + "required_plugins": ["market_regime_control"], + "allowed_platform_ids": ["paper_platform"], + } + ], + "platform_health": [ + { + "schema": "qsl.platform_health_snapshot.v1", + "platform_id": "paper_platform", + "observed_at": "2026-08-29T00:00:00+00:00", + "healthy": True, + "shadow_capable": True, + "reconciliation_ok": True, + "capacity_score": 0.8, + "expected_cost_bps": 1.0, + } + ], + "plugin_adjustments": [ + {"plugin_id": "market_regime_control", "risk_multiplier": 0.8, "approved": True} + ], + "policy": { + "policy_id": "shadow-policy-v1", + "max_data_freshness_days": 1, + "minimum_regime_confidence": 0.6, + "minimum_score": 0.1, + "volatility_penalty": 0.5, + "cost_penalty": 0.01, + }, + } + return values | overrides + + +def test_build_shadow_selection_validates_a_versioned_bundle_and_keeps_no_order(): + decision = build_shadow_selection(_payload()) + + result = decision.to_dict() + assert result["authority"] == "shadow_only" + assert result["no_order"] is True + assert result["recommended_strategy_profile"] == "candidate_a" + assert result["candidates"][0]["proposed_weight"] == 0.0 + + +def test_build_shadow_selection_rejects_naive_platform_timestamp(): + payload = _payload() + payload["platform_health"][0]["observed_at"] = "2026-08-29T00:00:00" + + with pytest.raises(ValueError, match="timezone"): + build_shadow_selection(payload) + + +def test_build_shadow_selection_requires_versioned_context_and_integer_freshness(): + missing_schema = _payload() + del missing_schema["market_context"]["schema"] + fractional_freshness = _payload() + fractional_freshness["market_context"]["data_freshness_days"] = 0.5 + + with pytest.raises(ValueError, match="schema"): + build_shadow_selection(missing_schema) + with pytest.raises(ValueError, match="integer"): + build_shadow_selection(fractional_freshness) + + +def test_cli_writes_only_a_json_decision_artifact(tmp_path): + source = tmp_path / "input.json" + output = tmp_path / "output" / "decision.json" + source.write_text(json.dumps(_payload()), encoding="utf-8") + + assert main(["--input", str(source), "--output", str(output)]) == 0 + + result = json.loads(output.read_text(encoding="utf-8")) + assert result["no_order"] is True + assert result["candidates"][0]["proposed_weight"] == 0.0 + assert load_shadow_selection_input(source)["schema"] == SELECTION_INPUT_SCHEMA