Support income layer runtime controls (#176)

* Support income layer runtime controls

* Keep runtime controls compatible with test stubs

Author: Pigbibi

.github/workflows/sync-cloud-run-env.yml

@@ -156,6 +156,9 @@ jobs:
       # Optional strategy overrides; leave unset to inherit the UsEquityStrategies profile defaults.
       INCOME_THRESHOLD_USD: ${{ vars.INCOME_THRESHOLD_USD }}
       QQQI_INCOME_RATIO: ${{ vars.QQQI_INCOME_RATIO }}
+      INCOME_LAYER_ENABLED: ${{ vars.INCOME_LAYER_ENABLED }}
+      INCOME_LAYER_MAX_RATIO: ${{ vars.INCOME_LAYER_MAX_RATIO }}
+      RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }}
       NOTIFY_LANG: ${{ vars.NOTIFY_LANG }}
       EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI }}
       LONGBRIDGE_DRY_RUN_ONLY: ${{ vars.LONGBRIDGE_DRY_RUN_ONLY }}
@@ -933,6 +936,24 @@ jobs:
             remove_env_vars+=("QQQI_INCOME_RATIO")
           fi
 
+          if [ -n "${INCOME_LAYER_ENABLED:-}" ]; then
+            env_pairs+=("INCOME_LAYER_ENABLED=${INCOME_LAYER_ENABLED}")
+          else
+            remove_env_vars+=("INCOME_LAYER_ENABLED")
+          fi
+
+          if [ -n "${INCOME_LAYER_MAX_RATIO:-}" ]; then
+            env_pairs+=("INCOME_LAYER_MAX_RATIO=${INCOME_LAYER_MAX_RATIO}")
+          else
+            remove_env_vars+=("INCOME_LAYER_MAX_RATIO")
+          fi
+
+          if [ -n "${RUNTIME_TARGET_ENABLED:-}" ]; then
+            env_pairs+=("RUNTIME_TARGET_ENABLED=${RUNTIME_TARGET_ENABLED}")
+          else
+            remove_env_vars+=("RUNTIME_TARGET_ENABLED")
+          fi
+
           gcloud_args=(
             run services update "${CLOUD_RUN_SERVICE}"
             --region "${CLOUD_RUN_REGION}"

main.py

@@ -428,6 +428,9 @@ def publish_strategy_plugin_alerts(signals, *, report=None):
 
 
 def run_strategy(*, force_run: bool = False, validation_only: bool = False, validation_label: str = "backfill"):
+    if not validation_only and not force_run and not getattr(RUNTIME_SETTINGS, "runtime_target_enabled", True):
+        print(f"[{datetime.now()}] Runtime target disabled; skip strategy execution.", flush=True)
+        return True
     composer = build_composer(dry_run_only_override=True if validation_only else None)
     reporting_adapters = composer.build_reporting_adapters()
     log_context, report = reporting_adapters.start_run()

runtime_config_support.py

@@ -86,6 +86,7 @@ class PlatformRuntimeSettings:
     tg_token: str | None
     tg_chat_id: str | None
     dry_run_only: bool
+    runtime_target_enabled: bool = True
     market: str = DEFAULT_MARKET
     market_calendar: str = DEFAULT_MARKET_CALENDAR
     market_timezone: str = DEFAULT_MARKET_TIMEZONE
@@ -98,6 +99,8 @@ class PlatformRuntimeSettings:
     debug_position_snapshot: bool = False
     income_threshold_usd: float | None = None
     qqqi_income_ratio: float | None = None
+    income_layer_enabled: bool | None = None
+    income_layer_max_ratio: float | None = None
     runtime_execution_window_trading_days: int | None = None
     feature_snapshot_path: str | None = None
     feature_snapshot_manifest_path: str | None = None
@@ -263,6 +266,7 @@ def load_platform_runtime_settings(
         tg_token=os.getenv("TELEGRAM_TOKEN"),
         tg_chat_id=os.getenv("GLOBAL_TELEGRAM_CHAT_ID"),
         dry_run_only=resolve_bool_value(os.getenv("LONGBRIDGE_DRY_RUN_ONLY")),
+        runtime_target_enabled=_runtime_target_enabled_env(),
         reserved_cash_floor_usd=_resolve_non_negative_float_env(
             "LONGBRIDGE_MIN_RESERVED_CASH_USD",
             default=DEFAULT_RESERVED_CASH_FLOOR_USD,
@@ -283,6 +287,8 @@ def load_platform_runtime_settings(
         debug_position_snapshot=resolve_bool_value(os.getenv("LONGBRIDGE_DEBUG_POSITION_SNAPSHOT")),
         income_threshold_usd=resolve_optional_float_env(os.environ, "INCOME_THRESHOLD_USD"),
         qqqi_income_ratio=_qqqi_income_ratio_env(),
+        income_layer_enabled=_optional_bool_env("INCOME_LAYER_ENABLED"),
+        income_layer_max_ratio=_optional_ratio_env("INCOME_LAYER_MAX_RATIO"),
         runtime_execution_window_trading_days=_runtime_execution_window_trading_days_env(
             strategy_definition.profile
         ),
@@ -366,6 +372,34 @@ def _qqqi_income_ratio_env() -> float | None:
     return value
 
 
+def _optional_bool_env(name: str) -> bool | None:
+    raw_value = os.getenv(name)
+    if raw_value is None or str(raw_value).strip() == "":
+        return None
+    value = str(raw_value).strip().lower()
+    if value in {"1", "true", "yes", "y", "on"}:
+        return True
+    if value in {"0", "false", "no", "n", "off"}:
+        return False
+    raise ValueError(f"{name} must be boolean, got {raw_value!r}")
+
+
+def _runtime_target_enabled_env() -> bool:
+    value = _optional_bool_env("RUNTIME_TARGET_ENABLED")
+    return True if value is None else value
+
+
+def _optional_ratio_env(name: str) -> float | None:
+    value = resolve_optional_float_env(os.environ, name)
+    if value is None:
+        return None
+    if not math.isfinite(value):
+        raise ValueError(f"{name} must be finite, got {value}")
+    if not (0.0 <= value <= 1.0):
+        raise ValueError(f"{name} must be in [0,1], got {value}")
+    return value
+
+
 def _resolve_non_negative_float_env(name: str, *, default: float) -> float:
     value = resolve_optional_float_env(os.environ, name)
     if value is None:

strategy_runtime.py

@@ -166,6 +166,12 @@ def _default_runtime_settings(profile: str, display_name: str) -> PlatformRuntim
 
 def _build_runtime_overrides(profile: str, runtime_settings: PlatformRuntimeSettings) -> dict[str, Any]:
     overrides: dict[str, Any] = {}
+    income_layer_enabled = getattr(runtime_settings, "income_layer_enabled", None)
+    income_layer_max_ratio = getattr(runtime_settings, "income_layer_max_ratio", None)
+    if income_layer_enabled is not None:
+        overrides["income_layer_enabled"] = income_layer_enabled
+    if income_layer_max_ratio is not None:
+        overrides["income_layer_max_ratio"] = income_layer_max_ratio
     if profile == "tqqq_growth_income":
         if runtime_settings.income_threshold_usd is not None:
             overrides["income_threshold_usd"] = runtime_settings.income_threshold_usd

tests/test_runtime_config_support.py

@@ -149,6 +149,7 @@ def test_load_platform_runtime_settings_uses_defaults_with_explicit_strategy_pro
         self.assertIsNotNone(settings.runtime_target)
         self.assertEqual(settings.runtime_target.platform_id, "longbridge")
         self.assertEqual(settings.runtime_target.execution_mode, "live")
+        self.assertTrue(settings.runtime_target_enabled)
         self.assertIsNone(settings.income_threshold_usd)
         self.assertIsNone(settings.qqqi_income_ratio)
         self.assertIsNone(settings.feature_snapshot_path)
@@ -247,6 +248,31 @@ def test_dry_run_only_is_loaded_from_env(self):
 
         self.assertTrue(settings.dry_run_only)
 
+    def test_runtime_target_enabled_is_loaded_from_env(self):
+        with patch.dict(
+            os.environ,
+            {
+                "RUNTIME_TARGET_JSON": runtime_target_json(SAMPLE_STRATEGY_PROFILE),
+                "RUNTIME_TARGET_ENABLED": "false",
+            },
+            clear=True,
+        ):
+            settings = load_platform_runtime_settings(project_id_resolver=lambda: "project-1")
+
+        self.assertFalse(settings.runtime_target_enabled)
+
+    def test_invalid_runtime_target_enabled_is_rejected(self):
+        with patch.dict(
+            os.environ,
+            {
+                "RUNTIME_TARGET_JSON": runtime_target_json(SAMPLE_STRATEGY_PROFILE),
+                "RUNTIME_TARGET_ENABLED": "maybe",
+            },
+            clear=True,
+        ):
+            with self.assertRaisesRegex(ValueError, "RUNTIME_TARGET_ENABLED"):
+                load_platform_runtime_settings(project_id_resolver=lambda: "project-1")
+
     def test_debug_position_snapshot_is_loaded_from_env(self):
         with patch.dict(
             os.environ,
@@ -455,6 +481,8 @@ def test_income_layer_overrides_are_loaded_from_env(self):
                 "RUNTIME_TARGET_JSON": runtime_target_json("tqqq_growth_income"),
                 "INCOME_THRESHOLD_USD": "100000",
                 "QQQI_INCOME_RATIO": "0.5",
+                "INCOME_LAYER_ENABLED": "false",
+                "INCOME_LAYER_MAX_RATIO": "0.25",
             },
             clear=True,
         ):
@@ -463,6 +491,8 @@ def test_income_layer_overrides_are_loaded_from_env(self):
         self.assertEqual(settings.strategy_profile, "tqqq_growth_income")
         self.assertEqual(settings.income_threshold_usd, 100000.0)
         self.assertEqual(settings.qqqi_income_ratio, 0.5)
+        self.assertFalse(settings.income_layer_enabled)
+        self.assertEqual(settings.income_layer_max_ratio, 0.25)
 
     def test_tech_runtime_execution_window_override_rejects_research_only_profile(self):
         with patch.dict(
@@ -490,6 +520,18 @@ def test_rejects_invalid_qqqi_income_ratio(self):
             with self.assertRaisesRegex(ValueError, "QQQI_INCOME_RATIO"):
                 load_platform_runtime_settings(project_id_resolver=lambda: "project-1")
 
+    def test_rejects_invalid_income_layer_max_ratio(self):
+        with patch.dict(
+            os.environ,
+            {
+                "RUNTIME_TARGET_JSON": runtime_target_json("tqqq_growth_income"),
+                "INCOME_LAYER_MAX_RATIO": "1.5",
+            },
+            clear=True,
+        ):
+            with self.assertRaisesRegex(ValueError, "INCOME_LAYER_MAX_RATIO"):
+                load_platform_runtime_settings(project_id_resolver=lambda: "project-1")
+
     def test_rejects_human_readable_alias(self):
         with patch.dict(
             os.environ,

tests/test_strategy_runtime.py

@@ -100,6 +100,8 @@ def _build_runtime_settings(
     feature_snapshot_path: str | None = None,
     income_threshold_usd: float | None = None,
     qqqi_income_ratio: float | None = None,
+    income_layer_enabled: bool | None = None,
+    income_layer_max_ratio: float | None = None,
     runtime_execution_window_trading_days: int | None = None,
 ) -> PlatformRuntimeSettings:
     return PlatformRuntimeSettings(
@@ -118,6 +120,8 @@ def _build_runtime_settings(
         dry_run_only=False,
         income_threshold_usd=income_threshold_usd,
         qqqi_income_ratio=qqqi_income_ratio,
+        income_layer_enabled=income_layer_enabled,
+        income_layer_max_ratio=income_layer_max_ratio,
         runtime_execution_window_trading_days=runtime_execution_window_trading_days,
         feature_snapshot_path=feature_snapshot_path,
         feature_snapshot_manifest_path=None,
@@ -257,13 +261,19 @@ def test_load_strategy_runtime_applies_tqqq_income_overrides_from_settings(self)
                         "tqqq_growth_income",
                         income_threshold_usd=100000.0,
                         qqqi_income_ratio=0.5,
+                        income_layer_enabled=False,
+                        income_layer_max_ratio=0.25,
                     ),
                 )
 
         self.assertEqual(runtime.runtime_overrides["income_threshold_usd"], 100000.0)
         self.assertEqual(runtime.runtime_overrides["qqqi_income_ratio"], 0.5)
+        self.assertFalse(runtime.runtime_overrides["income_layer_enabled"])
+        self.assertEqual(runtime.runtime_overrides["income_layer_max_ratio"], 0.25)
         self.assertEqual(runtime.merged_runtime_config["income_threshold_usd"], 100000.0)
         self.assertEqual(runtime.merged_runtime_config["qqqi_income_ratio"], 0.5)
+        self.assertFalse(runtime.merged_runtime_config["income_layer_enabled"])
+        self.assertEqual(runtime.merged_runtime_config["income_layer_max_ratio"], 0.25)
 
     def test_load_strategy_runtime_applies_tech_execution_window_overrides_from_settings(self):
         entrypoint = _TechEntrypoint()