runtime_config_support.py

from __future__ import annotations

import math
import os
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Callable, Mapping

if TYPE_CHECKING:
    pass

from quant_platform_kit.common.runtime_config import (
    resolve_bool_value,
    resolve_cash_only_execution_env,
    resolve_dry_run_env,
    resolve_optional_bool_env,
    resolve_optional_dca_mode_env,
    resolve_optional_float_env,
    resolve_optional_ibit_zscore_exit_mode_env,
    resolve_optional_positive_float_env,
    resolve_optional_ratio_env,
    resolve_optional_symbol_env,
    resolve_split_env_list,
    resolve_strategy_runtime_path_settings,
)
from quant_platform_kit.common.runtime_target import (
    RuntimeTarget,
    build_runtime_target,
    resolve_runtime_target_from_env,
)
from quant_platform_kit.common.live_continuity import runtime_target_permits_standard_execution
from strategy_registry import (
    FIRSTRADE_PLATFORM,
    resolve_strategy_definition,
    resolve_strategy_metadata,
)
from us_equity_strategies import get_strategy_catalog

DEFAULT_ACCOUNT_REGION = "US"
DEFAULT_RESERVED_CASH_FLOOR_USD = 0.0
DEFAULT_RESERVED_CASH_RATIO = 0.0
DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD = 1000.0
IBIT_SMART_DCA_PROFILE = "ibit_smart_dca"


FIRSTRADE_SECRET_PROJECT_ID = "firstradequant"
# Legacy per-platform bot; only used when no unified token is injected.
LEGACY_TELEGRAM_TOKEN_SECRET_NAME = "firstrade-telegram-token"


def _read_secret(secret_name: str) -> str | None:
    try:
        from quant_platform_kit.cloud import get_secret_store

        return get_secret_store().get_secret(secret_name, project_id=FIRSTRADE_SECRET_PROJECT_ID)
    except Exception:
        return None


def _get_credential(secret_name: str, env_var: str) -> str | None:
    val = os.environ.get(env_var)
    if val is not None:
        return val
    return _read_secret(secret_name)


def resolve_telegram_token(
    env: Mapping[str, str] | None = None,
    *,
    secret_reader: Callable[[str], str | None] | None = None,
) -> str | None:
    """Resolve the bot token: injected TELEGRAM_TOKEN, then TELEGRAM_TOKEN_SECRET_NAME, then legacy."""

    source = os.environ if env is None else env
    reader = _read_secret if secret_reader is None else secret_reader
    direct = str(source.get("TELEGRAM_TOKEN") or "").strip()
    if direct:
        return direct
    configured_name = str(source.get("TELEGRAM_TOKEN_SECRET_NAME") or "").strip()
    names = [configured_name] if configured_name else []
    if LEGACY_TELEGRAM_TOKEN_SECRET_NAME not in names:
        names.append(LEGACY_TELEGRAM_TOKEN_SECRET_NAME)
    for name in names:
        value = str(reader(name) or "").strip()
        if value:
            return value
    return None


@dataclass(frozen=True)
class PlatformRuntimeSettings:
    project_id: str | None
    account_prefix: str
    account_region: str
    strategy_profile: str
    strategy_display_name: str
    strategy_domain: str
    notify_lang: str
    tg_token: str | None
    tg_chat_id: str | None
    dry_run_only: bool
    live_trading_enabled: bool
    run_strategy_on_http: bool
    live_order_ack: bool
    max_order_notional_usd: float | None
    strategy_metadata: Any = None
    runtime_target_enabled: bool = False
    cash_only_execution: bool = True
    reserved_cash_floor_usd: float = DEFAULT_RESERVED_CASH_FLOOR_USD
    reserved_cash_ratio: float = DEFAULT_RESERVED_CASH_RATIO
    persist_strategy_runs: bool = False
    safe_haven_cash_substitute_threshold_usd: float = DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD
    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_start_usd: float | None = None
    income_layer_max_ratio: float | None = None
    dca_mode: str | None = None
    dca_base_investment_usd: float | None = None
    ibit_zscore_exit_enabled: bool | None = None
    ibit_zscore_exit_mode: str | None = None
    ibit_zscore_exit_parking_symbol: str | None = None
    ibit_zscore_exit_risk_reduced_exposure: float | None = None
    ibit_zscore_exit_risk_off_exposure: float | None = None
    ibit_zscore_exit_allow_outside_execution_window: bool | None = None
    runtime_execution_window_trading_days: int | None = None
    market_signal_handoff_index_uri: str | None = None
    market_signal_handoff_manifest_uri: str | None = None
    market_signal_consumption_audit_uri: str | None = None
    market_signal_cache_dir: str | None = None
    market_signal_required: bool = False
    market_signal_fallback_mode: str | None = None
    market_signal_max_stale_days: int | None = None
    feature_snapshot_path: str | None = None
    feature_snapshot_manifest_path: str | None = None
    feature_snapshot_fallback_mode: str | None = None
    feature_snapshot_fallback_cache_dir: str | None = None
    feature_snapshot_fallback_max_stale_days: int | None = None
    strategy_config_path: str | None = None
    strategy_config_source: str | None = None
    strategy_plugin_mounts_json: str | None = None
    strategy_plugin_alert_channels: tuple[str, ...] = ()
    strategy_plugin_alert_email_recipients: tuple[str, ...] = ()
    strategy_plugin_alert_email_sender_email: str | None = None
    strategy_plugin_alert_email_sender_password: str | None = None
    strategy_plugin_alert_email_smtp_host: str | None = None
    strategy_plugin_alert_email_smtp_port: str | None = None
    strategy_plugin_alert_email_smtp_security: str | None = None
    strategy_plugin_alert_sms_recipients: tuple[str, ...] = ()
    strategy_plugin_alert_sms_provider: str | None = None
    strategy_plugin_alert_sms_account_id: str | None = None
    strategy_plugin_alert_sms_auth_token: str | None = None
    strategy_plugin_alert_sms_sender: str | None = None
    strategy_plugin_alert_sms_messaging_service_id: str | None = None
    strategy_plugin_alert_sms_api_base_url: str | None = None
    strategy_plugin_alert_sms_body_max_chars: str | None = None
    strategy_plugin_alert_push_recipients: tuple[str, ...] = ()
    strategy_plugin_alert_push_provider: str | None = None
    strategy_plugin_alert_push_app_token: str | None = None
    strategy_plugin_alert_push_access_token: str | None = None
    strategy_plugin_alert_push_api_base_url: str | None = None
    strategy_plugin_alert_push_device: str | None = None
    strategy_plugin_alert_push_priority: str | None = None
    strategy_plugin_alert_push_tags: str | None = None
    strategy_plugin_alert_push_body_max_chars: str | None = None
    strategy_plugin_alert_telegram_chat_ids: tuple[str, ...] = ()
    strategy_plugin_alert_telegram_bot_token: str | None = None
    strategy_plugin_alert_telegram_api_base_url: str | None = None
    strategy_plugin_alert_telegram_parse_mode: str | None = None
    strategy_plugin_alert_telegram_disable_web_page_preview: str | None = None
    strategy_plugin_alert_telegram_body_max_chars: str | None = None
    runtime_target: RuntimeTarget | None = None


def infer_account_region(raw_value: str | None, *, account_prefix: str) -> str:
    for candidate in (raw_value, account_prefix):
        normalized = _normalize_region(candidate)
        if normalized is not None:
            return normalized
    return DEFAULT_ACCOUNT_REGION


