From e4bae8fcea365435dc0e316f880cf1f5b4514b2a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 16 Apr 2026 19:59:44 +0800 Subject: [PATCH] Align LongBridge strategy runtime contracts --- .github/workflows/sync-cloud-run-env.yml | 46 ++---- runtime_config_support.py | 87 +++-------- scripts/print_strategy_switch_env_plan.py | 20 ++- strategy_runtime.py | 167 +++------------------- tests/test_runtime_config_support.py | 34 ++++- tests/test_strategy_runtime.py | 2 + tests/test_sync_cloud_run_env_workflow.sh | 14 +- 7 files changed, 105 insertions(+), 265 deletions(-) diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 9709192..5efbc2f 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -80,10 +80,7 @@ jobs: import os import subprocess import sys - from pathlib import Path - - from quant_platform_kit.common.strategies import derive_strategy_artifact_paths - from us_equity_strategies import get_strategy_catalog, resolve_canonical_profile + from us_equity_strategies import resolve_canonical_profile profile = os.environ.get("STRATEGY_PROFILE", "").strip().lower() if not profile: @@ -102,16 +99,6 @@ jobs: if not selected.get("eligible") or not selected.get("enabled"): raise SystemExit(f"STRATEGY_PROFILE={profile!r} is not eligible/enabled: {selected}") - artifact_paths = derive_strategy_artifact_paths( - get_strategy_catalog(), - canonical_profile, - repo_root=Path.cwd(), - ) - has_bundled_strategy_config = bool( - artifact_paths.bundled_config_path is not None - and Path(artifact_paths.bundled_config_path).exists() - ) - output_path = os.environ["GITHUB_OUTPUT"] with open(output_path, "a", encoding="utf-8") as output: output.write( @@ -124,7 +111,7 @@ jobs: f"requires_strategy_config_path={str(bool(selected.get('requires_strategy_config_path'))).lower()}\n" ) output.write( - f"has_bundled_strategy_config={str(has_bundled_strategy_config).lower()}\n" + f"config_source_policy={str(selected.get('config_source_policy') or 'none')}\n" ) PY @@ -134,7 +121,7 @@ jobs: REQUIRES_SNAPSHOT_ARTIFACTS: ${{ steps.strategy_requirements.outputs.requires_snapshot_artifacts }} REQUIRES_SNAPSHOT_MANIFEST_PATH: ${{ steps.strategy_requirements.outputs.requires_snapshot_manifest_path }} REQUIRES_STRATEGY_CONFIG_PATH: ${{ steps.strategy_requirements.outputs.requires_strategy_config_path }} - HAS_BUNDLED_STRATEGY_CONFIG: ${{ steps.strategy_requirements.outputs.has_bundled_strategy_config }} + CONFIG_SOURCE_POLICY: ${{ steps.strategy_requirements.outputs.config_source_policy }} run: | set -euo pipefail @@ -175,8 +162,8 @@ jobs: fi if [ "${REQUIRES_STRATEGY_CONFIG_PATH:-}" = "true" ] \ - && [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ] \ - && [ "${HAS_BUNDLED_STRATEGY_CONFIG:-}" != "true" ]; then + && [ "${CONFIG_SOURCE_POLICY:-}" = "env_only" ] \ + && [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ]; then missing_vars+=("LONGBRIDGE_STRATEGY_CONFIG_PATH") fi @@ -390,10 +377,7 @@ jobs: import os import subprocess import sys - from pathlib import Path - - from quant_platform_kit.common.strategies import derive_strategy_artifact_paths - from us_equity_strategies import get_strategy_catalog, resolve_canonical_profile + from us_equity_strategies import resolve_canonical_profile profile = os.environ.get("STRATEGY_PROFILE", "").strip().lower() if not profile: @@ -412,16 +396,6 @@ jobs: if not selected.get("eligible") or not selected.get("enabled"): raise SystemExit(f"STRATEGY_PROFILE={profile!r} is not eligible/enabled: {selected}") - artifact_paths = derive_strategy_artifact_paths( - get_strategy_catalog(), - canonical_profile, - repo_root=Path.cwd(), - ) - has_bundled_strategy_config = bool( - artifact_paths.bundled_config_path is not None - and Path(artifact_paths.bundled_config_path).exists() - ) - output_path = os.environ["GITHUB_OUTPUT"] with open(output_path, "a", encoding="utf-8") as output: output.write( @@ -434,7 +408,7 @@ jobs: f"requires_strategy_config_path={str(bool(selected.get('requires_strategy_config_path'))).lower()}\n" ) output.write( - f"has_bundled_strategy_config={str(has_bundled_strategy_config).lower()}\n" + f"config_source_policy={str(selected.get('config_source_policy') or 'none')}\n" ) PY @@ -444,7 +418,7 @@ jobs: REQUIRES_SNAPSHOT_ARTIFACTS: ${{ steps.strategy_requirements.outputs.requires_snapshot_artifacts }} REQUIRES_SNAPSHOT_MANIFEST_PATH: ${{ steps.strategy_requirements.outputs.requires_snapshot_manifest_path }} REQUIRES_STRATEGY_CONFIG_PATH: ${{ steps.strategy_requirements.outputs.requires_strategy_config_path }} - HAS_BUNDLED_STRATEGY_CONFIG: ${{ steps.strategy_requirements.outputs.has_bundled_strategy_config }} + CONFIG_SOURCE_POLICY: ${{ steps.strategy_requirements.outputs.config_source_policy }} run: | set -euo pipefail @@ -485,8 +459,8 @@ jobs: fi if [ "${REQUIRES_STRATEGY_CONFIG_PATH:-}" = "true" ] \ - && [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ] \ - && [ "${HAS_BUNDLED_STRATEGY_CONFIG:-}" != "true" ]; then + && [ "${CONFIG_SOURCE_POLICY:-}" = "env_only" ] \ + && [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ]; then missing_vars+=("LONGBRIDGE_STRATEGY_CONFIG_PATH") fi diff --git a/runtime_config_support.py b/runtime_config_support.py index 57fe1b3..74e6d32 100644 --- a/runtime_config_support.py +++ b/runtime_config_support.py @@ -5,7 +5,10 @@ from pathlib import Path from typing import Callable -from quant_platform_kit.common.strategies import derive_strategy_artifact_paths +from quant_platform_kit.common.runtime_config import ( + resolve_bool_value, + resolve_strategy_runtime_path_settings, +) from strategy_registry import ( LONGBRIDGE_PLATFORM, resolve_strategy_definition, @@ -73,34 +76,21 @@ def load_platform_runtime_settings( strategy_definition.profile, platform_id=LONGBRIDGE_PLATFORM, ) - artifact_root = _first_non_empty( - os.getenv("LONGBRIDGE_STRATEGY_ARTIFACT_ROOT"), - os.getenv("STRATEGY_ARTIFACT_ROOT"), - ) - derived_artifact_paths = derive_strategy_artifact_paths( - get_strategy_catalog(), - strategy_definition.profile, - artifact_root=artifact_root, + runtime_paths = resolve_strategy_runtime_path_settings( + strategy_catalog=get_strategy_catalog(), + strategy_definition=strategy_definition, + strategy_metadata=strategy_metadata, + platform_env_prefix="LONGBRIDGE", + env=os.environ, repo_root=Path(__file__).resolve().parent, ) - strategy_config_path, strategy_config_source = resolve_strategy_config_path( - explicit_path=_first_non_empty( - os.getenv("LONGBRIDGE_STRATEGY_CONFIG_PATH"), - os.getenv("STRATEGY_CONFIG_PATH"), - ), - bundled_path=( - str(derived_artifact_paths.bundled_config_path) - if derived_artifact_paths.bundled_config_path is not None - else None - ), - ) return PlatformRuntimeSettings( project_id=project_id_resolver(), secret_name=os.getenv("LONGPORT_SECRET_NAME", DEFAULT_LONGPORT_SECRET_NAME), account_prefix=account_prefix, - strategy_profile=strategy_definition.profile, - strategy_display_name=strategy_metadata.display_name, - strategy_domain=strategy_definition.domain, + strategy_profile=runtime_paths.strategy_profile, + strategy_display_name=runtime_paths.strategy_display_name, + strategy_domain=runtime_paths.strategy_domain, account_region=infer_account_region( os.getenv("ACCOUNT_REGION"), account_prefix=account_prefix, @@ -108,43 +98,16 @@ def load_platform_runtime_settings( notify_lang=os.getenv("NOTIFY_LANG", "en"), tg_token=os.getenv("TELEGRAM_TOKEN"), tg_chat_id=os.getenv("GLOBAL_TELEGRAM_CHAT_ID"), - dry_run_only=_resolve_bool_env("LONGBRIDGE_DRY_RUN_ONLY"), + dry_run_only=resolve_bool_value(os.getenv("LONGBRIDGE_DRY_RUN_ONLY")), income_threshold_usd=_optional_float_env("INCOME_THRESHOLD_USD"), qqqi_income_ratio=_qqqi_income_ratio_env(), - feature_snapshot_path=_first_non_empty( - os.getenv("LONGBRIDGE_FEATURE_SNAPSHOT_PATH"), - os.getenv("FEATURE_SNAPSHOT_PATH"), - str(derived_artifact_paths.feature_snapshot_path) - if derived_artifact_paths.feature_snapshot_path is not None - else None, - ), - feature_snapshot_manifest_path=_first_non_empty( - os.getenv("LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"), - os.getenv("FEATURE_SNAPSHOT_MANIFEST_PATH"), - str(derived_artifact_paths.feature_snapshot_manifest_path) - if derived_artifact_paths.feature_snapshot_manifest_path is not None - else None, - ), - strategy_config_path=strategy_config_path, - strategy_config_source=strategy_config_source, + feature_snapshot_path=runtime_paths.feature_snapshot_path, + feature_snapshot_manifest_path=runtime_paths.feature_snapshot_manifest_path, + strategy_config_path=runtime_paths.strategy_config_path, + strategy_config_source=runtime_paths.strategy_config_source, ) -def resolve_strategy_config_path( - *, - explicit_path: str | None, - bundled_path: str | None, -) -> tuple[str | None, str | None]: - path = _first_non_empty(explicit_path) - if path is not None: - return path, "env" - - bundled = _first_non_empty(bundled_path) - if bundled is not None and Path(bundled).exists(): - return bundled, "bundled_canonical_default" - return None, None - - def _normalize_region(raw_value: str | None) -> str | None: if raw_value is None: return None @@ -154,13 +117,6 @@ def _normalize_region(raw_value: str | None) -> str | None: return value.upper() -def _resolve_bool_env(name: str) -> bool: - raw_value = os.getenv(name) - if raw_value is None: - return False - return str(raw_value).strip().lower() in {"1", "true", "yes", "on"} - - def _optional_float_env(name: str) -> float | None: raw_value = os.getenv(name) if raw_value is None or raw_value.strip() == "": @@ -174,10 +130,3 @@ def _qqqi_income_ratio_env() -> float | None: raise ValueError(f"QQQI_INCOME_RATIO must be in [0,1], got {value}") return value - -def _first_non_empty(*values: str | None) -> str | None: - for value in values: - text = str(value or "").strip() - if text: - return text - return None diff --git a/scripts/print_strategy_switch_env_plan.py b/scripts/print_strategy_switch_env_plan.py index 287e069..fe41ad5 100644 --- a/scripts/print_strategy_switch_env_plan.py +++ b/scripts/print_strategy_switch_env_plan.py @@ -42,7 +42,11 @@ def build_switch_plan(profile: str, *, account_region: str | None = None) -> dic platform_id=LONGBRIDGE_PLATFORM, ) requires_feature_snapshot = bool(runtime_requirements["requires_snapshot_artifacts"]) + requires_snapshot_manifest_path = bool( + runtime_requirements["requires_snapshot_manifest_path"] + ) requires_strategy_config_path = bool(runtime_requirements["requires_strategy_config_path"]) + config_source_policy = str(runtime_requirements.get("config_source_policy") or "none") normalized_region = (account_region or "").strip().upper() set_env: dict[str, str] = {"STRATEGY_PROFILE": definition.profile} @@ -66,9 +70,19 @@ def build_switch_plan(profile: str, *, account_region: str | None = None) -> dic if requires_feature_snapshot: set_env["LONGBRIDGE_FEATURE_SNAPSHOT_PATH"] = "" - set_env["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"] = "" - if requires_strategy_config_path and artifact_paths.bundled_config_path is not None: - set_env["LONGBRIDGE_STRATEGY_CONFIG_PATH"] = str(artifact_paths.bundled_config_path) + if requires_snapshot_manifest_path: + set_env["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"] = "" + else: + remove_if_present.append("LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH") + if requires_strategy_config_path and config_source_policy == "env_only": + set_env["LONGBRIDGE_STRATEGY_CONFIG_PATH"] = "" + elif requires_strategy_config_path and config_source_policy == "bundled_or_env": + remove_if_present.append("LONGBRIDGE_STRATEGY_CONFIG_PATH") + notes.append( + "LONGBRIDGE_STRATEGY_CONFIG_PATH is optional for bundled_or_env profiles; leave it unset to use the packaged canonical config." + ) + elif requires_strategy_config_path: + set_env["LONGBRIDGE_STRATEGY_CONFIG_PATH"] = "" else: remove_if_present.append("LONGBRIDGE_STRATEGY_CONFIG_PATH") else: diff --git a/strategy_runtime.py b/strategy_runtime.py index f1b7aed..fddf0f0 100644 --- a/strategy_runtime.py +++ b/strategy_runtime.py @@ -5,6 +5,10 @@ from typing import Any, Callable, Mapping from quant_platform_kit.common.feature_snapshot import load_feature_snapshot_guarded +from quant_platform_kit.common.feature_snapshot_runtime import ( + FeatureSnapshotRuntimeSettings, + evaluate_feature_snapshot_strategy, +) from quant_platform_kit.strategy_contracts import ( StrategyDecision, StrategyEntrypoint, @@ -20,7 +24,6 @@ _FEATURE_SNAPSHOT_INPUT = "feature_snapshot" -_SINGLE_EXECUTION_WINDOW_PROFILES = frozenset({"tech_communication_pullback_enhancement"}) @dataclass(frozen=True) @@ -92,159 +95,29 @@ def _evaluate_feature_snapshot_strategy( runtime_config: Mapping[str, Any], available_inputs: Mapping[str, Any], ) -> StrategyEvaluationResult: - feature_snapshot_path = self.runtime_settings.feature_snapshot_path - runtime_config_name = str( - self.merged_runtime_config.get("runtime_config_name") - or self.runtime_settings.strategy_profile - ) - runtime_config_path = ( - self.merged_runtime_config.get("runtime_config_path") - or self.runtime_settings.strategy_config_path - ) - runtime_config_source = ( - self.merged_runtime_config.get("runtime_config_source") - or self.runtime_settings.strategy_config_source - ) - safe_haven_symbol = str(self.merged_runtime_config.get("safe_haven") or "BOXX").strip().upper() or None - - if not feature_snapshot_path: - metadata = { - "strategy_profile": self.profile, - "strategy_display_name": self.display_name, - "feature_snapshot_path": None, - "strategy_config_path": runtime_config_path, - "strategy_config_source": runtime_config_source, - "dry_run_only": self.runtime_settings.dry_run_only, - "snapshot_guard_decision": "fail_closed", - "fail_reason": "feature_snapshot_path_missing", - "managed_symbols": self.managed_symbols, - "safe_haven_symbol": safe_haven_symbol, - "status_icon": "🛑", - } - decision = StrategyDecision( - risk_flags=("no_execute",), - diagnostics={ - "signal_description": "feature snapshot required", - "status_description": "fail_closed | reason=feature_snapshot_path_missing", - "actionable": False, - "snapshot_guard_decision": "fail_closed", - "fail_reason": "feature_snapshot_path_missing", - }, - ) - return StrategyEvaluationResult(decision=decision, metadata=metadata) - - evaluation_as_of = datetime.now(timezone.utc) - runtime_config = dict(runtime_config) - runtime_config.setdefault("run_as_of", evaluation_as_of) - if self.profile in _SINGLE_EXECUTION_WINDOW_PROFILES: - runtime_config.setdefault("runtime_execution_window_trading_days", 1) - - guard_result = load_feature_snapshot_guarded( - feature_snapshot_path, - run_as_of=evaluation_as_of, - required_columns=self._required_feature_columns(), - snapshot_date_columns=self._snapshot_date_columns(), - max_snapshot_month_lag=self._max_snapshot_month_lag(), - manifest_path=self.runtime_settings.feature_snapshot_manifest_path, - require_manifest=self._require_snapshot_manifest(), - expected_strategy_profile=self.profile, - expected_config_name=runtime_config_name, - expected_config_path=runtime_config_path, - expected_contract_version=self._snapshot_contract_version(), - ) - guard_metadata = dict(guard_result.metadata) - if guard_metadata.get("snapshot_guard_decision") != "proceed": - decision_text = str(guard_metadata.get("snapshot_guard_decision") or "fail_closed") - reason = guard_metadata.get("fail_reason") or guard_metadata.get("no_op_reason") - metadata = { - "strategy_profile": self.profile, - "strategy_display_name": self.display_name, - "strategy_config_path": runtime_config_path, - "strategy_config_source": runtime_config_source, - "dry_run_only": self.runtime_settings.dry_run_only, - "managed_symbols": self.managed_symbols, - "safe_haven_symbol": safe_haven_symbol, - "status_icon": "🛑", - **guard_metadata, - } - decision = StrategyDecision( - risk_flags=("no_execute",), - diagnostics={ - "signal_description": "feature snapshot guard blocked execution", - "status_description": f"{decision_text} | reason={reason}", - "actionable": False, - "snapshot_guard_decision": decision_text, - "fail_reason": guard_metadata.get("fail_reason"), - "no_op_reason": guard_metadata.get("no_op_reason"), - }, - ) - return StrategyEvaluationResult(decision=decision, metadata=metadata) - - feature_snapshot = guard_result.frame - evaluation_inputs = dict(available_inputs) - evaluation_inputs[_FEATURE_SNAPSHOT_INPUT] = feature_snapshot - ctx = build_strategy_context_from_available_inputs( + result = evaluate_feature_snapshot_strategy( entrypoint=self.entrypoint, runtime_adapter=self.runtime_adapter, - as_of=evaluation_as_of, - available_inputs=evaluation_inputs, + runtime_settings=FeatureSnapshotRuntimeSettings( + feature_snapshot_path=self.runtime_settings.feature_snapshot_path, + feature_snapshot_manifest_path=self.runtime_settings.feature_snapshot_manifest_path, + strategy_config_path=self.runtime_settings.strategy_config_path, + strategy_config_source=self.runtime_settings.strategy_config_source, + dry_run_only=self.runtime_settings.dry_run_only, + ), runtime_config=runtime_config, - ) - decision = self.entrypoint.evaluate(ctx) - managed_symbols = self._extract_managed_symbols( - feature_snapshot, - safe_haven_symbol=safe_haven_symbol, + merged_runtime_config=self.merged_runtime_config, + available_inputs=available_inputs, + base_managed_symbols=self.managed_symbols, + include_strategy_display_name=True, + set_run_as_of=True, + snapshot_loader=load_feature_snapshot_guarded, ) return StrategyEvaluationResult( - decision=decision, - metadata={ - "strategy_profile": self.profile, - "strategy_display_name": self.display_name, - "strategy_config_path": runtime_config_path, - "strategy_config_source": runtime_config_source, - "feature_snapshot_path": feature_snapshot_path, - "dry_run_only": self.runtime_settings.dry_run_only, - "managed_symbols": managed_symbols, - "safe_haven_symbol": safe_haven_symbol, - "status_icon": self.runtime_adapter.status_icon, - **guard_metadata, - }, + decision=result.decision, + metadata=result.metadata, ) - def _required_feature_columns(self) -> tuple[str, ...] | frozenset[str]: - return self.runtime_adapter.required_feature_columns - - def _snapshot_date_columns(self) -> tuple[str, ...]: - return tuple(self.runtime_adapter.snapshot_date_columns) - - def _max_snapshot_month_lag(self) -> int: - return int(self.runtime_adapter.max_snapshot_month_lag) - - def _require_snapshot_manifest(self) -> bool: - return bool(self.runtime_adapter.require_snapshot_manifest) - - def _snapshot_contract_version(self) -> str | None: - return self.runtime_adapter.snapshot_contract_version - - def _extract_managed_symbols( - self, - feature_snapshot, - *, - safe_haven_symbol: str | None, - ) -> tuple[str, ...]: - extractor = self.runtime_adapter.managed_symbols_extractor - if callable(extractor): - return tuple( - extractor( - feature_snapshot, - benchmark_symbol=str(self.merged_runtime_config.get("benchmark_symbol") or "QQQ"), - safe_haven=safe_haven_symbol, - ) - ) - if safe_haven_symbol: - return (safe_haven_symbol,) - return self.managed_symbols - def load_runtime_parameters(self) -> dict[str, Any]: runtime_loader = self.runtime_adapter.runtime_parameter_loader if not callable(runtime_loader): diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index 00f9fcb..d9abd18 100644 --- a/tests/test_runtime_config_support.py +++ b/tests/test_runtime_config_support.py @@ -382,11 +382,13 @@ def test_print_strategy_switch_env_plan_for_russell(self): self.assertEqual(plan["profile_group"], "snapshot_backed") self.assertEqual(plan["input_mode"], "feature_snapshot") self.assertTrue(plan["requires_snapshot_artifacts"]) + self.assertFalse(plan["requires_snapshot_manifest_path"]) self.assertFalse(plan["requires_strategy_config_path"]) self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_PATH"], "") - self.assertEqual( - plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"], - "", + self.assertNotIn("LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH", plan["set_env"]) + self.assertIn( + "LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH", + plan["remove_if_present"], ) self.assertIn("LONGBRIDGE_STRATEGY_CONFIG_PATH", plan["remove_if_present"]) @@ -459,6 +461,32 @@ def test_print_strategy_switch_env_plan_for_dynamic_mega_leveraged_pullback_sg(s "dynamic_mega_leveraged_pullback_feature_snapshot_latest.csv", ) + def test_print_strategy_switch_env_plan_for_tech_uses_bundled_config_by_default(self): + result = subprocess.run( + [ + sys.executable, + str(SWITCH_PLAN_SCRIPT_PATH), + "--profile", + "tech_communication_pullback_enhancement", + "--account-region", + "hk", + "--json", + ], + check=True, + capture_output=True, + text=True, + ) + + plan = json.loads(result.stdout) + self.assertEqual(plan["canonical_profile"], "tech_communication_pullback_enhancement") + self.assertEqual(plan["config_source_policy"], "bundled_or_env") + self.assertTrue(plan["requires_strategy_config_path"]) + self.assertEqual(plan["runtime_execution_window_trading_days"], 1) + self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_PATH"], "") + self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"], "") + self.assertNotIn("LONGBRIDGE_STRATEGY_CONFIG_PATH", plan["set_env"]) + self.assertIn("LONGBRIDGE_STRATEGY_CONFIG_PATH", plan["remove_if_present"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py index 84841d0..5074eb9 100644 --- a/tests/test_strategy_runtime.py +++ b/tests/test_strategy_runtime.py @@ -8,6 +8,7 @@ StrategyDecision, StrategyManifest, StrategyRuntimeAdapter, + StrategyRuntimePolicy, ) from runtime_config_support import PlatformRuntimeSettings @@ -247,6 +248,7 @@ def test_feature_snapshot_runtime_loads_snapshot_into_context(self): require_snapshot_manifest=False, managed_symbols_extractor=lambda *_args, **_kwargs: ("AAPL", "MSFT", "BOXX"), portfolio_input_name="portfolio_snapshot", + runtime_policy=StrategyRuntimePolicy(runtime_execution_window_trading_days=1), ), runtime_settings=_build_runtime_settings( "tech_communication_pullback_enhancement", diff --git a/tests/test_sync_cloud_run_env_workflow.sh b/tests/test_sync_cloud_run_env_workflow.sh index 5826cb2..ebfd16a 100644 --- a/tests/test_sync_cloud_run_env_workflow.sh +++ b/tests/test_sync_cloud_run_env_workflow.sh @@ -10,18 +10,17 @@ grep -Fq 'permissions:' "$workflow_file" grep -Fq 'id-token: write' "$workflow_file" grep -Fq 'workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}' "$workflow_file" grep -Fq 'service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}' "$workflow_file" -grep -Fq 'uses: actions/checkout@v4' "$workflow_file" -grep -Fq 'uses: actions/setup-python@v5' "$workflow_file" +grep -Fq 'uses: actions/checkout@v6' "$workflow_file" +grep -Fq 'uses: actions/setup-python@v6' "$workflow_file" grep -Fq 'python -m pip install -r requirements.txt' "$workflow_file" grep -Fq 'id: strategy_requirements' "$workflow_file" grep -Fq 'scripts/print_strategy_profile_status.py' "$workflow_file" -grep -Fq 'from quant_platform_kit.common.strategies import derive_strategy_artifact_paths' "$workflow_file" -grep -Fq 'from us_equity_strategies import get_strategy_catalog, resolve_canonical_profile' "$workflow_file" +grep -Fq 'from us_equity_strategies import resolve_canonical_profile' "$workflow_file" grep -Fq 'canonical_profile = resolve_canonical_profile(profile)' "$workflow_file" grep -Fq 'requires_snapshot_artifacts=' "$workflow_file" grep -Fq 'requires_snapshot_manifest_path=' "$workflow_file" grep -Fq 'requires_strategy_config_path=' "$workflow_file" -grep -Fq 'has_bundled_strategy_config=' "$workflow_file" +grep -Fq 'config_source_policy=' "$workflow_file" grep -Fq 'Wait for Cloud Run deployment of current commit' "$workflow_file" grep -Fq 'target_sha="${GITHUB_SHA}"' "$workflow_file" grep -Fq "gcloud run services describe \"\${CLOUD_RUN_SERVICE}\" --region \"\${CLOUD_RUN_REGION}\" --format='value(spec.template.metadata.labels.commit-sha)'" "$workflow_file" @@ -63,10 +62,11 @@ grep -Fq 'missing_vars+=("LONGBRIDGE_STRATEGY_CONFIG_PATH")' "$workflow_file" grep -Fq 'REQUIRES_SNAPSHOT_ARTIFACTS: ${{ steps.strategy_requirements.outputs.requires_snapshot_artifacts }}' "$workflow_file" grep -Fq 'REQUIRES_SNAPSHOT_MANIFEST_PATH: ${{ steps.strategy_requirements.outputs.requires_snapshot_manifest_path }}' "$workflow_file" grep -Fq 'REQUIRES_STRATEGY_CONFIG_PATH: ${{ steps.strategy_requirements.outputs.requires_strategy_config_path }}' "$workflow_file" -grep -Fq 'HAS_BUNDLED_STRATEGY_CONFIG: ${{ steps.strategy_requirements.outputs.has_bundled_strategy_config }}' "$workflow_file" +grep -Fq 'CONFIG_SOURCE_POLICY: ${{ steps.strategy_requirements.outputs.config_source_policy }}' "$workflow_file" grep -Fq 'if [ "${REQUIRES_SNAPSHOT_ARTIFACTS:-}" = "true" ] && [ -z "${LONGBRIDGE_FEATURE_SNAPSHOT_PATH:-}" ]; then' "$workflow_file" grep -Fq 'if [ "${REQUIRES_SNAPSHOT_MANIFEST_PATH:-}" = "true" ] && [ -z "${LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH:-}" ]; then' "$workflow_file" -grep -Fq '&& [ "${HAS_BUNDLED_STRATEGY_CONFIG:-}" != "true" ]; then' "$workflow_file" +grep -Fq '&& [ "${CONFIG_SOURCE_POLICY:-}" = "env_only" ] \' "$workflow_file" +grep -Fq '&& [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ]; then' "$workflow_file" grep -Fq 'secret_pairs+=("TELEGRAM_TOKEN=${TELEGRAM_TOKEN_SECRET_NAME}:latest")' "$workflow_file" grep -Fq 'secret_pairs+=("LONGPORT_APP_KEY=${LONGPORT_APP_KEY_SECRET_NAME}:latest")' "$workflow_file" grep -Fq 'secret_pairs+=("LONGPORT_APP_SECRET=${LONGPORT_APP_SECRET_SECRET_NAME}:latest")' "$workflow_file"