From 777c90dae4da68530c3f14090abbd8829a181f74 Mon Sep 17 00:00:00 2001 From: AuroNyx <162569106+Atlas2005@users.noreply.github.com> Date: Sat, 1 Aug 2026 14:20:37 +0800 Subject: [PATCH] fix(continuous-paper): bind production runtime inputs --- scripts/run_continuous_paper_cycle_v1.py | 97 +++- scripts/serve_continuous_paper_v1.py | 98 +++- .../continuous_paper/__init__.py | 15 +- .../continuous_paper/active_shadow.py | 25 + .../continuous_paper/manifest_bridge.py | 262 +++++++++ .../continuous_paper/prefect_flow.py | 52 +- .../continuous_paper/test_manifest_bridge.py | 512 ++++++++++++++++++ .../test_prefect_and_active_shadow.py | 210 +++++++ 8 files changed, 1219 insertions(+), 52 deletions(-) create mode 100644 src/quantpilot_core/continuous_paper/manifest_bridge.py create mode 100644 tests/continuous_paper/test_manifest_bridge.py diff --git a/scripts/run_continuous_paper_cycle_v1.py b/scripts/run_continuous_paper_cycle_v1.py index 1fac0bc..69a56e3 100644 --- a/scripts/run_continuous_paper_cycle_v1.py +++ b/scripts/run_continuous_paper_cycle_v1.py @@ -1,33 +1,86 @@ #!/usr/bin/env python3 """Manual, offline-safe entry point for one continuous paper cycle.""" + from __future__ import annotations + import argparse import os -from pathlib import Path -from quantpilot_core.continuous_paper import ActiveShadowConfig, ContinuousPaperCycle, ContinuousPaperCycleConfig, InMemoryReportingStore, PostgreSQLReportingStore -from quantpilot_core.production_candidate import load_runtime_manifest -from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig + +from quantpilot_core.continuous_paper import ( + ActiveShadowConfig, + ContinuousPaperCycle, + ContinuousPaperCycleConfig, + InMemoryReportingStore, + PostgreSQLReportingStore, + load_production_input_payload, + load_production_pipeline_config, + normalize_active_shadow_roles, +) def main() -> None: - p = argparse.ArgumentParser() - p.add_argument("--production-manifest", required=True); p.add_argument("--decision-session", required=True) - p.add_argument("--state-path", required=True); p.add_argument("--report-path", required=True) - p.add_argument("--dsn-env-var", default="QUANTPILOT_POSTGRES_DSN"); p.add_argument("--test-store", action="store_true") - p.add_argument("--cache-path", default=".cache/quantpilot/active_shadow"); p.add_argument("--timeout", type=float, default=30.0) - p.add_argument("--allowed-roles", default=""); p.add_argument("--enable-active-shadow", action="store_true"); p.add_argument("--enable-live-ai", action="store_true") - p.add_argument("--max-physical-model-calls", type=int, default=0); p.add_argument("--max-estimated-api-cost", type=float, default=0.0); p.add_argument("--estimated-cost-per-call", type=float, default=0.0) - args = p.parse_args() - if args.enable_live_ai and (not args.enable_active_shadow or args.max_physical_model_calls <= 0 or args.max_estimated_api_cost <= 0 or args.estimated_cost_per_call <= 0): - p.error("live AI requires active shadow and positive explicit call, cycle-cost, and per-call cost limits") + parser = argparse.ArgumentParser() + parser.add_argument("--production-manifest", required=True) + parser.add_argument("--decision-session", required=True) + parser.add_argument("--state-path", required=True) + parser.add_argument("--report-path", required=True) + parser.add_argument("--input-json", default=None) + parser.add_argument("--live-market-data", action="store_true") + parser.add_argument("--dsn-env-var", default="QUANTPILOT_POSTGRES_DSN") + parser.add_argument("--test-store", action="store_true") + parser.add_argument("--cache-path", default=".cache/quantpilot/active_shadow") + parser.add_argument("--timeout", type=float, default=30.0) + parser.add_argument("--allowed-roles", default="") + parser.add_argument("--enable-active-shadow", action="store_true") + parser.add_argument("--enable-live-ai", action="store_true") + parser.add_argument("--max-physical-model-calls", type=int, default=0) + parser.add_argument("--max-estimated-api-cost", type=float, default=0.0) + parser.add_argument("--estimated-cost-per-call", type=float, default=0.0) + args = parser.parse_args() + + try: + shadow = ActiveShadowConfig( + enabled=args.enable_active_shadow, + enable_live_calls=args.enable_live_ai, + max_physical_model_calls_per_cycle=args.max_physical_model_calls, + max_estimated_cost_per_cycle=args.max_estimated_api_cost, + estimated_cost_per_call=args.estimated_cost_per_call, + cache_path=args.cache_path, + allow_roles=normalize_active_shadow_roles(args.allowed_roles), + timeout=args.timeout, + ) + except ValueError as exc: + parser.error(str(exc)) + dsn = os.environ.get(args.dsn_env_var) - if not args.test_store and not dsn: p.error("set the DSN environment variable or use explicit --test-store") - manifest = load_runtime_manifest(Path(args.production_manifest)) - pipeline = RealCandidatePipelineConfig(decision_session=args.decision_session, state_path=args.state_path, report_path=args.report_path, production_manifest=manifest) - store = InMemoryReportingStore() if args.test_store else PostgreSQLReportingStore(dsn) - shadow = ActiveShadowConfig(enabled=args.enable_active_shadow, enable_live_calls=args.enable_live_ai, max_physical_model_calls_per_cycle=args.max_physical_model_calls, max_estimated_cost_per_cycle=args.max_estimated_api_cost, estimated_cost_per_call=args.estimated_cost_per_call, cache_path=args.cache_path, timeout=args.timeout) - result = ContinuousPaperCycle(store).run_once(ContinuousPaperCycleConfig(pipeline, run_id=f"manual-{args.decision_session}", active_shadow=shadow)) - print(f"session_id={result.session_id} status={result.status} run_id={result.run_id}") + if not args.test_store and not dsn: + parser.error("set the DSN environment variable or use explicit --test-store") + + pipeline = load_production_pipeline_config( + manifest_path=args.production_manifest, + decision_session=args.decision_session, + state_path=args.state_path, + report_path=args.report_path, + live_market_data=args.live_market_data, + input_payload=load_production_input_payload(args.input_json), + ) + + store = ( + InMemoryReportingStore() + if args.test_store + else PostgreSQLReportingStore(dsn) + ) + result = ContinuousPaperCycle(store).run_once( + ContinuousPaperCycleConfig( + pipeline, + run_id=f"manual-{args.decision_session}", + active_shadow=shadow, + ) + ) + print( + f"session_id={result.session_id} status={result.status} run_id={result.run_id}" + ) -if __name__ == "__main__": main() +if __name__ == "__main__": + main() diff --git a/scripts/serve_continuous_paper_v1.py b/scripts/serve_continuous_paper_v1.py index 5818c8e..1d2a4da 100644 --- a/scripts/serve_continuous_paper_v1.py +++ b/scripts/serve_continuous_paper_v1.py @@ -1,35 +1,97 @@ #!/usr/bin/env python3 """Explicit Prefect deployment helper; no schedule starts during import.""" + from __future__ import annotations + import argparse +from typing import Any def parser() -> argparse.ArgumentParser: - p = argparse.ArgumentParser() - p.add_argument("--production-manifest", required=True) - p.add_argument("--decision-session", required=True) - p.add_argument("--state-path", required=True) - p.add_argument("--report-path", required=True) - p.add_argument("--run-id", required=True) - p.add_argument("--dsn-env-var", default="QUANTPILOT_POSTGRES_DSN") - p.add_argument("--cron", default="30 16 * * 1-5") - p.add_argument("--timezone", default="Asia/Shanghai") - p.add_argument("--max-physical-model-calls", type=int, default=0) - p.add_argument("--max-estimated-api-cost", type=float, default=0.0) - p.add_argument("--estimated-cost-per-call", type=float, default=0.0) - return p + value = argparse.ArgumentParser() + value.add_argument("--production-manifest", required=True) + value.add_argument("--decision-session", required=True) + value.add_argument("--state-path", required=True) + value.add_argument("--report-path", required=True) + value.add_argument("--run-id", required=True) + value.add_argument("--input-json", default=None) + value.add_argument("--live-market-data", action="store_true") + value.add_argument("--dsn-env-var", default="QUANTPILOT_POSTGRES_DSN") + value.add_argument("--cron", default="30 16 * * 1-5") + value.add_argument("--timezone", default="Asia/Shanghai") + value.add_argument("--cache-path", default=".cache/quantpilot/active_shadow") + value.add_argument("--timeout", type=float, default=30.0) + value.add_argument("--allowed-roles", default="") + value.add_argument("--enable-active-shadow", action="store_true") + value.add_argument("--enable-live-ai", action="store_true") + value.add_argument("--max-physical-model-calls", type=int, default=0) + value.add_argument("--max-estimated-api-cost", type=float, default=0.0) + value.add_argument("--estimated-cost-per-call", type=float, default=0.0) + return value def main() -> None: - args = parser().parse_args() + argument_parser = parser() + args = argument_parser.parse_args() + # Deliberately lazy: --help stays useful in minimal/offline environments. + from quantpilot_core.continuous_paper import ( + ActiveShadowConfig, + load_production_input_payload, + normalize_active_shadow_roles, + ) + + try: + shadow = ActiveShadowConfig( + enabled=args.enable_active_shadow, + enable_live_calls=args.enable_live_ai, + max_physical_model_calls_per_cycle=args.max_physical_model_calls, + max_estimated_cost_per_cycle=args.max_estimated_api_cost, + estimated_cost_per_call=args.estimated_cost_per_call, + cache_path=args.cache_path, + allow_roles=normalize_active_shadow_roles(args.allowed_roles), + timeout=args.timeout, + ) + except ValueError as exc: + argument_parser.error(str(exc)) + from quantpilot_core.continuous_paper.prefect_flow import continuous_paper_flow + if not hasattr(continuous_paper_flow, "serve"): - raise RuntimeError("install quantpilot-ai-next[continuous-paper] to serve a Prefect deployment") + raise RuntimeError( + "install quantpilot-ai-next[continuous-paper] to serve a Prefect deployment" + ) from prefect.schedules import Cron - parameters = {"production_manifest_path": args.production_manifest, "decision_session": args.decision_session, "state_path": args.state_path, "report_path": args.report_path, "run_id": args.run_id, "dsn_env_var": args.dsn_env_var, "active_shadow": {"max_physical_model_calls_per_cycle": args.max_physical_model_calls, "max_estimated_cost_per_cycle": args.max_estimated_api_cost, "estimated_cost_per_call": args.estimated_cost_per_call}} - continuous_paper_flow.serve(name="quantpilot-continuous-paper-v1", schedules=[Cron(args.cron, timezone=args.timezone)], parameters=parameters) + parameters: dict[str, Any] = { + "production_manifest_path": args.production_manifest, + "decision_session": args.decision_session, + "state_path": args.state_path, + "report_path": args.report_path, + "run_id": args.run_id, + "dsn_env_var": args.dsn_env_var, + "live_market_data": args.live_market_data, + "input_payload": load_production_input_payload(args.input_json), + "active_shadow": { + "enabled": shadow.enabled, + "enable_live_calls": shadow.enable_live_calls, + "max_physical_model_calls_per_cycle": ( + shadow.max_physical_model_calls_per_cycle + ), + "max_estimated_cost_per_cycle": shadow.max_estimated_cost_per_cycle, + "estimated_cost_per_call": shadow.estimated_cost_per_call, + "cache_path": str(shadow.cache_path), + "allow_roles": [role.value for role in shadow.allow_roles], + "timeout": shadow.timeout, + "failure_policy": shadow.failure_policy, + }, + } + continuous_paper_flow.serve( + name="quantpilot-continuous-paper-v1", + schedules=[Cron(args.cron, timezone=args.timezone)], + parameters=parameters, + ) -if __name__ == "__main__": main() +if __name__ == "__main__": + main() diff --git a/src/quantpilot_core/continuous_paper/__init__.py b/src/quantpilot_core/continuous_paper/__init__.py index 046ee9b..592d994 100644 --- a/src/quantpilot_core/continuous_paper/__init__.py +++ b/src/quantpilot_core/continuous_paper/__init__.py @@ -1,7 +1,16 @@ """One-cycle, advisory-only continuous paper orchestration.""" -from .active_shadow import ActiveShadowConfig, ActiveShadowRunner +from .active_shadow import ( + ActiveShadowConfig, + ActiveShadowRunner, + normalize_active_shadow_roles, +) from .contracts import ContinuousPaperCycleConfig, ContinuousPaperCycleResult, ReportingCycleBundle +from .manifest_bridge import ( + load_production_input_payload, + load_production_pipeline_config, + normalize_production_input_payload, +) from .service import ContinuousPaperCycle from .store import InMemoryReportingStore, PostgreSQLReportingStore, ReportingConflictError @@ -9,5 +18,7 @@ "ActiveShadowConfig", "ActiveShadowRunner", "ContinuousPaperCycle", "ContinuousPaperCycleConfig", "ContinuousPaperCycleResult", "InMemoryReportingStore", "PostgreSQLReportingStore", "ReportingConflictError", - "ReportingCycleBundle", + "ReportingCycleBundle", "load_production_input_payload", + "load_production_pipeline_config", "normalize_active_shadow_roles", + "normalize_production_input_payload", ] diff --git a/src/quantpilot_core/continuous_paper/active_shadow.py b/src/quantpilot_core/continuous_paper/active_shadow.py index 94b311d..347648d 100644 --- a/src/quantpilot_core/continuous_paper/active_shadow.py +++ b/src/quantpilot_core/continuous_paper/active_shadow.py @@ -24,14 +24,39 @@ class ActiveShadowConfig: failure_policy: str = "abstain" def __post_init__(self) -> None: + object.__setattr__(self, "allow_roles", normalize_active_shadow_roles(self.allow_roles)) if self.failure_policy != "abstain": raise ValueError("active shadow supports only failure_policy='abstain'") if self.timeout <= 0: raise ValueError("timeout must be positive") + if self.enable_live_calls and not self.enabled: + raise ValueError("live AI requires active shadow to be enabled") if self.enable_live_calls and (self.max_physical_model_calls_per_cycle <= 0 or self.max_estimated_cost_per_cycle <= 0 or self.estimated_cost_per_call <= 0): raise ValueError("live active shadow requires positive call, cycle-cost, and per-call cost limits") +def normalize_active_shadow_roles( + roles: str | tuple[DeepSeekAdvisoryRole | str, ...] | list[str], +) -> tuple[DeepSeekAdvisoryRole, ...]: + """Normalize CLI or serialized role values to unique runtime enums.""" + if isinstance(roles, str): + values = tuple(item.strip() for item in roles.split(",") if item.strip()) + if not values: + return tuple(DeepSeekAdvisoryRole) + else: + values = tuple(roles) + + normalized: list[DeepSeekAdvisoryRole] = [] + for value in values: + if isinstance(value, DeepSeekAdvisoryRole): + role = value + else: + role = DeepSeekAdvisoryRole(str(value).strip().lower().replace("-", "_")) + if role not in normalized: + normalized.append(role) + return tuple(normalized) + + class ActiveShadowRunner: def __init__(self, config: ActiveShadowConfig, *, agent: Any | None = None) -> None: self.config = config diff --git a/src/quantpilot_core/continuous_paper/manifest_bridge.py b/src/quantpilot_core/continuous_paper/manifest_bridge.py new file mode 100644 index 0000000..eca06d1 --- /dev/null +++ b/src/quantpilot_core/continuous_paper/manifest_bridge.py @@ -0,0 +1,262 @@ +"""Production-boundary helpers for Continuous Paper runtime configuration.""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any, Mapping + +from quantpilot_core.data_provider_normalization import canonicalize_a_share_symbol +from quantpilot_core.production_candidate import load_runtime_manifest +from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig +from quantpilot_core.runtime_account import AccountCapabilities + + +FROZEN_MANIFEST_PARAMETER_NAMES = frozenset( + { + "initial_capital", + "target_symbol_count", + "max_execution_symbols", + "target_position_count", + "max_position_weight", + "reserve_cash_weight", + "min_order_lot", + "capital_profile_id", + "strategy_id", + } +) +PRODUCTION_INPUT_PAYLOAD_FIELDS = frozenset( + { + "symbols", + "bars", + "information_signals", + "information_provenance", + "advisory_provenance", + "quant_firm_context", + } +) +_ALLOWED_PIPELINE_OPTION_NAMES = frozenset( + { + "input_calendar_sessions", + "input_calendar_provider", + "account_capabilities", + "shadow_desk_evidence", + } +) + + +def load_production_input_payload(path: str | Path | None) -> dict[str, Any]: + """Read an optional input JSON object without creating runtime objects.""" + if path is None: + return {} + raw = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(raw, Mapping): + raise TypeError( + f"production input JSON must contain an object, got {type(raw).__name__}" + ) + normalize_production_input_payload(raw) + return dict(raw) + + +def normalize_production_input_payload( + payload: Mapping[str, Any] | None, +) -> dict[str, Any]: + """Validate the shared production-input schema and create runtime values.""" + if payload is None: + payload = {} + if not isinstance(payload, Mapping): + raise TypeError( + f"production input payload must be an object, got {type(payload).__name__}" + ) + + unknown = set(payload) - PRODUCTION_INPUT_PAYLOAD_FIELDS + if unknown: + raise ValueError( + "unknown production input fields: " + ", ".join(sorted(map(str, unknown))) + ) + + symbols = tuple( + dict.fromkeys( + _normalize_symbol(symbol) + for symbol in _array_field(payload, "symbols") + if str(symbol).strip() + ) + ) + bars = _mapping_array_field(payload, "bars") + information_signals = _mapping_array_field(payload, "information_signals") + + return { + "symbols": symbols, + "input_bars": bars, + "information_signals": information_signals, + "information_provenance": _mapping_field( + payload, "information_provenance" + ), + "advisory_provenance": _mapping_field(payload, "advisory_provenance"), + "quant_firm_context": _mapping_field(payload, "quant_firm_context"), + } + + +def load_production_pipeline_config( + *, + manifest_path: str | Path, + decision_session: str, + state_path: str | Path = ".cache/real_candidate_daily_paper/state.json", + report_path: str | Path | None = ( + ".cache/real_candidate_daily_paper/latest_report.json" + ), + live_market_data: bool = False, + input_payload: Mapping[str, Any] | None = None, + pipeline_options: Mapping[str, Any] | None = None, +) -> RealCandidatePipelineConfig: + """Bind a manifest file and shared input payload to the real pipeline.""" + raw_manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8")) + if not isinstance(raw_manifest, Mapping): + raise TypeError( + "production manifest must be a JSON object, " + f"got {type(raw_manifest).__name__}" + ) + manifest = load_runtime_manifest(raw_manifest) + + frozen: dict[str, Any] = {} + for group in ( + manifest.frozen_strategy_parameters, + manifest.frozen_portfolio_parameters, + manifest.frozen_execution_parameters, + ): + frozen.update(dict(group)) + + initial_capital = float(frozen["initial_capital"]) + target_symbol_count = int(frozen["target_symbol_count"]) + max_execution_symbols = int(frozen["max_execution_symbols"]) + target_position_count = int(frozen["target_position_count"]) + max_position_weight = float(frozen["max_position_weight"]) + reserve_cash_weight = float(frozen["reserve_cash_weight"]) + min_order_lot = int(frozen["min_order_lot"]) + strategy_id = str(frozen["strategy_id"]) + + if target_symbol_count > max_execution_symbols: + raise ValueError( + f"production manifest target_symbol_count ({target_symbol_count}) " + f"exceeds max_execution_symbols ({max_execution_symbols}); " + "rebuild the manifest with compatible values" + ) + + normalized_input = normalize_production_input_payload(input_payload) + symbols = normalized_input["symbols"] + if live_market_data and not symbols: + raise ValueError( + "live_market_data requires an explicit non-empty symbols list" + ) + if live_market_data and len(symbols) > min(max_execution_symbols, 6): + raise ValueError("--live-market-data supports at most 6 explicit symbols") + + options = _normalize_pipeline_options(pipeline_options) + return RealCandidatePipelineConfig( + decision_session=decision_session, + initial_capital=initial_capital, + state_path=state_path, + report_path=report_path, + strategy_id=strategy_id, + production_strategy_id=manifest.strategy_candidate_id, + target_position_count=target_position_count, + max_position_weight=max_position_weight, + reserve_cash_weight=reserve_cash_weight, + min_order_lot=min_order_lot, + capital_profile_id=manifest.capital_profile_id, + max_execution_symbols=max_execution_symbols, + target_symbol_count=target_symbol_count, + live_market_data=live_market_data, + production_manifest=manifest, + **normalized_input, + **options, + ) + + +def _normalize_pipeline_options( + pipeline_options: Mapping[str, Any] | None, +) -> dict[str, Any]: + if pipeline_options is None: + return {} + if not isinstance(pipeline_options, Mapping): + raise TypeError("pipeline_options must be an object") + + frozen_conflicts = FROZEN_MANIFEST_PARAMETER_NAMES.intersection(pipeline_options) + if frozen_conflicts: + raise ValueError( + "pipeline_options conflict with frozen manifest parameters: " + + ", ".join(sorted(frozen_conflicts)) + ) + unknown = set(pipeline_options) - _ALLOWED_PIPELINE_OPTION_NAMES + if unknown: + raise ValueError( + "unsupported pipeline_options: " + ", ".join(sorted(map(str, unknown))) + ) + + normalized = dict(pipeline_options) + if "input_calendar_sessions" in normalized: + value = normalized["input_calendar_sessions"] + if not isinstance(value, (list, tuple)): + raise TypeError("pipeline_options.input_calendar_sessions must be an array") + normalized["input_calendar_sessions"] = tuple(str(item) for item in value) + if "input_calendar_provider" in normalized: + normalized["input_calendar_provider"] = str( + normalized["input_calendar_provider"] + ) + if "shadow_desk_evidence" in normalized: + value = normalized["shadow_desk_evidence"] + if not isinstance(value, Mapping): + raise TypeError("pipeline_options.shadow_desk_evidence must be an object") + normalized["shadow_desk_evidence"] = dict(value) + if "account_capabilities" in normalized: + value = normalized["account_capabilities"] + if value is not None and not isinstance(value, Mapping): + raise TypeError("pipeline_options.account_capabilities must be an object") + if isinstance(value, Mapping): + account_payload = dict(value) + if "supported_order_types" in account_payload: + order_types = account_payload["supported_order_types"] + if not isinstance(order_types, (list, tuple)): + raise TypeError( + "pipeline_options.account_capabilities.supported_order_types " + "must be an array" + ) + account_payload["supported_order_types"] = tuple( + str(item) for item in order_types + ) + normalized["account_capabilities"] = AccountCapabilities(**account_payload) + return normalized + + +def _array_field(payload: Mapping[str, Any], name: str) -> tuple[Any, ...]: + value = payload.get(name, []) + if not isinstance(value, (list, tuple)): + raise TypeError(f"production input field {name} must be an array") + return tuple(value) + + +def _mapping_array_field( + payload: Mapping[str, Any], name: str +) -> tuple[Mapping[str, Any], ...]: + values = _array_field(payload, name) + for index, value in enumerate(values): + if not isinstance(value, Mapping): + raise TypeError( + f"production input field {name}[{index}] must be an object" + ) + return tuple(dict(value) for value in values) + + +def _mapping_field(payload: Mapping[str, Any], name: str) -> dict[str, Any]: + value = payload.get(name, {}) + if not isinstance(value, Mapping): + raise TypeError(f"production input field {name} must be an object") + return dict(value) + + +def _normalize_symbol(symbol: Any) -> str: + normalized = canonicalize_a_share_symbol(symbol) + if not re.fullmatch(r"\d{6}\.(SH|SZ)", normalized): + raise ValueError(f"invalid A-share symbol: {symbol}") + return normalized diff --git a/src/quantpilot_core/continuous_paper/prefect_flow.py b/src/quantpilot_core/continuous_paper/prefect_flow.py index 0e41d6a..0b7a97b 100644 --- a/src/quantpilot_core/continuous_paper/prefect_flow.py +++ b/src/quantpilot_core/continuous_paper/prefect_flow.py @@ -2,13 +2,11 @@ from __future__ import annotations import os -from pathlib import Path from typing import Any -from quantpilot_core.production_candidate import load_runtime_manifest -from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig from .active_shadow import ActiveShadowConfig from .contracts import ContinuousPaperCycleConfig +from .manifest_bridge import load_production_pipeline_config from .service import ContinuousPaperCycle from .store import PostgreSQLReportingStore @@ -18,7 +16,10 @@ def retryable_exception(exc: BaseException) -> bool: if exc.__class__.__name__ in {"ReportingConflictError", "ValueError", "TypeError"}: return False - return any(word in str(exc).lower() for word in ("connection", "timeout", "temporar", "postgres", "network")) + return any( + word in str(exc).lower() + for word in ("connection", "timeout", "temporar", "postgres", "network") + ) def prefect_retry_condition(_task: Any, _task_run: Any, state: Any) -> bool: @@ -26,14 +27,41 @@ def prefect_retry_condition(_task: Any, _task_run: Any, state: Any) -> bool: return retryable_exception(state.result(raise_on_failure=False)) -def run_continuous_paper_flow(*, production_manifest_path: str, decision_session: str, state_path: str, report_path: str, run_id: str, dsn_env_var: str = "QUANTPILOT_POSTGRES_DSN", active_shadow: dict[str, Any] | None = None, **pipeline_options: Any) -> Any: +def run_continuous_paper_flow( + *, + production_manifest_path: str, + decision_session: str, + state_path: str, + report_path: str, + run_id: str, + dsn_env_var: str = "QUANTPILOT_POSTGRES_DSN", + active_shadow: dict[str, Any] | None = None, + live_market_data: bool = False, + input_payload: dict[str, Any] | None = None, + pipeline_options: dict[str, Any] | None = None, +) -> Any: """Construct all runtime objects from serializable scheduled-flow parameters.""" dsn = os.environ.get(dsn_env_var) if not dsn: - raise RuntimeError(f"required PostgreSQL DSN environment variable is not set: {dsn_env_var}") - manifest = load_runtime_manifest(Path(production_manifest_path)) - pipeline = RealCandidatePipelineConfig(decision_session=decision_session, state_path=state_path, report_path=report_path, production_manifest=manifest, **pipeline_options) - config = ContinuousPaperCycleConfig(pipeline_config=pipeline, run_id=run_id, active_shadow=ActiveShadowConfig(**(active_shadow or {}))) + raise RuntimeError( + f"required PostgreSQL DSN environment variable is not set: {dsn_env_var}" + ) + + pipeline = load_production_pipeline_config( + manifest_path=production_manifest_path, + decision_session=decision_session, + state_path=state_path, + report_path=report_path, + live_market_data=live_market_data, + input_payload=input_payload, + pipeline_options=pipeline_options, + ) + + config = ContinuousPaperCycleConfig( + pipeline_config=pipeline, + run_id=run_id, + active_shadow=ActiveShadowConfig(**(active_shadow or {})), + ) return ContinuousPaperCycle(PostgreSQLReportingStore(dsn)).run_once(config) @@ -62,6 +90,8 @@ def continuous_paper_flow( run_id: str, dsn_env_var: str = "QUANTPILOT_POSTGRES_DSN", active_shadow: dict[str, Any] | None = None, + live_market_data: bool = False, + input_payload: dict[str, Any] | None = None, pipeline_options: dict[str, Any] | None = None, ) -> Any: """Prefect-visible parameters plus an explicit serializable option mapping.""" @@ -73,5 +103,7 @@ def continuous_paper_flow( run_id=run_id, dsn_env_var=dsn_env_var, active_shadow=active_shadow, - **(pipeline_options or {}), + live_market_data=live_market_data, + input_payload=input_payload, + pipeline_options=pipeline_options, ) diff --git a/tests/continuous_paper/test_manifest_bridge.py b/tests/continuous_paper/test_manifest_bridge.py new file mode 100644 index 0000000..e4622a0 --- /dev/null +++ b/tests/continuous_paper/test_manifest_bridge.py @@ -0,0 +1,512 @@ +"""Focused tests for the manifest-bridge production boundary.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from quantpilot_core.continuous_paper.manifest_bridge import ( + FROZEN_MANIFEST_PARAMETER_NAMES, + load_production_input_payload, + load_production_pipeline_config, + normalize_production_input_payload, +) +from quantpilot_core.production_candidate import ( + build_production_candidate_manifest, + load_runtime_manifest, + write_manifest_atomic, +) +from quantpilot_core.production_candidate.contracts import ( + ProductionCandidateContractError, +) +from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig + + +# --------------------------------------------------------------------------- +# Shared test helper: build a valid manifest dict and write it to disk. +# --------------------------------------------------------------------------- + + +def _write_valid_manifest(tmp_path: Path, **overrides) -> Path: + """Write a valid runtime manifest JSON file and return its path.""" + a = tmp_path / "pr121.json" + a.write_text("{}") + b = tmp_path / "pr122.json" + b.write_text("{}") + values = dict( + created_at="2026-07-12T00:00:00+00:00", + code_revision="revision", + snapshot_digest="a" * 64, + benchmark={"symbol": "000300.SH"}, + source_artifacts={"pr121": str(a), "pr122": str(b)}, + frozen_strategy_parameters={ + "initial_capital": 100000.0, + "target_symbol_count": 2, + "max_execution_symbols": 6, + "strategy_id": "equal_weight_baseline", + }, + frozen_portfolio_parameters={ + "target_position_count": 2, + "max_position_weight": 0.1, + "reserve_cash_weight": 0.02, + }, + frozen_execution_parameters={"min_order_lot": 100}, + fee_profile_policy={ + "profile_id": "engineering-default", + "required_provenance": "engineering_fallback", + }, + account_capability_policy={"capability_digest": "null"}, + ) + values.update(overrides) + manifest = build_production_candidate_manifest(**values) + path = write_manifest_atomic(manifest, tmp_path / "manifest.json") + return Path(path) + + +# --------------------------------------------------------------------------- +# 1. Manifest file JSON is parsed and passed as Mapping. +# --------------------------------------------------------------------------- + + +def test_manifest_json_parsed_and_passed_as_mapping(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + state_path=str(tmp_path / "state.json"), + report_path=str(tmp_path / "report.json"), + ) + assert isinstance(config, RealCandidatePipelineConfig) + # Verify the manifest was loaded and its frozen params applied. + assert config.initial_capital == 100000.0 + assert config.target_symbol_count == 2 + assert config.max_execution_symbols == 6 + assert config.production_manifest is not None + + +def test_manifest_file_not_a_json_object_is_rejected(tmp_path: Path) -> None: + bad = tmp_path / "bad.json" + bad.write_text("[1, 2, 3]") + with pytest.raises(TypeError, match="JSON object"): + load_production_pipeline_config( + manifest_path=bad, + decision_session="2026-04-03", + ) + + +def test_input_file_not_a_json_object_is_rejected(tmp_path: Path) -> None: + bad = tmp_path / "input.json" + bad.write_text("[1, 2, 3]") + with pytest.raises(TypeError, match="input JSON.*object"): + load_production_input_payload(bad) + + +def test_input_file_uses_shared_field_validation(tmp_path: Path) -> None: + bad = tmp_path / "input.json" + bad.write_text('{"symbols": "600000.SH"}') + with pytest.raises(TypeError, match="symbols must be an array"): + load_production_input_payload(bad) + + +def test_shared_payload_schema_normalizes_every_supported_field() -> None: + payload = normalize_production_input_payload( + { + "symbols": ["sh.600000", "600000.SH", "000001"], + "bars": [{"symbol": "600000.SH"}], + "information_signals": [ + {"signal": {"target": "600000.SH"}, "available_at": "timestamp"} + ], + "information_provenance": {"provider": "fixture"}, + "advisory_provenance": {"deepseek_live_call": False}, + "quant_firm_context": {"requested_action": "hold"}, + } + ) + assert payload == { + "symbols": ("600000.SH", "000001.SZ"), + "input_bars": ({"symbol": "600000.SH"},), + "information_signals": ( + {"signal": {"target": "600000.SH"}, "available_at": "timestamp"}, + ), + "information_provenance": {"provider": "fixture"}, + "advisory_provenance": {"deepseek_live_call": False}, + "quant_firm_context": {"requested_action": "hold"}, + } + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ([], "payload must be an object"), + ({"symbols": "600000.SH"}, "symbols must be an array"), + ({"bars": [1]}, r"bars\[0\] must be an object"), + ({"information_signals": ["buy"]}, r"information_signals\[0\]"), + ({"information_provenance": []}, "information_provenance must be an object"), + ({"unknown": 1}, "unknown production input fields"), + ], +) +def test_shared_payload_schema_rejects_invalid_values(payload, message) -> None: + with pytest.raises((TypeError, ValueError), match=message): + normalize_production_input_payload(payload) + + +# --------------------------------------------------------------------------- +# 2. Old Path-to-load_runtime_manifest failure is eliminated. +# --------------------------------------------------------------------------- + + +def test_path_passed_to_bridge_not_to_load_runtime_manifest(tmp_path: Path) -> None: + """The shared helper reads the file; callers never pass a Path to + load_runtime_manifest directly.""" + manifest_path = _write_valid_manifest(tmp_path) + # This must succeed — the bridge reads the file and passes a Mapping. + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + state_path=str(tmp_path / "state.json"), + report_path=str(tmp_path / "report.json"), + ) + assert config.target_symbol_count == 2 + + +def test_raw_path_still_rejected_by_strict_loader() -> None: + """Sanity: load_runtime_manifest itself still rejects a Path.""" + with pytest.raises(TypeError, match="ProductionCandidateManifest or mapping"): + load_runtime_manifest(Path("/nonexistent/manifest.json")) + + +# --------------------------------------------------------------------------- +# 3. Manifest frozen parameters are actually applied to config. +# --------------------------------------------------------------------------- + + +def test_frozen_parameters_applied_to_pipeline_config(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + ) + assert config.initial_capital == 100000.0 + assert config.target_symbol_count == 2 + assert config.max_execution_symbols == 6 + assert config.target_position_count == 2 + assert config.max_position_weight == 0.1 + assert config.reserve_cash_weight == 0.02 + assert config.min_order_lot == 100 + assert config.strategy_id == "equal_weight_baseline" + assert config.capital_profile_id == "default_paper_capital" + + +def test_frozen_params_override_dataclass_defaults(tmp_path: Path) -> None: + """Frozen manifest values must be authoritative, not dataclass defaults.""" + manifest_path = _write_valid_manifest( + tmp_path, + frozen_strategy_parameters={ + "initial_capital": 250000.0, + "target_symbol_count": 4, + "max_execution_symbols": 5, + "strategy_id": "equal_weight_baseline", + }, + frozen_portfolio_parameters={ + "target_position_count": 4, + "max_position_weight": 0.15, + "reserve_cash_weight": 0.05, + }, + frozen_execution_parameters={"min_order_lot": 200}, + ) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + ) + # Dataclass defaults: initial_capital=100000, max_execution_symbols=6, + # target_symbol_count=1, max_position_weight=0.10, reserve_cash_weight=0.02, + # min_order_lot=100. All must be overridden by frozen values. + assert config.initial_capital == 250000.0 + assert config.target_symbol_count == 4 + assert config.max_execution_symbols == 5 + assert config.max_position_weight == 0.15 + assert config.reserve_cash_weight == 0.05 + assert config.min_order_lot == 200 + + +# --------------------------------------------------------------------------- +# 4. A conflicting caller override cannot replace frozen production values. +# --------------------------------------------------------------------------- + + +def test_frozen_params_are_authoritative_in_constructed_config(tmp_path: Path) -> None: + """Once loaded, the RealCandidatePipelineConfig carries the frozen values + and the production_manifest. The pipeline itself will reject a runtime + parameter that conflicts with a frozen value (via effective_parameter_payload).""" + manifest_path = _write_valid_manifest(tmp_path) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + ) + # The config was built from the manifest; the manifest is attached. + assert config.production_manifest is not None + # The manifest's frozen initial_capital is 100000; verify it is set. + assert config.initial_capital == 100000.0 + # The effective_parameter_payload function (called by the pipeline) will + # reject any runtime parameter that conflicts with a frozen value. + from quantpilot_core.production_candidate import effective_parameter_payload + + runtime = { + "initial_capital": 100000.0, + "target_symbol_count": 2, + "max_execution_symbols": 6, + "target_position_count": 2, + "max_position_weight": 0.1, + "reserve_cash_weight": 0.02, + "min_order_lot": 100, + "strategy_id": "equal_weight_baseline", + "capital_profile_id": "default_paper_capital", + } + # Matching runtime → OK. + result = effective_parameter_payload( + config.production_manifest, runtime_parameters=runtime + ) + assert result["target_symbol_count"] == 2 + + # Conflicting runtime → rejected. + with pytest.raises(ProductionCandidateContractError, match="frozen parameter mismatch"): + effective_parameter_payload( + config.production_manifest, + runtime_parameters={**runtime, "target_symbol_count": 99}, + ) + + +@pytest.mark.parametrize("name", sorted(FROZEN_MANIFEST_PARAMETER_NAMES)) +def test_pipeline_options_cannot_override_any_frozen_parameter( + tmp_path: Path, name: str +) -> None: + manifest_path = _write_valid_manifest(tmp_path) + with pytest.raises(ValueError, match="frozen manifest parameters"): + load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + pipeline_options={name: "caller-value"}, + ) + + +def test_serialized_pipeline_options_become_runtime_contracts(tmp_path: Path) -> None: + config = load_production_pipeline_config( + manifest_path=_write_valid_manifest(tmp_path), + decision_session="2026-04-03", + pipeline_options={ + "input_calendar_sessions": ["2026-04-03", "2026-04-06"], + "input_calendar_provider": "fixture", + "account_capabilities": { + "account_id": "paper-account", + "supported_order_types": ["limit"], + }, + }, + ) + assert config.input_calendar_sessions == ("2026-04-03", "2026-04-06") + assert config.input_calendar_provider == "fixture" + assert config.account_capabilities.account_id == "paper-account" + assert config.account_capabilities.supported_order_types == ("limit",) + + +# --------------------------------------------------------------------------- +# 5. target_symbol_count > max_execution_symbols is rejected clearly. +# --------------------------------------------------------------------------- + + +def test_target_symbol_count_exceeds_max_execution_rejected(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest( + tmp_path, + frozen_strategy_parameters={ + "initial_capital": 100000.0, + "target_symbol_count": 10, + "max_execution_symbols": 6, + "strategy_id": "equal_weight_baseline", + }, + ) + with pytest.raises(ValueError, match="target_symbol_count.*exceeds max_execution_symbols"): + load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + ) + + +def test_compatible_target_and_max_symbols_are_accepted(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest( + tmp_path, + frozen_strategy_parameters={ + "initial_capital": 100000.0, + "target_symbol_count": 6, + "max_execution_symbols": 6, + "strategy_id": "equal_weight_baseline", + }, + frozen_portfolio_parameters={ + "target_position_count": 6, + "max_position_weight": 0.1, + "reserve_cash_weight": 0.02, + }, + ) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + ) + assert config.target_symbol_count == 6 + assert config.max_execution_symbols == 6 + + +# --------------------------------------------------------------------------- +# 6. Live mode without symbols is rejected. +# --------------------------------------------------------------------------- + + +def test_live_mode_without_symbols_rejected(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + with pytest.raises(ValueError, match="live_market_data requires.*symbols"): + load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + live_market_data=True, + input_payload={"symbols": []}, + ) + + +# --------------------------------------------------------------------------- +# 7. Live mode with explicit bounded symbol universe is forwarded. +# --------------------------------------------------------------------------- + + +def test_live_mode_with_bounded_symbols_is_forwarded(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + live_market_data=True, + input_payload={"symbols": ["sh.600000", "000001"]}, + ) + assert config.live_market_data is True + assert config.symbols == ("600000.SH", "000001.SZ") + assert len(config.symbols) <= config.max_execution_symbols + + +def test_live_mode_rejects_more_than_six_normalized_symbols(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + with pytest.raises(ValueError, match="at most 6"): + load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + live_market_data=True, + input_payload={ + "symbols": [ + "600000.SH", + "600001.SH", + "600002.SH", + "600003.SH", + "600004.SH", + "600005.SH", + "600006.SH", + ] + }, + ) + + +# --------------------------------------------------------------------------- +# 8. Offline mode remains deterministic and offline-safe. +# --------------------------------------------------------------------------- + + +def test_offline_mode_no_live_market_data_by_default(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + ) + assert config.live_market_data is False + assert config.symbols == () + assert config.input_bars == () + + +def test_offline_mode_with_input_bars_stays_offline(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + bars = ( + { + "symbol": "600000.SH", + "date": "2026-04-03", + "open": 10.0, + "high": 10.5, + "low": 9.8, + "close": 10.2, + "previous_close": 10.0, + "volume": 100000.0, + "amount": 1020000.0, + "is_suspended": False, + "provider": "test", + }, + ) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + live_market_data=False, + input_payload={"bars": list(bars)}, + ) + assert config.live_market_data is False + assert len(config.input_bars) == 1 + + +# --------------------------------------------------------------------------- +# 9. Active shadow remains non-authoritative. +# --------------------------------------------------------------------------- + + +def test_shadow_desk_evidence_is_forwarded_but_not_authoritative(tmp_path: Path) -> None: + manifest_path = _write_valid_manifest(tmp_path) + shadow = {"desk_vote": "abstain", "recommendation": "hold"} + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + pipeline_options={"shadow_desk_evidence": shadow}, + ) + # Shadow evidence is stored but never drives the production config. + assert config.shadow_desk_evidence == shadow + # Production execution arm must remain non_llm_baseline (enforced by + # the manifest, not by shadow evidence). + manifest = load_runtime_manifest( + json.loads(Path(manifest_path).read_text(encoding="utf-8")) + ) + assert manifest.production_execution_arm == "non_llm_baseline" + + +# --------------------------------------------------------------------------- +# 10. No broker/order-submission path is enabled. +# --------------------------------------------------------------------------- + + +def test_no_broker_execution_path_in_pipeline_config(tmp_path: Path) -> None: + """RealCandidatePipelineConfig has no broker/order fields — execution is + paper-only by construction.""" + manifest_path = _write_valid_manifest(tmp_path) + config = load_production_pipeline_config( + manifest_path=manifest_path, + decision_session="2026-04-03", + ) + # Verify there is no broker-related attribute. + for attr in dir(config): + assert "broker" not in attr.lower() + assert "order_submit" not in attr.lower() + assert "live_trading" not in attr.lower() + + +def test_broker_provider_remains_none_in_manifest(tmp_path: Path) -> None: + """The production manifest enforces non_llm_baseline execution arm.""" + manifest_path = _write_valid_manifest(tmp_path) + manifest = load_runtime_manifest( + json.loads(Path(manifest_path).read_text(encoding="utf-8")) + ) + assert manifest.production_execution_arm == "non_llm_baseline" + # Any attempt to use a live execution arm is rejected at manifest level. + sub = tmp_path / "sub" + sub.mkdir() + with pytest.raises(ProductionCandidateContractError, match="non_llm_baseline"): + _write_valid_manifest( + sub, + production_execution_arm="live_broker", + ) diff --git a/tests/continuous_paper/test_prefect_and_active_shadow.py b/tests/continuous_paper/test_prefect_and_active_shadow.py index a3517f7..cf1a236 100644 --- a/tests/continuous_paper/test_prefect_and_active_shadow.py +++ b/tests/continuous_paper/test_prefect_and_active_shadow.py @@ -9,6 +9,22 @@ from quantpilot_core.continuous_paper.active_shadow import ActiveShadowConfig, ActiveShadowRunner from quantpilot_core.quant_firm.deepseek_advisory import DeepSeekAdvisoryRole +from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig + + +def _assert_stable_serializable(value) -> None: + if value is None or isinstance(value, (str, bool, int, float)): + return + if isinstance(value, list): + for item in value: + _assert_stable_serializable(item) + return + if isinstance(value, dict): + assert all(isinstance(key, str) for key in value) + for item in value.values(): + _assert_stable_serializable(item) + return + raise AssertionError(f"non-serializable deployment value: {type(value).__name__}") def test_prefect_module_import_has_no_scheduling_side_effect(monkeypatch) -> None: @@ -72,6 +88,9 @@ def fake_cron(expression, *, timezone): "--report-path", "report.json", "--run-id", "run-1", "--timezone", "Asia/Shanghai", + "--max-physical-model-calls", "2", + "--max-estimated-api-cost", "1.5", + "--estimated-cost-per-call", "0.5", ] with monkeypatch.context() as scoped: scoped.setitem(sys.modules, "quantpilot_core.continuous_paper.prefect_flow", fake_flow_module) @@ -85,6 +104,197 @@ def fake_cron(expression, *, timezone): assert serve_kwargs["schedules"] == ["schedule"] assert "timezone" not in serve_kwargs and "cron" not in serve_kwargs assert serve_kwargs["parameters"]["decision_session"] == "2026-07-13" + shadow = serve_kwargs["parameters"]["active_shadow"] + assert shadow["enabled"] is False + assert shadow["enable_live_calls"] is False + assert shadow["max_physical_model_calls_per_cycle"] == 2 + _assert_stable_serializable(serve_kwargs["parameters"]) + + +def test_serve_helper_explicitly_enables_shadow_and_live_ai(monkeypatch) -> None: + from scripts import serve_continuous_paper_v1 as serve_script + + captured: dict[str, object] = {} + + class FakeFlow: + def serve(self, **kwargs): + captured.update(kwargs) + + fake_flow_module = types.ModuleType( + "quantpilot_core.continuous_paper.prefect_flow" + ) + fake_flow_module.continuous_paper_flow = FakeFlow() + fake_prefect = types.ModuleType("prefect") + fake_prefect.__path__ = [] + fake_schedules = types.ModuleType("prefect.schedules") + fake_schedules.Cron = lambda expression, *, timezone: { + "expression": expression, + "timezone": timezone, + } + argv = [ + "serve_continuous_paper_v1.py", + "--production-manifest", + "manifest.json", + "--decision-session", + "2026-07-13", + "--state-path", + "state.json", + "--report-path", + "report.json", + "--run-id", + "run-1", + "--enable-active-shadow", + "--enable-live-ai", + "--allowed-roles", + "research-desk, investment_committee,research_desk", + "--max-physical-model-calls", + "2", + "--max-estimated-api-cost", + "1.5", + "--estimated-cost-per-call", + "0.5", + ] + with monkeypatch.context() as scoped: + scoped.setitem( + sys.modules, + "quantpilot_core.continuous_paper.prefect_flow", + fake_flow_module, + ) + scoped.setitem(sys.modules, "prefect", fake_prefect) + scoped.setitem(sys.modules, "prefect.schedules", fake_schedules) + scoped.setattr(sys, "argv", argv) + serve_script.main() + + shadow = captured["parameters"]["active_shadow"] + assert shadow["enabled"] is True + assert shadow["enable_live_calls"] is True + assert shadow["allow_roles"] == [ + "research_desk", + "investment_committee", + ] + _assert_stable_serializable(captured["parameters"]) + + +def test_serve_helper_rejects_live_ai_without_positive_limits(monkeypatch) -> None: + from scripts import serve_continuous_paper_v1 as serve_script + + argv = [ + "serve_continuous_paper_v1.py", + "--production-manifest", + "manifest.json", + "--decision-session", + "2026-07-13", + "--state-path", + "state.json", + "--report-path", + "report.json", + "--run-id", + "run-1", + "--enable-active-shadow", + "--enable-live-ai", + ] + monkeypatch.setattr(sys, "argv", argv) + with pytest.raises(SystemExit): + serve_script.main() + + +def test_one_cycle_cli_forwards_normalized_allowed_roles_and_shared_payload( + monkeypatch, capsys +) -> None: + from scripts import run_continuous_paper_cycle_v1 as run_script + + captured: dict[str, object] = {} + + def fake_load_pipeline(**kwargs): + captured["pipeline_kwargs"] = kwargs + return RealCandidatePipelineConfig(decision_session="2026-07-13") + + class FakeCycle: + def __init__(self, _store): + pass + + def run_once(self, config): + captured["cycle_config"] = config + return types.SimpleNamespace( + session_id="session-1", status="completed", run_id=config.run_id + ) + + argv = [ + "run_continuous_paper_cycle_v1.py", + "--production-manifest", + "manifest.json", + "--decision-session", + "2026-07-13", + "--state-path", + "state.json", + "--report-path", + "report.json", + "--test-store", + "--enable-active-shadow", + "--allowed-roles", + " research-desk,investment_committee,research_desk ", + ] + with monkeypatch.context() as scoped: + scoped.setattr(run_script, "load_production_pipeline_config", fake_load_pipeline) + scoped.setattr(run_script, "ContinuousPaperCycle", FakeCycle) + scoped.setattr(sys, "argv", argv) + run_script.main() + + assert captured["pipeline_kwargs"]["input_payload"] == {} + assert "symbols" not in captured["pipeline_kwargs"] + assert captured["cycle_config"].active_shadow.allow_roles == ( + DeepSeekAdvisoryRole.RESEARCH_DESK, + DeepSeekAdvisoryRole.INVESTMENT_COMMITTEE, + ) + assert "status=completed" in capsys.readouterr().out + + +def test_flow_forwards_one_payload_and_serialized_options_to_shared_bridge( + monkeypatch, +) -> None: + import quantpilot_core.continuous_paper.prefect_flow as module + + captured: dict[str, object] = {} + + def fake_load_pipeline(**kwargs): + captured["pipeline_kwargs"] = kwargs + return RealCandidatePipelineConfig(decision_session="2026-07-13") + + class FakeCycle: + def __init__(self, _store): + pass + + def run_once(self, config): + captured["cycle_config"] = config + return "completed" + + payload = {"symbols": ["600000.SH"]} + options = {"input_calendar_sessions": ["2026-07-13"]} + with monkeypatch.context() as scoped: + scoped.setenv("TEST_CONTINUOUS_PAPER_DSN", "postgresql://fixture") + scoped.setattr(module, "load_production_pipeline_config", fake_load_pipeline) + scoped.setattr(module, "PostgreSQLReportingStore", lambda _dsn: object()) + scoped.setattr(module, "ContinuousPaperCycle", FakeCycle) + result = module.run_continuous_paper_flow( + production_manifest_path="manifest.json", + decision_session="2026-07-13", + state_path="state.json", + report_path="report.json", + run_id="run-1", + dsn_env_var="TEST_CONTINUOUS_PAPER_DSN", + active_shadow={"allow_roles": ["research_desk"]}, + live_market_data=True, + input_payload=payload, + pipeline_options=options, + ) + + assert result == "completed" + assert captured["pipeline_kwargs"]["input_payload"] is payload + assert captured["pipeline_kwargs"]["pipeline_options"] is options + assert "symbols" not in captured["pipeline_kwargs"] + assert captured["cycle_config"].active_shadow.allow_roles == ( + DeepSeekAdvisoryRole.RESEARCH_DESK, + ) def test_real_prefect_flow_smoke_has_explicit_parameters_and_no_deployment(monkeypatch) -> None: