Skip to content

fix(ops): make scheduler policy target-driven #347

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Pigbibi wants to merge 3 commits into from
Closed

fix(ops): make scheduler policy target-driven #347

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7140456
fix(ops): make scheduler policy target-driven
Pigbibi Jul 15, 2026
b38fb7b
fix: validate scheduler attempt deadlines
Pigbibi Jul 16, 2026
eebac97
fix: derive live scheduler deadline from dry-run policy
Pigbibi Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 33 additions & 29 deletions scripts/build_cloud_run_env_sync_plan.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import argparse
import json
import os
import re
import sys
from collections.abc import Mapping, Sequence
from pathlib import Path
Expand Down Expand Up @@ -134,17 +135,9 @@ def _should_add_local_src(candidate: Path) -> bool:
"probe_time": "CLOUD_SCHEDULER_PROBE_TIME",
"precheck_time": "CLOUD_SCHEDULER_PRECHECK_TIME",
}
NEAR_RUN_WARMUP_SCHEDULE = "43 9,15 * * 1-5"
RUN_SCHEDULER_ATTEMPT_DEADLINE = "330s"
NEAR_RUN_WARMUP_SERVICES = frozenset(
{
"interactive-brokers-quant-live-u15998061-service",
"interactive-brokers-quant-live-u16608560-service",
"interactive-brokers-quant-live-u18336562-service",
"interactive-brokers-quant-live-u18308207-service",
}
)

DEFAULT_LIVE_RUN_ATTEMPT_DEADLINE = "330s"
CLOUD_RUN_REQUEST_TIMEOUT_SECONDS = 300
MAX_SCHEDULER_ATTEMPT_DEADLINE_SECONDS = 30 * 60
# Strategy-derived vars: auto-populated from platform-config.json defaults.
def _derive_strategy_env_defaults(strategy_config: dict) -> dict[str, str]:
"""Derive env var defaults from a strategy's platform-config.json entry."""
Expand Down Expand Up @@ -468,10 +461,6 @@ def _build_target_plan(
env_values=env_values,
per_service_mode=per_service_mode,
)
if service_name in NEAR_RUN_WARMUP_SERVICES:
scheduler["attempt_deadline"] = RUN_SCHEDULER_ATTEMPT_DEADLINE
_validate_near_run_warmup_schedule(service_name, scheduler)

return {
"service_name": service_name,
"strategy_profile": canonical_profile,
Expand Down Expand Up @@ -509,23 +498,38 @@ def _build_scheduler_plan(
allow_shared_fallback=True,
)
scheduler[key] = str(runtime_scheduler.get(key) or configured_value or SCHEDULER_TIME_DEFAULTS[key])
has_attempt_deadline = "attempt_deadline" in runtime_scheduler
attempt_deadline = runtime_scheduler.get("attempt_deadline")
execution_mode = str(runtime_target.get("execution_mode") or "").strip()
dry_run_only = runtime_target.get("dry_run_only")
is_live = execution_mode == "live" or (
not execution_mode and type(dry_run_only) is bool and not dry_run_only
Comment on lines +505 to +506

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Normalize live detection before defaulting deadlines

Fresh evidence: the pinned resolve_runtime_target_from_env accepts string booleans such as "false" for dry_run_only and compares execution_mode case-insensitively. For a valid live target that omits execution_mode but has "dry_run_only":"false" (or uses "LIVE"), this check treats the target as non-live, so the plan omits scheduler.attempt_deadline and the workflow skips --attempt-deadline for the /run scheduler even though the Cloud Run timeout is 300s.

Useful? React with 👍 / 👎.

)
if not has_attempt_deadline and is_live:
attempt_deadline = DEFAULT_LIVE_RUN_ATTEMPT_DEADLINE
if attempt_deadline is not None:
if type(attempt_deadline) is not str or not attempt_deadline.strip():
raise ValueError("scheduler.attempt_deadline must be a non-empty string")
normalized_deadline = attempt_deadline.strip()
match = re.fullmatch(r"([1-9][0-9]*)s", normalized_deadline)
if match is None:
raise ValueError("scheduler.attempt_deadline must be whole seconds, for example '330s'")
deadline_seconds = int(match.group(1))
if not (
CLOUD_RUN_REQUEST_TIMEOUT_SECONDS
< deadline_seconds
<= MAX_SCHEDULER_ATTEMPT_DEADLINE_SECONDS
):
raise ValueError(
"scheduler.attempt_deadline must exceed the 300s Cloud Run timeout "
"and be at most 1800s"
)
scheduler["attempt_deadline"] = normalized_deadline
elif has_attempt_deadline:
raise ValueError("scheduler.attempt_deadline must not be null")
return scheduler


def _validate_near_run_warmup_schedule(
service_name: str,
scheduler: Mapping[str, str],
) -> None:
if service_name not in NEAR_RUN_WARMUP_SERVICES:
return
if scheduler.get("timezone") != "America/New_York":
raise ValueError(f"{service_name} warmup timezone must be America/New_York")
if scheduler.get("probe_time") != NEAR_RUN_WARMUP_SCHEDULE:
raise ValueError(
f"{service_name} warmup schedule must be {NEAR_RUN_WARMUP_SCHEDULE!r}"
)


def _validate_profile_inputs(
*,
service_name: str,
Expand Down
18 changes: 9 additions & 9 deletions tests/test_ibkr_portfolio.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,15 +77,15 @@ def positions(self):

def accountValues(self):
return [
SimpleNamespace(account="U16608560", currency="USD", tag="NetLiquidation", value="1130"),
SimpleNamespace(account="U16608560", currency="USD", tag="AvailableFunds", value="885.99"),
SimpleNamespace(account="U16608560", currency="USD", tag="CashBalance", value="477.10"),
SimpleNamespace(account="U16608560", currency="HKD", tag="CashBalance", value="408.98"),
SimpleNamespace(account="DEMO_ACCOUNT", currency="USD", tag="NetLiquidation", value="1130"),
SimpleNamespace(account="DEMO_ACCOUNT", currency="USD", tag="AvailableFunds", value="885.99"),
SimpleNamespace(account="DEMO_ACCOUNT", currency="USD", tag="CashBalance", value="477.10"),
SimpleNamespace(account="DEMO_ACCOUNT", currency="HKD", tag="CashBalance", value="408.98"),
]

snapshot = fetch_portfolio_snapshot(
MultiCurrencyIB(),
account_ids=("U16608560",),
account_ids=("DEMO_ACCOUNT",),
wait_seconds=0,
currency="USD",
)
Expand All @@ -103,14 +103,14 @@ def positions(self):

def accountValues(self):
return [
SimpleNamespace(account="U16608560", currency="USD", tag="NetLiquidation", value="2160"),
SimpleNamespace(account="U16608560", currency="USD", tag="AvailableFunds", value="1588.89"),
SimpleNamespace(account="U16608560", currency="USD", tag="CashBalance", value="-284.0"),
SimpleNamespace(account="DEMO_ACCOUNT", currency="USD", tag="NetLiquidation", value="2160"),
SimpleNamespace(account="DEMO_ACCOUNT", currency="USD", tag="AvailableFunds", value="1588.89"),
SimpleNamespace(account="DEMO_ACCOUNT", currency="USD", tag="CashBalance", value="-284.0"),
]

snapshot = fetch_portfolio_snapshot(
NegativeCashIB(),
account_ids=("U16608560",),
account_ids=("DEMO_ACCOUNT",),
wait_seconds=0,
currency="USD",
)
Expand Down
142 changes: 40 additions & 102 deletions tests/test_runtime_config_support.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1036,9 +1036,9 @@ def test_print_strategy_switch_env_plan_for_hk_global_etf_dry_run():
def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
payload = {
"defaults": {
"GLOBAL_TELEGRAM_CHAT_ID": "5992562050",
"GLOBAL_TELEGRAM_CHAT_ID": "test-chat",
"NOTIFY_LANG": "zh",
"IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME": "ibkr-account-groups",
"IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME": "example-account-groups",
"IB_GATEWAY_ZONE": "us-central1-c",
"IB_GATEWAY_IP_MODE": "internal",
"IBKR_MARKET": "HK",
Expand All @@ -1059,7 +1059,7 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
runtime_target_json(
"tqqq_growth_income",
deployment_selector="live-slot-a",
account_selector=["U1234567"],
account_selector=["DEMO_SLOT_A"],
account_scope="live-slot-a",
service_name="interactive-brokers-live-slot-a-service",
)
Expand All @@ -1084,7 +1084,7 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
runtime_target_json(
"russell_top50_leader_rotation",
deployment_selector="live-u7654-mega",
account_selector=["U7654321"],
account_selector=["DEMO_MEGA"],
account_scope="live-u7654-mega",
service_name="interactive-brokers-live-u7654-mega-service",
)
Expand Down Expand Up @@ -1136,6 +1136,7 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
"main_time": "10 16",
"probe_time": "40 9,15",
"precheck_time": "45 9",
"attempt_deadline": "330s",
}
assert "IBKR_FEATURE_SNAPSHOT_PATH" not in slot_a["env"]
assert "IBKR_FEATURE_SNAPSHOT_PATH" in slot_a["remove_env_vars"]
Expand All @@ -1151,6 +1152,7 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
"main_time": "45 15 1-7 * *",
"probe_time": "35 9,15 1-7 * *",
"precheck_time": "45 9 1-7 * *",
"attempt_deadline": "330s",
}
assert u7654_mega["env"]["IBKR_FEATURE_SNAPSHOT_PATH"] == "gs://runtime/mega/snapshot.csv"
assert (
Expand All @@ -1164,96 +1166,56 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
assert u7654_mega["env"]["DCA_BASE_INVESTMENT_USD"] == "500"


def _four_gateway_warmup_payload(probe_time: str) -> dict[str, object]:
gateway_targets = (
(
"interactive-brokers-quant-live-u15998061-service",
"live-u15998061",
"soxl_soxx_trend_income",
True,
),
(
"interactive-brokers-quant-live-u16608560-service",
"live-u16608560",
"tqqq_growth_income",
True,
),
(
"interactive-brokers-quant-live-u18336562-service",
"live-u18336562",
"russell_top50_leader_rotation",
False,
),
(
"interactive-brokers-quant-live-u18308207-service",
"live-u18308207",
"global_etf_rotation",
False,
),
)
def _generic_scheduler_payload() -> dict[str, object]:
targets = []
for service_name, account_group, strategy_profile, enabled in gateway_targets:
for suffix, dry_run_only, attempt_deadline in (
("alpha", False, None),
("beta", True, None),
("gamma", False, "420s"),
):
service_name = f"example-broker-{suffix}-service"
account_group = f"example-{suffix}"
scheduler = {
"timezone": "Etc/UTC",
"main_time": "10 16 * * 1-5",
"probe_time": "8 16 * * 1-5",
"precheck_time": "5 16 * * 1-5",
}
if attempt_deadline is not None:
scheduler["attempt_deadline"] = attempt_deadline
targets.append(
{
"service": service_name,
"account_group": account_group,
"runtime_target_enabled": enabled,
"runtime_target_enabled": True,
"runtime_target": {
**json.loads(
runtime_target_json(
strategy_profile,
"tqqq_growth_income",
dry_run_only=dry_run_only,
deployment_selector=account_group,
account_selector=[f"DEMO_{suffix.upper()}"],
account_scope=account_group,
service_name=service_name,
)
),
"scheduler": {
"timezone": "America/New_York",
"main_time": "45 15 * * 1-5",
"probe_time": probe_time,
"precheck_time": "45 9 * * 1-5",
},
"scheduler": scheduler,
},
}
)

targets.append(
{
"service": "interactive-brokers-us-combo-shadow-service",
"account_group": "us-combo-shadow",
"runtime_target_enabled": False,
"runtime_target": {
**json.loads(
runtime_target_json(
"us_equity_combo_leveraged",
dry_run_only=True,
deployment_selector="us-combo-shadow",
account_scope="us-combo-shadow",
service_name="interactive-brokers-us-combo-shadow-service",
)
),
"scheduler": {
"timezone": "America/New_York",
"main_time": "45 15 * * 1-5",
"probe_time": "35 9,15 * * *",
"precheck_time": "45 9 * * *",
},
},
}
)
return {
"defaults": {
"GLOBAL_TELEGRAM_CHAT_ID": "test-chat",
"NOTIFY_LANG": "zh",
"IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME": "ibkr-account-groups",
"EXECUTION_REPORT_GCS_URI": "gs://runtime/execution-reports",
"IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME": "example-account-groups",
"EXECUTION_REPORT_GCS_URI": "gs://example-runtime/execution-reports",
},
"targets": targets,
}


def test_build_cloud_run_env_sync_plan_generates_four_gateway_near_run_warmups() -> None:
payload = _four_gateway_warmup_payload("43 9,15 * * 1-5")
def test_build_cloud_run_env_sync_plan_uses_per_target_scheduler_policy() -> None:
payload = _generic_scheduler_payload()
result = subprocess.run(
[sys.executable, str(SYNC_PLAN_SCRIPT_PATH), "--json"],
check=True,
Expand All @@ -1264,39 +1226,15 @@ def test_build_cloud_run_env_sync_plan_generates_four_gateway_near_run_warmups()

plan = json.loads(result.stdout)
by_service = {target["service_name"]: target for target in plan["targets"]}
gateway_accounts = ("u15998061", "u16608560", "u18336562", "u18308207")
for account in gateway_accounts:
service_name = f"interactive-brokers-quant-live-{account}-service"
assert by_service[service_name]["scheduler"] == {
"timezone": "America/New_York",
"main_time": "45 15 * * 1-5",
"probe_time": "43 9,15 * * 1-5",
"precheck_time": "45 9 * * 1-5",
"attempt_deadline": "330s",
}
assert f"{service_name.removesuffix('-service')}-warmup-scheduler" == (
f"interactive-brokers-quant-live-{account}-warmup-scheduler"
)

assert by_service["interactive-brokers-us-combo-shadow-service"]["scheduler"]["probe_time"] == (
"35 9,15 * * *"
)
assert "attempt_deadline" not in by_service["interactive-brokers-us-combo-shadow-service"]["scheduler"]


def test_build_cloud_run_env_sync_plan_rejects_stale_gateway_warmup_schedule() -> None:
payload = _four_gateway_warmup_payload("35 9,15 * * 1-5")
result = subprocess.run(
[sys.executable, str(SYNC_PLAN_SCRIPT_PATH), "--json"],
check=False,
capture_output=True,
text=True,
env={**os.environ, "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(payload)},
)

assert result.returncode != 0
assert "interactive-brokers-quant-live-u15998061-service" in result.stderr
assert "43 9,15 * * 1-5" in result.stderr
assert by_service["example-broker-alpha-service"]["scheduler"] == {
"timezone": "Etc/UTC",
"main_time": "10 16 * * 1-5",
"probe_time": "8 16 * * 1-5",
"precheck_time": "5 16 * * 1-5",
"attempt_deadline": "330s",
}
assert "attempt_deadline" not in by_service["example-broker-beta-service"]["scheduler"]
assert by_service["example-broker-gamma-service"]["scheduler"]["attempt_deadline"] == "420s"


def test_build_cloud_run_env_sync_plan_requires_target_snapshot_in_per_service_mode():
Expand Down
Loading
Loading