def load_platform_runtime_settings(
    *,
    project_id_resolver: Callable[[], str | None],
) -> PlatformRuntimeSettings:
    # Unset or empty means dry-run; only an explicit false enables live orders.
    dry_run_only = resolve_dry_run_env(os.environ, "FIRSTRADE_DRY_RUN_ONLY")
    account_prefix = os.getenv("ACCOUNT_PREFIX", "FIRSTRADE")
    safe_haven_cash_substitute_threshold_usd = resolve_optional_float_env(
        os.environ,
        "FIRSTRADE_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD",
    )
    runtime_target = _resolve_runtime_target(dry_run_only=dry_run_only)
    strategy_definition = resolve_strategy_definition(
        runtime_target.strategy_profile,
        platform_id=FIRSTRADE_PLATFORM,
    )
    strategy_metadata = resolve_strategy_metadata(
        strategy_definition.profile,
        platform_id=FIRSTRADE_PLATFORM,
    )
    runtime_paths = resolve_strategy_runtime_path_settings(
        strategy_catalog=get_strategy_catalog(),
        strategy_definition=strategy_definition,
        strategy_metadata=strategy_metadata,
        platform_env_prefix="FIRSTRADE",
        env=os.environ,
        repo_root=Path(__file__).resolve().parent,
    )
    return PlatformRuntimeSettings(
        project_id=project_id_resolver(),
        account_prefix=account_prefix,
        account_region=infer_account_region(
            os.getenv("ACCOUNT_REGION"),
            account_prefix=account_prefix,
        ),
        strategy_profile=runtime_paths.strategy_profile,
        strategy_display_name=runtime_paths.strategy_display_name,
        strategy_domain=runtime_paths.strategy_domain,
        strategy_metadata=strategy_metadata,
        notify_lang=os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG", "en"),
        tg_token=resolve_telegram_token(),
        tg_chat_id=os.getenv("QSL_GLOBAL_TELEGRAM_CHAT_ID") or os.getenv("GLOBAL_TELEGRAM_CHAT_ID"),
        dry_run_only=dry_run_only,
        runtime_target_enabled=(
            _runtime_target_enabled_env()
            and runtime_target_permits_standard_execution(runtime_target)
        ),
        cash_only_execution=resolve_cash_only_execution_env(
            os.environ,
            platform_env_prefix="FIRSTRADE",
        ),
        live_trading_enabled=resolve_bool_value(os.getenv("FIRSTRADE_ENABLE_LIVE_TRADING")),
        run_strategy_on_http=resolve_bool_value(os.getenv("FIRSTRADE_RUN_STRATEGY_ON_HTTP")),
        live_order_ack=resolve_bool_value(os.getenv("FIRSTRADE_LIVE_ORDER_ACK")),
        persist_strategy_runs=resolve_bool_value(os.getenv("FIRSTRADE_PERSIST_STRATEGY_RUNS")),
        max_order_notional_usd=resolve_optional_float_env(
            os.environ,
            "FIRSTRADE_MAX_ORDER_NOTIONAL_USD",
        ),
        reserved_cash_floor_usd=_resolve_non_negative_float_env(
            "FIRSTRADE_MIN_RESERVED_CASH_USD",
            default=DEFAULT_RESERVED_CASH_FLOOR_USD,
        ),
        reserved_cash_ratio=_resolve_ratio_env(
            "FIRSTRADE_RESERVED_CASH_RATIO",
            default=DEFAULT_RESERVED_CASH_RATIO,
        ),
        safe_haven_cash_substitute_threshold_usd=(
            max(0.0, safe_haven_cash_substitute_threshold_usd)
            if safe_haven_cash_substitute_threshold_usd is not None
            else DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD
        ),
        debug_position_snapshot=resolve_bool_value(os.getenv("FIRSTRADE_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_start_usd=_optional_non_negative_float_env("INCOME_LAYER_START_USD"),
        income_layer_max_ratio=resolve_optional_ratio_env("INCOME_LAYER_MAX_RATIO"),
        dca_mode=resolve_optional_dca_mode_env("DCA_MODE"),
        dca_base_investment_usd=resolve_optional_positive_float_env("DCA_BASE_INVESTMENT_USD"),
        ibit_zscore_exit_enabled=_optional_bool_env("IBIT_ZSCORE_EXIT_ENABLED"),
        ibit_zscore_exit_mode=resolve_optional_ibit_zscore_exit_mode_env("IBIT_ZSCORE_EXIT_MODE"),
        ibit_zscore_exit_parking_symbol=resolve_optional_symbol_env("IBIT_ZSCORE_EXIT_PARKING_SYMBOL"),
        ibit_zscore_exit_risk_reduced_exposure=resolve_optional_ratio_env(
            "IBIT_ZSCORE_EXIT_RISK_REDUCED_EXPOSURE"
        ),
        ibit_zscore_exit_risk_off_exposure=resolve_optional_ratio_env("IBIT_ZSCORE_EXIT_RISK_OFF_EXPOSURE"),
        ibit_zscore_exit_allow_outside_execution_window=_optional_bool_env(
            "IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW"
        ),
        runtime_execution_window_trading_days=_runtime_execution_window_trading_days_env(),
        market_signal_handoff_index_uri=_resolve_market_signal_handoff_index_uri(),
        market_signal_handoff_manifest_uri=_first_non_empty(
            os.getenv("FIRSTRADE_MARKET_SIGNAL_HANDOFF_MANIFEST_URI"),
            os.getenv("MARKET_SIGNAL_HANDOFF_MANIFEST_URI"),
        ),
        market_signal_consumption_audit_uri=_first_non_empty(
            os.getenv("FIRSTRADE_MARKET_SIGNAL_CONSUMPTION_AUDIT_URI"),
            os.getenv("MARKET_SIGNAL_CONSUMPTION_AUDIT_URI"),
        ),
        market_signal_cache_dir=_first_non_empty(
            os.getenv("FIRSTRADE_MARKET_SIGNAL_CACHE_DIR"),
            os.getenv("MARKET_SIGNAL_CACHE_DIR"),
        ),
        market_signal_required=_resolve_market_signal_required(
            strategy_profile=strategy_definition.profile,
            dca_mode=resolve_optional_dca_mode_env("DCA_MODE"),
        ),
        market_signal_fallback_mode=_resolve_market_signal_fallback_mode(
            strategy_profile=strategy_definition.profile,
            dca_mode=resolve_optional_dca_mode_env("DCA_MODE"),
        ),
        market_signal_max_stale_days=_optional_int(
            _first_non_empty(
                os.getenv("FIRSTRADE_MARKET_SIGNAL_MAX_STALE_DAYS"),
                os.getenv("FIRSTRADE_MARKET_SIGNAL_FALLBACK_MAX_STALE_DAYS"),
                os.getenv("MARKET_SIGNAL_MAX_STALE_DAYS"),
                os.getenv("MARKET_SIGNAL_FALLBACK_MAX_STALE_DAYS"),
            )
        ),
        feature_snapshot_path=runtime_paths.feature_snapshot_path,
        feature_snapshot_manifest_path=runtime_paths.feature_snapshot_manifest_path,
        feature_snapshot_fallback_mode=_first_non_empty(
            os.getenv("FIRSTRADE_FEATURE_SNAPSHOT_FALLBACK_MODE"),
            os.getenv("FEATURE_SNAPSHOT_FALLBACK_MODE"),
        ),
        feature_snapshot_fallback_cache_dir=_first_non_empty(
            os.getenv("FIRSTRADE_FEATURE_SNAPSHOT_FALLBACK_CACHE_DIR"),
            os.getenv("FEATURE_SNAPSHOT_FALLBACK_CACHE_DIR"),
        ),
        feature_snapshot_fallback_max_stale_days=_optional_int(
            _first_non_empty(
                os.getenv("FIRSTRADE_FEATURE_SNAPSHOT_MAX_STALE_DAYS"),
                os.getenv("FIRSTRADE_FEATURE_SNAPSHOT_FALLBACK_MAX_STALE_DAYS"),
                os.getenv("FEATURE_SNAPSHOT_MAX_STALE_DAYS"),
                os.getenv("FEATURE_SNAPSHOT_FALLBACK_MAX_STALE_DAYS"),
            )
        ),
        strategy_config_path=runtime_paths.strategy_config_path,
        strategy_config_source=runtime_paths.strategy_config_source,
        strategy_plugin_mounts_json=(
            os.getenv("FIRSTRADE_STRATEGY_PLUGIN_MOUNTS_JSON")
            or os.getenv("QSL_STRATEGY_PLUGIN_MOUNTS_JSON")
            or os.getenv("STRATEGY_PLUGIN_MOUNTS_JSON")
        ),
        **_load_strategy_plugin_alert_settings(),
        runtime_target=runtime_target,
    )


def _load_strategy_plugin_alert_settings() -> dict[str, object]:
    return {
        'strategy_plugin_alert_channels': resolve_split_env_list("STRATEGY_PLUGIN_ALERT_CHANNELS"),
        'strategy_plugin_alert_email_recipients': resolve_split_env_list("STRATEGY_PLUGIN_ALERT_EMAIL_RECIPIENTS"),
        'strategy_plugin_alert_email_sender_email': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_EMAIL_SENDER_EMAIL")),
        'strategy_plugin_alert_email_sender_password': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_EMAIL_SENDER_PASSWORD")
        ),
        'strategy_plugin_alert_email_smtp_host': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_EMAIL_SMTP_HOST")),
        'strategy_plugin_alert_email_smtp_port': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_EMAIL_SMTP_PORT")),
        'strategy_plugin_alert_email_smtp_security': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_EMAIL_SMTP_SECURITY")
        ),
        'strategy_plugin_alert_sms_recipients': resolve_split_env_list("STRATEGY_PLUGIN_ALERT_SMS_RECIPIENTS"),
        'strategy_plugin_alert_sms_provider': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_SMS_PROVIDER")),
        'strategy_plugin_alert_sms_account_id': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_SMS_ACCOUNT_ID")),
        'strategy_plugin_alert_sms_auth_token': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_SMS_AUTH_TOKEN")),
        'strategy_plugin_alert_sms_sender': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_SMS_SENDER")),
        'strategy_plugin_alert_sms_messaging_service_id': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_SMS_MESSAGING_SERVICE_ID")
        ),
        'strategy_plugin_alert_sms_api_base_url': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_SMS_API_BASE_URL")),
        'strategy_plugin_alert_sms_body_max_chars': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_SMS_BODY_MAX_CHARS")
        ),
        'strategy_plugin_alert_push_recipients': resolve_split_env_list("STRATEGY_PLUGIN_ALERT_PUSH_RECIPIENTS"),
        'strategy_plugin_alert_push_provider': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_PROVIDER")),
        'strategy_plugin_alert_push_app_token': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_APP_TOKEN")),
        'strategy_plugin_alert_push_access_token': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_ACCESS_TOKEN")),
        'strategy_plugin_alert_push_api_base_url': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_API_BASE_URL")),
        'strategy_plugin_alert_push_device': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_DEVICE")),
        'strategy_plugin_alert_push_priority': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_PRIORITY")),
        'strategy_plugin_alert_push_tags': _first_non_empty(os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_TAGS")),
        'strategy_plugin_alert_push_body_max_chars': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_PUSH_BODY_MAX_CHARS")
        ),
        'strategy_plugin_alert_telegram_chat_ids': resolve_split_env_list(
            "STRATEGY_PLUGIN_ALERT_TELEGRAM_CHAT_IDS"
        ),
        'strategy_plugin_alert_telegram_bot_token': _first_non_empty(
            os.getenv("QSL_STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN")
            or os.getenv("STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN")
        ),
        'strategy_plugin_alert_telegram_api_base_url': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_TELEGRAM_API_BASE_URL")
        ),
        'strategy_plugin_alert_telegram_parse_mode': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_TELEGRAM_PARSE_MODE")
        ),
        'strategy_plugin_alert_telegram_disable_web_page_preview': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_TELEGRAM_DISABLE_WEB_PAGE_PREVIEW")
        ),
        'strategy_plugin_alert_telegram_body_max_chars': _first_non_empty(
            os.getenv("STRATEGY_PLUGIN_ALERT_TELEGRAM_BODY_MAX_CHARS")
        ),
    }


