Add scheduler plans to runtime strategy switches

Author: Pigbibi

scripts/build_runtime_switch.py

@@ -32,6 +32,43 @@
         "mega_cap_leader_rotation_top50_balanced",
     }
 )
+US_DAILY_SCHEDULER = {
+    "timezone": "America/New_York",
+    "main_time": "45 15 * * *",
+    "probe_time": "35 9,15 * * *",
+    "precheck_time": "45 9 * * *",
+}
+US_DCA_SCHEDULER = {
+    "timezone": "America/New_York",
+    "main_time": "45 15 25-29 * *",
+    "probe_time": "35 9,15 25-29 * *",
+    "precheck_time": "45 9 25-29 * *",
+}
+US_SNAPSHOT_SCHEDULER = {
+    "timezone": "America/New_York",
+    "main_time": "45 15 1-7 * *",
+    "probe_time": "35 9,15 1-7 * *",
+    "precheck_time": "45 9 1-7 * *",
+}
+HK_DAILY_SCHEDULER = {
+    "timezone": "Asia/Hong_Kong",
+    "main_time": "45 15 * * *",
+    "probe_time": "35 9,15 * * *",
+    "precheck_time": "45 9 * * *",
+}
+HK_SNAPSHOT_SCHEDULER = {
+    "timezone": "Asia/Hong_Kong",
+    "main_time": "45 15 1-7 * *",
+    "probe_time": "35 9,15 1-7 * *",
+    "precheck_time": "45 9 1-7 * *",
+}
+STRATEGY_SCHEDULER_PROFILES = {
+    "nasdaq_sp500_smart_dca": US_DCA_SCHEDULER,
+    "ibit_smart_dca": US_DCA_SCHEDULER,
+    "russell_1000_multi_factor_defensive": US_SNAPSHOT_SCHEDULER,
+    "mega_cap_leader_rotation_top50_balanced": US_SNAPSHOT_SCHEDULER,
+    "hk_low_vol_dividend_quality_snapshot": HK_SNAPSHOT_SCHEDULER,
+}
 PLATFORM_DRY_RUN_VARIABLES = {
     "schwab": "SCHWAB_DRY_RUN_ONLY",
     "longbridge": "LONGBRIDGE_DRY_RUN_ONLY",
@@ -216,6 +253,14 @@ def _execution_mode_and_dry_run(raw_mode: str) -> tuple[str, bool]:
     raise ValueError("execution_mode must be live or paper")
 
 
+def _scheduler_plan_for_strategy(strategy_profile: str) -> dict[str, str]:
+    profile = str(strategy_profile or "").strip().lower()
+    scheduler = STRATEGY_SCHEDULER_PROFILES.get(profile)
+    if scheduler is None:
+        scheduler = HK_DAILY_SCHEDULER if profile.startswith("hk_") else US_DAILY_SCHEDULER
+    return dict(scheduler)
+
+
 def _build_runtime_target(args: argparse.Namespace) -> dict[str, Any]:
     platform = _normalize_platform(args.platform)
     target_name = _normalize_target_name(args.target_name)
@@ -232,15 +277,17 @@ def _build_runtime_target(args: argparse.Namespace) -> dict[str, Any]:
     )
     account_selector = _split_csv(args.account_selector) or _account_selector_default(platform, account_scope)
     service_name = args.service_name.strip() if args.service_name else _default_service_name(platform, target_name)
+    strategy_profile = args.strategy_profile.strip().lower()
     runtime_target: dict[str, Any] = {
         "platform_id": platform,
-        "strategy_profile": args.strategy_profile.strip().lower(),
+        "strategy_profile": strategy_profile,
         "dry_run_only": dry_run_only,
         "deployment_selector": deployment_selector,
         "account_selector": account_selector,
         "account_scope": account_scope,
         "service_name": service_name,
         "execution_mode": execution_mode,
+        "scheduler": _scheduler_plan_for_strategy(strategy_profile),
     }
     execution_windows = _load_json_object(args.execution_windows_json, field_name="execution_windows_json")
     if execution_windows:

scripts/runtime_settings.py

@@ -44,6 +44,7 @@
     "precheck": {"notify_only", "dry_run"},
     "execution": {"live", "paper", "dry_run"},
 }
+SCHEDULER_FIELDS = frozenset({"timezone", "main_time", "probe_time", "precheck_time"})
 GENERATED_VARIABLES = {"RUNTIME_TARGET_JSON", "STRATEGY_PROFILE"}
 SECRET_MARKERS = ("PASSWORD", "PRIVATE_KEY", "TOKEN", "API_KEY", "ACCESS_KEY", "CLIENT_SECRET", "SECRET")
 PLATFORM_DRY_RUN_VARIABLES = {
@@ -311,6 +312,24 @@ def validate_runtime_target(target: dict[str, Any], errors: list[str]) -> None:
                     )
                     break
 
+    scheduler = runtime_target.get("scheduler")
+    if scheduler is not None:
+        if not isinstance(scheduler, dict):
+            errors.append("runtime_target.scheduler must be an object when present")
+        else:
+            for field in scheduler:
+                if field not in SCHEDULER_FIELDS:
+                    errors.append(f"runtime_target.scheduler.{field} is unsupported")
+            timezone = scheduler.get("timezone")
+            if not isinstance(timezone, str) or not timezone.strip():
+                errors.append("runtime_target.scheduler.timezone must be a non-empty string")
+            for field in ("main_time", "probe_time", "precheck_time"):
+                value = scheduler.get(field)
+                if not isinstance(value, str) or len(value.split()) not in {2, 5}:
+                    errors.append(
+                        f"runtime_target.scheduler.{field} must have 2 time fields or 5 cron fields"
+                    )
+
 
 def validate_plugin_mounts(target: dict[str, Any], errors: list[str]) -> None:
     runtime_target = target.get("runtime_target") if isinstance(target.get("runtime_target"), dict) else {}

tests/test_runtime_settings.py

@@ -181,6 +181,15 @@ def test_build_switch_target_defaults_longbridge_sg_tqqq(self):
         self.assertEqual(target["github"]["environment"], "longbridge-sg")
         self.assertEqual(target["runtime_target"]["service_name"], "longbridge-quant-sg-service")
         self.assertEqual(target["runtime_target"]["account_scope"], "SG")
+        self.assertEqual(
+            target["runtime_target"]["scheduler"],
+            {
+                "timezone": "America/New_York",
+                "main_time": "45 15 * * *",
+                "probe_time": "35 9,15 * * *",
+                "precheck_time": "45 9 * * *",
+            },
+        )
         self.assertEqual(assignments["STRATEGY_PROFILE"], "tqqq_growth_income")
         self.assertEqual(assignments["LONGBRIDGE_DRY_RUN_ONLY"], "false")
         plugin_payload = json.loads(assignments["LONGBRIDGE_STRATEGY_PLUGIN_MOUNTS_JSON"])
@@ -285,6 +294,74 @@ def test_build_switch_target_defaults_firstrade_repository_scope(self):
         plugin_payload = json.loads(assignments["FIRSTRADE_STRATEGY_PLUGIN_MOUNTS_JSON"])
         self.assertEqual(plugin_payload["strategy_plugins"][0]["plugin"], "market_regime_control")
 
+    def test_build_switch_target_uses_dca_monthly_scheduler_window(self):
+        parser = build_runtime_switch.build_parser()
+        args = parser.parse_args(
+            [
+                "--platform",
+                "ibkr",
+                "--target-name",
+                "dca",
+                "--strategy-profile",
+                "nasdaq_sp500_smart_dca",
+                "--plugin-mode",
+                "none",
+            ]
+        )
+
+        target = build_runtime_switch.build_switch_target(args)
+
+        self.assertEqual(
+            target["runtime_target"]["scheduler"],
+            {
+                "timezone": "America/New_York",
+                "main_time": "45 15 25-29 * *",
+                "probe_time": "35 9,15 25-29 * *",
+                "precheck_time": "45 9 25-29 * *",
+            },
+        )
+
+    def test_build_switch_target_uses_snapshot_scheduler_window(self):
+        parser = build_runtime_switch.build_parser()
+        args = parser.parse_args(
+            [
+                "--platform",
+                "longbridge",
+                "--target-name",
+                "hk",
+                "--strategy-profile",
+                "hk_low_vol_dividend_quality_snapshot",
+                "--plugin-mode",
+                "none",
+            ]
+        )
+
+        target = build_runtime_switch.build_switch_target(args)
+
+        self.assertEqual(
+            target["runtime_target"]["scheduler"],
+            {
+                "timezone": "Asia/Hong_Kong",
+                "main_time": "45 15 1-7 * *",
+                "probe_time": "35 9,15 1-7 * *",
+                "precheck_time": "45 9 1-7 * *",
+            },
+        )
+
+    def test_runtime_target_scheduler_rejects_invalid_cron_shape(self):
+        _, target = self.load_target("examples/targets/schwab/live.example.json")
+        target["runtime_target"]["scheduler"] = {
+            "timezone": "America/New_York",
+            "main_time": "45",
+            "probe_time": "35 9,15 * * *",
+            "precheck_time": "45 9 * * *",
+        }
+
+        self.assertIn(
+            "runtime_target.scheduler.main_time must have 2 time fields or 5 cron fields",
+            runtime_settings.validate_target(target),
+        )
+
     def test_build_switch_target_rejects_secret_extra_variable(self):
         parser = build_runtime_switch.build_parser()
         args = parser.parse_args(