add controlled option overlay switch (#74)

Author: Pigbibi

.github/workflows/manual-strategy-switch.yml

@@ -90,6 +90,15 @@ on:
         description: "Optional income layer maximum allocation ratio."
         required: false
         type: string
+      option_overlay_mode:
+        description: "Option overlay policy: current preserves variables, enabled applies profile defaults, disabled clears it."
+        required: true
+        type: choice
+        default: current
+        options:
+          - current
+          - enabled
+          - disabled
       service_targets_mode:
         description: "auto patches IBKR CLOUD_RUN_SERVICE_TARGETS_JSON when it exists."
         required: true
@@ -149,6 +158,7 @@ jobs:
       MIN_RESERVED_CASH_USD: ${{ inputs.min_reserved_cash_usd }}
       INCOME_LAYER_START_USD: ${{ inputs.income_layer_start_usd }}
       INCOME_LAYER_MAX_RATIO: ${{ inputs.income_layer_max_ratio }}
+      OPTION_OVERLAY_MODE: ${{ inputs.option_overlay_mode }}
       SERVICE_TARGETS_MODE: ${{ inputs.service_targets_mode }}
       APPLY_SWITCH: ${{ inputs.apply }}
       TRIGGER_PLATFORM_SYNC: ${{ inputs.trigger_platform_sync }}
@@ -249,6 +259,7 @@ jobs:
             --strategy-profile "${STRATEGY_PROFILE}"
             --execution-mode "${EXECUTION_MODE}"
             --plugin-mode "${PLUGIN_MODE}"
+            --option-overlay-mode "${OPTION_OVERLAY_MODE}"
             --output "${target_file}"
           )
           if [ "${VARIABLE_SCOPE}" != "default" ]; then
@@ -374,6 +385,7 @@ jobs:
               "execution_mode": runtime_target["execution_mode"],
               "variable_scope": github["variable_scope"],
               "plugin_mode": os.environ["PLUGIN_MODE"],
+              "option_overlay_mode": os.environ.get("OPTION_OVERLAY_MODE", "current"),
               "deployment_selector": runtime_target["deployment_selector"],
               "account_selector": ",".join(runtime_target["account_selector"]),
               "account_scope": runtime_target["account_scope"],

scripts/build_runtime_switch.py

@@ -13,6 +13,7 @@
 SCRIPT_DIR = Path(__file__).resolve().parent
 if str(SCRIPT_DIR) not in sys.path:
     sys.path.insert(0, str(SCRIPT_DIR))
+ROOT = SCRIPT_DIR.parent
 
 from runtime_settings import (  # noqa: E402
     SUPPORTED_PLATFORMS,
@@ -118,6 +119,8 @@
     "option_income_overlay_nav_risk_ratio",
 )
 OPTION_OVERLAY_VARIABLES = tuple(field.upper() for field in OPTION_OVERLAY_CONTROL_FIELDS)
+OPTION_OVERLAY_MODES = frozenset({"current", "enabled", "disabled"})
+OPTION_OVERLAY_PROFILE_PATH = ROOT / "web" / "strategy-switch-console" / "strategy-profiles.example.json"
 RUNTIME_TARGET_VARIABLES = (
     "RUNTIME_TARGET_ENABLED",
 )
@@ -285,6 +288,16 @@ def _normalize_positive_decimal(value: str, *, field_name: str) -> str:
     return text
 
 
+def _normalize_nonnegative_decimal(value: str, *, field_name: str) -> str:
+    text = str(value or "").strip()
+    if not text or not re.fullmatch(r"(?:\d+|\d*\.\d+)", text):
+        raise ValueError(f"{field_name} must be a non-negative decimal number")
+    numeric = float(text)
+    if numeric < 0:
+        raise ValueError(f"{field_name} must be non-negative")
+    return text
+
+
 def _normalize_ratio_decimal(value: str, *, field_name: str) -> str:
     text = str(value or "").strip()
     if not text or not re.fullmatch(r"(?:\d+|\d*\.\d+)", text):
@@ -296,14 +309,28 @@ def _normalize_ratio_decimal(value: str, *, field_name: str) -> str:
 
 
 def _normalize_optional_bool_text(value: str, *, field_name: str) -> str:
-    text = str(value or "").strip().lower()
+    text = str(value if value is not None else "").strip().lower()
     if text in {"1", "true", "yes", "y", "on"}:
         return "true"
     if text in {"0", "false", "no", "n", "off"}:
         return "false"
     raise ValueError(f"{field_name} must be true or false")
 
 