def _resolve_runtime_target(*, dry_run_only: bool) -> RuntimeTarget:
    if os.getenv("QSL_RUNTIME_TARGET_JSON") or os.getenv("RUNTIME_TARGET_JSON"):
        return resolve_runtime_target_from_env(
            env=os.environ,
            expected_platform_id=FIRSTRADE_PLATFORM,
        )
    profile = os.getenv("STRATEGY_PROFILE")
    if not profile:
        raise EnvironmentError("STRATEGY_PROFILE or RUNTIME_TARGET_JSON is required")
    return build_runtime_target(
        platform_id=FIRSTRADE_PLATFORM,
        strategy_profile=profile,
        dry_run_only=dry_run_only,
        deployment_selector=os.getenv("K_SERVICE"),
        account_selector=os.getenv("FIRSTRADE_ACCOUNT"),
        account_scope=os.getenv("ACCOUNT_REGION") or DEFAULT_ACCOUNT_REGION,
        service_name=os.getenv("K_SERVICE"),
    )


def _normalize_region(raw_value: str | None) -> str | None:
    if raw_value is None:
        return None
    value = str(raw_value).strip()
    if not value:
        return None
    return value.upper()


def _qqqi_income_ratio_env() -> float | None:
    value = resolve_optional_float_env(os.environ, "QQQI_INCOME_RATIO")
    if value is not None and not (0.0 <= value <= 1.0):
        raise ValueError(f"QQQI_INCOME_RATIO must be in [0,1], got {value}")
    return value


