Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/strategy_plugin_runtime_contract.md
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,12 @@ The loader validates:
For `shadow`, platform runtimes should only add logs, runtime report fields, and
notification context.

Runtime reports retain an artifact's source `execution_controls` for audit, and
also publish `runtime_consumption`. Operators must use the latter for the
effective authority: the current shared contract reports `authority=shadow_only`,
`direct_position_control_allowed=false`, and no broker or live-allocation
permission even when a legacy V1 artifact requests position control.

Artifacts may include display-only i18n fields:

- `localized_messages.schema_version = strategy_plugin_messages.v1`
Expand Down
5 changes: 5 additions & 0 deletions docs/strategy_plugin_runtime_contract.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,11 @@ loader 会校验:

对于 `shadow` 插件,平台运行时只能附加日志、运行报告字段和通知上下文。

运行报告会保留 artifact 的原始 `execution_controls` 供审计,同时发布
`runtime_consumption`。运营判断必须以后者为准:当前共享契约固定报告
`authority=shadow_only`、`direct_position_control_allowed=false`,且不授予
broker 或 live allocation 权限;即使 legacy V1 artifact 自身请求了仓位控制也一样。

artifact 可以包含展示层 i18n 字段:

- `localized_messages.schema_version = strategy_plugin_messages.v1`
Expand Down
29 changes: 29 additions & 0 deletions src/quant_platform_kit/common/strategy_plugins.py
Original file line number Diff line number Diff line change
Expand Up @@ -321,11 +321,40 @@ def report_summary(self) -> dict[str, Any]:
"suggested_action": self.suggested_action,
"would_trade_if_enabled": self.would_trade_if_enabled,
"execution_controls": dict(self.execution_controls),
"runtime_consumption": _strategy_plugin_runtime_consumption_summary(self),
"source_uri": self.source_uri,
"local_path": self.local_path,
}


def _strategy_plugin_runtime_consumption_summary(signal: StrategyPluginSignal) -> dict[str, Any]:
"""Describe the authority actually granted to a loaded sidecar.

Artifact controls remain in ``execution_controls`` as source evidence. This
separate summary prevents a legacy artifact from making a runtime report
look authorized when the shared runtime deliberately clamps all direct
position control to shadow-only metadata.
"""

controls = _strategy_plugin_execution_controls(signal)
policy = _strategy_plugin_consumption_policy(signal)
artifact_position_control_requested = _as_bool(
controls.get("position_control_allowed"), default=False
) or _as_bool(policy.get("position_control_allowed"), default=False)
direct_position_control_allowed = bool(STRATEGY_PLUGIN_DIRECT_POSITION_CONTROL_ALLOWED)
return {
"authority": (
"direct_position_control" if direct_position_control_allowed else "shadow_only"
),
"direct_position_control_allowed": direct_position_control_allowed,
"broker_order_allowed": False,
"live_allocation_mutation_allowed": False,
"runtime_metadata_non_authoritative": not direct_position_control_allowed,
"artifact_position_control_requested": artifact_position_control_requested,
"artifact_schema_is_legacy_v1": signal.schema_version.endswith(".v1"),
}


@dataclass(frozen=True)
class StrategyPluginAlertMessage:
subject: str
Expand Down
25 changes: 25 additions & 0 deletions tests/test_strategy_plugins.py
Original file line number Diff line number Diff line change
Expand Up @@ -555,8 +555,33 @@ def test_build_strategy_plugin_report_payload_uses_compact_summary(self):

self.assertEqual(report_payload["strategy_plugins"][0]["strategy"], "tqqq_growth_income")
self.assertEqual(report_payload["strategy_plugins"][0]["plugin"], "crisis_response_shadow")
runtime_consumption = report_payload["strategy_plugins"][0]["runtime_consumption"]
self.assertEqual(runtime_consumption["authority"], "shadow_only")
self.assertFalse(runtime_consumption["direct_position_control_allowed"])
self.assertFalse(runtime_consumption["broker_order_allowed"])
self.assertTrue(runtime_consumption["artifact_schema_is_legacy_v1"])
self.assertNotIn("payload", report_payload["strategy_plugins"][0])

def test_runtime_report_separates_legacy_position_request_from_runtime_authority(self):
signal = validate_strategy_plugin_signal_payload(
{
**_signal_payload(plugin=PLUGIN_MARKET_REGIME_CONTROL),
"execution_controls": _auditable_position_control_controls(),
"consumption_policy": {
"position_control_allowed": True,
"evidence_status": "automation_approved",
},
}
)

runtime_consumption = signal.report_summary()["runtime_consumption"]

self.assertTrue(runtime_consumption["artifact_position_control_requested"])
self.assertEqual(runtime_consumption["authority"], "shadow_only")
self.assertFalse(runtime_consumption["direct_position_control_allowed"])
self.assertTrue(runtime_consumption["runtime_metadata_non_authoritative"])
self.assertTrue(runtime_consumption["artifact_schema_is_legacy_v1"])

def test_attach_strategy_plugin_metadata_adds_payloads_to_snapshot(self):
signal = validate_strategy_plugin_signal_payload(
{
Expand Down