+def _normalize_option_overlay_mode(value: str) -> str:
+    mode = str(value or "current").strip().lower()
+    if mode not in OPTION_OVERLAY_MODES:
+        raise ValueError("option_overlay_mode must be current, enabled, or disabled")
+    return mode
+
+
+def _normalize_option_recipe(value: str, *, field_name: str) -> str:
+    text = str(value or "").strip()
+    if not text or not re.fullmatch(r"[A-Za-z0-9._=-]{1,120}", text):
+        raise ValueError(f"{field_name} must be a recipe slug")
+    return text
+
+
 def _normalize_ibit_zscore_exit_mode(value: str) -> str:
     mode = str(value or "").strip().lower()
     aliases = {
@@ -346,6 +373,95 @@ def _extract_ibit_zscore_exit_control_fields(extra_variables: dict[str, Any]) ->
     return controls
 
 
+def _disabled_option_overlay_extra_variables() -> dict[str, str]:
+    values = {variable: "" for variable in OPTION_OVERLAY_VARIABLES}
+    values["OPTION_OVERLAY_ENABLED"] = "false"
+    values["OPTION_GROWTH_OVERLAY_ENABLED"] = "false"
+    values["OPTION_INCOME_OVERLAY_ENABLED"] = "false"
+    return values
+
+
+def _profile_bool(item: dict[str, Any], field_name: str, *, default: bool = False) -> bool:
+    if item.get(field_name) is None or str(item.get(field_name)).strip() == "":
+        return default
+    return _normalize_optional_bool_text(item[field_name], field_name=field_name) == "true"
+
+
+def _option_family_defaults(item: dict[str, Any], family: str) -> dict[str, str]:
+    control_prefix = f"option_{family}_overlay"
+    env_prefix = f"OPTION_{family.upper()}_OVERLAY"
+    enabled = _profile_bool(item, f"{control_prefix}_enabled", default=False)
+    values = {
+        f"{env_prefix}_ENABLED": "true" if enabled else "false",
+        f"{env_prefix}_RECIPE": "",
+        f"{env_prefix}_START_USD": "",
+    }
+    if family == "growth":
+        ratio_field = "option_growth_overlay_nav_budget_ratio"
+        ratio_variable = "OPTION_GROWTH_OVERLAY_NAV_BUDGET_RATIO"
+    else:
+        ratio_field = "option_income_overlay_nav_risk_ratio"
+        ratio_variable = "OPTION_INCOME_OVERLAY_NAV_RISK_RATIO"
+    values[ratio_variable] = ""
+    if not enabled:
+        return values
+
+    values[f"{env_prefix}_RECIPE"] = _normalize_option_recipe(
+        item.get(f"{control_prefix}_recipe"),
+        field_name=f"{control_prefix}_recipe",
+    )
+    values[f"{env_prefix}_START_USD"] = _normalize_nonnegative_decimal(
+        item.get(f"{control_prefix}_start_usd"),
+        field_name=f"{control_prefix}_start_usd",
+    )
+    values[ratio_variable] = _normalize_ratio_decimal(item.get(ratio_field), field_name=ratio_field)
+    return values
+
+
+def _load_option_overlay_profile_defaults() -> dict[str, dict[str, str]]:
+    try:
+        payload = json.loads(OPTION_OVERLAY_PROFILE_PATH.read_text(encoding="utf-8"))
+    except OSError as exc:
+        raise ValueError(f"cannot read {OPTION_OVERLAY_PROFILE_PATH}") from exc
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"{OPTION_OVERLAY_PROFILE_PATH} must be valid JSON") from exc
+    if not isinstance(payload, list):
+        raise ValueError(f"{OPTION_OVERLAY_PROFILE_PATH} must contain a strategy profile list")
+
+    defaults: dict[str, dict[str, str]] = {}
+    for item in payload:
+        if not isinstance(item, dict):
+            continue
+        profile = str(item.get("profile") or item.get("strategy_profile") or "").strip().lower()
+        if not profile:
+            continue
+        if not _profile_bool(item, "option_overlay_enabled", default=False):
+            continue
+        values = _disabled_option_overlay_extra_variables()
+        values["OPTION_OVERLAY_ENABLED"] = "true"
+        values.update(_option_family_defaults(item, "growth"))
+        values.update(_option_family_defaults(item, "income"))
+        if values["OPTION_GROWTH_OVERLAY_ENABLED"] != "true" and values["OPTION_INCOME_OVERLAY_ENABLED"] != "true":
+            raise ValueError(f"{profile} option overlay is enabled without a growth or income family")
+        defaults[profile] = values
+    return defaults
+
+
+def _option_overlay_extra_variables(args: argparse.Namespace, strategy_profile: str) -> dict[str, str]:
+    mode = _normalize_option_overlay_mode(getattr(args, "option_overlay_mode", "current"))
+    if mode == "current":
+        return {}
+    if mode == "disabled":
+        return _disabled_option_overlay_extra_variables()
+
+    defaults = _load_option_overlay_profile_defaults().get(strategy_profile)
+    if not defaults:
+        raise ValueError(
+            "option_overlay_mode enabled is only supported for strategies with option overlay defaults"
+        )
+    return dict(defaults)
+
+
 def _dca_extra_variables(
     args: argparse.Namespace,
     strategy_profile: str,
@@ -670,6 +786,7 @@ def _preserve_reserved_cash_fields(
         PLATFORM_MIN_RESERVED_CASH_VARIABLES.get(platform),
         PLATFORM_RESERVED_CASH_RATIO_VARIABLES.get(platform),
         *INCOME_LAYER_VARIABLES,
+        *OPTION_OVERLAY_VARIABLES,
         *RUNTIME_TARGET_VARIABLES,
         *DCA_RUNTIME_VARIABLES,
         *IBIT_ZSCORE_EXIT_RUNTIME_VARIABLES,
@@ -754,6 +871,7 @@ def build_switch_target(args: argparse.Namespace) -> dict[str, Any]:
         extra_variables["INCOME_LAYER_START_USD"] = args.income_layer_start_usd
     if args.income_layer_max_ratio:
         extra_variables["INCOME_LAYER_MAX_RATIO"] = args.income_layer_max_ratio
+    extra_variables.update(_option_overlay_extra_variables(args, runtime_target["strategy_profile"]))
     extra_variables.update(_dca_extra_variables(args, runtime_target["strategy_profile"], dca_controls))
     extra_variables.update(
         _ibit_zscore_exit_extra_variables(
@@ -826,6 +944,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--min-reserved-cash-usd", default="")
     parser.add_argument("--income-layer-start-usd", default="")
     parser.add_argument("--income-layer-max-ratio", default="")
+    parser.add_argument("--option-overlay-mode", choices=sorted(OPTION_OVERLAY_MODES), default="current")
     parser.add_argument("--dca-mode", default="")
     parser.add_argument("--dca-base-investment-usd", default="")
     parser.add_argument("--ibit-zscore-exit-mode", choices=("disabled", "paper", "live"), default="")

scripts/runtime_settings.py

@@ -6,6 +6,7 @@
 import argparse
 import json
 import os
+import re
 import shlex
 import subprocess
 import sys
@@ -68,7 +69,32 @@
         "OPTION_INCOME_OVERLAY_NAV_RISK_RATIO",
     }
 )
-RESEARCH_ONLY_EXTRA_VARIABLES = LEGACY_INCOME_LAYER_VARIABLES | OPTION_OVERLAY_VARIABLES
+OPTION_OVERLAY_ENABLED_VARIABLES = frozenset(
+    {
+        "OPTION_OVERLAY_ENABLED",
+        "OPTION_GROWTH_OVERLAY_ENABLED",
+        "OPTION_INCOME_OVERLAY_ENABLED",
+    }
+)
+OPTION_OVERLAY_RECIPE_VARIABLES = frozenset(
+    {
+        "OPTION_GROWTH_OVERLAY_RECIPE",
+        "OPTION_INCOME_OVERLAY_RECIPE",
+    }
+)
+OPTION_OVERLAY_AMOUNT_VARIABLES = frozenset(
+    {
+        "OPTION_GROWTH_OVERLAY_START_USD",
+        "OPTION_INCOME_OVERLAY_START_USD",
+    }
+)
+OPTION_OVERLAY_RATIO_VARIABLES = frozenset(
+    {
+        "OPTION_GROWTH_OVERLAY_NAV_BUDGET_RATIO",
+        "OPTION_INCOME_OVERLAY_NAV_RISK_RATIO",
+    }
+)
+RESEARCH_ONLY_EXTRA_VARIABLES = LEGACY_INCOME_LAYER_VARIABLES
 PLATFORM_DRY_RUN_VARIABLES = {
     "schwab": "SCHWAB_DRY_RUN_ONLY",
     "longbridge": "LONGBRIDGE_DRY_RUN_ONLY",
@@ -429,6 +455,72 @@ def validate_plugin_mounts(target: dict[str, Any], errors: list[str]) -> None:
             errors.append(f"{strategy_profile} requires an enabled {plugin} plugin mount")
 
 
+def option_bool_value(value: Any) -> bool | None:
+    text = str(value if value is not None else "").strip().lower()
+    if text in {"1", "true", "yes", "y", "on"}:
+        return True
+    if text in {"0", "false", "no", "n", "off"}:
+        return False
+    return None
+
+
+def validate_option_overlay_variables(extra_variables: dict[str, Any], errors: list[str]) -> None:
+    if not any(name in extra_variables for name in OPTION_OVERLAY_VARIABLES):
+        return
+
+    values = {
+        name: str(extra_variables.get(name) if extra_variables.get(name) is not None else "").strip()
+        for name in OPTION_OVERLAY_VARIABLES
+    }
+    for name in OPTION_OVERLAY_ENABLED_VARIABLES:
+        if values[name] and option_bool_value(values[name]) is None:
+            errors.append(f"extra_variables.{name} must be true or false")
+    for name in OPTION_OVERLAY_RECIPE_VARIABLES:
+        if values[name] and not re.fullmatch(r"[A-Za-z0-9._=-]{1,120}", values[name]):
+            errors.append(f"extra_variables.{name} must be a recipe slug")
+    for name in OPTION_OVERLAY_AMOUNT_VARIABLES:
+        if values[name] and not re.fullmatch(r"(?:\d+|\d*\.\d+)", values[name]):
+            errors.append(f"extra_variables.{name} must be a non-negative decimal")
+    for name in OPTION_OVERLAY_RATIO_VARIABLES:
+        if values[name]:
+            if not re.fullmatch(r"(?:\d+|\d*\.\d+)", values[name]):
+                errors.append(f"extra_variables.{name} must be a ratio between 0 and 1")
+                continue
+            numeric = float(values[name])
+            if numeric < 0 or numeric > 1:
+                errors.append(f"extra_variables.{name} must be a ratio between 0 and 1")
+
+    overlay_enabled = option_bool_value(values["OPTION_OVERLAY_ENABLED"]) if values["OPTION_OVERLAY_ENABLED"] else None
+    family_enabled: dict[str, bool | None] = {}
+    family_fields = {
+        "GROWTH": (
+            "OPTION_GROWTH_OVERLAY_ENABLED",
+            "OPTION_GROWTH_OVERLAY_RECIPE",
+            "OPTION_GROWTH_OVERLAY_START_USD",
+            "OPTION_GROWTH_OVERLAY_NAV_BUDGET_RATIO",
+        ),
+        "INCOME": (
+            "OPTION_INCOME_OVERLAY_ENABLED",
+            "OPTION_INCOME_OVERLAY_RECIPE",
+            "OPTION_INCOME_OVERLAY_START_USD",
+            "OPTION_INCOME_OVERLAY_NAV_RISK_RATIO",
+        ),
+    }
+    for family, (enabled_name, recipe_name, start_name, ratio_name) in family_fields.items():
+        enabled = option_bool_value(values[enabled_name]) if values[enabled_name] else None
+        family_enabled[family] = enabled
+        family_payload = [values[recipe_name], values[start_name], values[ratio_name]]
+        if enabled is True and not all(family_payload):
+            errors.append(f"extra_variables.{enabled_name} requires recipe, start_usd, and ratio fields")
+        if enabled is False and any(family_payload):
+            errors.append(f"extra_variables.{enabled_name} is false but {family.lower()} overlay fields are still set")
+
+    if overlay_enabled is True and not any(value is True for value in family_enabled.values()):
+        errors.append("extra_variables.OPTION_OVERLAY_ENABLED is true but no option overlay family is enabled")
+    if overlay_enabled is False and any(value is True for value in family_enabled.values()):
+        errors.append("extra_variables.OPTION_OVERLAY_ENABLED is false but an option overlay family is enabled")
+
+
 def validate_extra_variables(target: dict[str, Any], errors: list[str]) -> None:
     extra_variables = target.get("extra_variables", {})
     if not isinstance(extra_variables, dict):
@@ -452,6 +544,8 @@ def validate_extra_variables(target: dict[str, Any], errors: list[str]) -> None:
         if isinstance(value, str) and "\n" in value:
             errors.append(f"extra_variables.{name} must be a single-line value")
 
+    validate_option_overlay_variables(extra_variables, errors)
+
     runtime_target = target.get("runtime_target") if isinstance(target.get("runtime_target"), dict) else {}
     dry_run_only = runtime_target.get("dry_run_only")
     platform_id = runtime_target.get("platform_id")