def _runtime_target_enabled_env() -> bool:
    value = _optional_bool_env("RUNTIME_TARGET_ENABLED")
    return False if value is None else value


def _optional_bool_env(name: str) -> bool | None:
    raw_value = os.getenv(f"QSL_{name}") or os.getenv(name)
    if raw_value is None or str(raw_value).strip() == "":
        return None
    return resolve_optional_bool_env(name)


def _optional_non_negative_float_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 value < 0:
        raise ValueError(f"{name} must be non-negative, got {value}")
    return float(value)


def _resolve_non_negative_float_env(name: str, *, default: float) -> float:
    value = resolve_optional_float_env(os.environ, name)
    if value is None:
        return float(default)
    if not math.isfinite(value):
        raise ValueError(f"{name} must be finite, got {value}")
    if value < 0:
        raise ValueError(f"{name} must be non-negative, got {value}")
    return float(value)


def _resolve_ratio_env(name: str, *, default: float) -> float:
    value = _resolve_non_negative_float_env(name, default=default)
    if value > 1.0:
        raise ValueError(f"{name} must be in [0,1], got {value}")
    return value


def _default_market_signal_handoff_index_uri_from_report_bucket(
    execution_report_gcs_uri: str | None,
) -> str | None:
    text = str(execution_report_gcs_uri or "").strip()
    if not text.startswith("gs://"):
        return None
    bucket = text[len("gs://") :].split("/", 1)[0].strip()
    if not bucket:
        return None
    return f"gs://{bucket}/platform_handoffs/index.json"


def _resolve_market_signal_handoff_index_uri() -> str | None:
    explicit = _first_non_empty(
        os.getenv("FIRSTRADE_MARKET_SIGNAL_HANDOFF_INDEX_URI"),
        os.getenv("MARKET_SIGNAL_HANDOFF_INDEX_URI"),
    )
    if explicit:
        return explicit
    return _default_market_signal_handoff_index_uri_from_report_bucket(
        os.getenv("QSL_EXECUTION_REPORT_GCS_URI") or os.getenv("EXECUTION_REPORT_GCS_URI"),
    )


def _resolve_market_signal_required(*, strategy_profile: str, dca_mode: str | None) -> bool:
    explicit = _first_non_empty(
        os.getenv("FIRSTRADE_MARKET_SIGNAL_REQUIRED"),
        os.getenv("MARKET_SIGNAL_REQUIRED"),
    )
    if explicit is not None:
        return resolve_bool_value(explicit)
    if strategy_profile == IBIT_SMART_DCA_PROFILE and str(dca_mode or "").strip().lower() == "smart":
        return True
    return False


def _resolve_market_signal_fallback_mode(
    *,
    strategy_profile: str,
    dca_mode: str | None,
) -> str | None:
    explicit = _first_non_empty(
        os.getenv("FIRSTRADE_MARKET_SIGNAL_FALLBACK_MODE"),
        os.getenv("MARKET_SIGNAL_FALLBACK_MODE"),
    )
    if explicit:
        return explicit
    if strategy_profile == IBIT_SMART_DCA_PROFILE and str(dca_mode or "").strip().lower() == "smart":
        return "last_valid"
    return None


def _first_non_empty(*raw_values: str | None) -> str | None:
    for raw_value in raw_values:
        value = str(raw_value or "").strip()
        if value:
            return value
    return None


def _optional_int(raw_value: str | None) -> int | None:
    value = str(raw_value or "").strip()
    if not value:
        return None
    return int(value)


def _runtime_execution_window_trading_days_env() -> int | None:
    env_name = "FIRSTRADE_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS"
    raw_value = os.getenv(env_name)
    if raw_value is None or not str(raw_value).strip():
        return None
    try:
        value = int(str(raw_value).strip())
    except ValueError as exc:
        raise ValueError(f"{env_name} must be a positive integer") from exc
    if value <= 0:
        raise ValueError(f"{env_name} must be a positive integer")
    return value