Skip heartbeat outside Firstrade scheduler window

Author: Pigbibi

.github/workflows/execution-report-heartbeat.yml

@@ -42,6 +42,7 @@ jobs:
       RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT: ${{ inputs.fail_workflow_on_alert || vars.RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT || 'true' }}
       RUNTIME_HEARTBEAT_ACCEPT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STAGES }}
       RUNTIME_HEARTBEAT_REJECT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STAGES }}
+      RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON }}
       FIRSTRADE_GCS_STATE_BUCKET: ${{ vars.FIRSTRADE_GCS_STATE_BUCKET }}
       FIRSTRADE_STATE_PREFIX: ${{ vars.FIRSTRADE_STATE_PREFIX }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }}

scripts/execution_report_heartbeat.py

@@ -12,6 +12,7 @@
 import urllib.parse
 import urllib.request
 from typing import Any
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 
 DEFAULT_ACCEPT_STATUSES = {"ok", "skipped", "success", "completed", "no_action"}
@@ -42,6 +43,79 @@ def _env_bool(name: str, default: bool = False) -> bool:
     return value in {"1", "true", "yes", "y", "on"}
 
 
+def _parse_day_of_month_field(raw: str) -> set[int] | None:
+    text = str(raw or "").strip()
+    if not text or text in {"*", "?"}:
+        return None
+    days: set[int] = set()
+    for part in text.split(","):
+        part = part.strip()
+        if not part:
+            continue
+        step = 1
+        if "/" in part:
+            part, step_text = part.split("/", 1)
+            try:
+                step = max(1, int(step_text))
+            except ValueError:
+                return None
+        if "-" in part:
+            start_text, end_text = part.split("-", 1)
+            try:
+                start = int(start_text)
+                end = int(end_text)
+            except ValueError:
+                return None
+            if start > end:
+                return None
+            days.update(day for day in range(start, end + 1, step) if 1 <= day <= 31)
+            continue
+        try:
+            day = int(part)
+        except ValueError:
+            return None
+        if 1 <= day <= 31:
+            days.add(day)
+    return days or None
+
+
+def _runtime_target_scheduler() -> dict[str, Any]:
+    raw = (os.environ.get("RUNTIME_TARGET_JSON") or "").strip()
+    if not raw:
+        return {}
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        return {}
+    scheduler = payload.get("scheduler") if isinstance(payload, dict) else None
+    return scheduler if isinstance(scheduler, dict) else {}
+
+
+def _heartbeat_skip_reason_for_schedule(now: dt.datetime) -> str | None:
+    scheduler = _runtime_target_scheduler()
+    cron = str(scheduler.get("main_time") or "").strip()
+    fields = cron.split()
+    if len(fields) != 5:
+        return None
+    expected_days = _parse_day_of_month_field(fields[2])
+    if not expected_days:
+        return None
+    timezone_name = str(scheduler.get("timezone") or "UTC").strip() or "UTC"
+    try:
+        timezone = ZoneInfo(timezone_name)
+    except ZoneInfoNotFoundError:
+        timezone = dt.timezone.utc
+        timezone_name = "UTC"
+    local_now = now.astimezone(timezone)
+    if local_now.day in expected_days:
+        return None
+    day_text = ",".join(str(day) for day in sorted(expected_days))
+    return (
+        f"runtime scheduler main_time is not due today "
+        f"({timezone_name} day={local_now.day}; expected day(s)={day_text})"
+    )
+
+
 def _parse_timestamp(value: Any) -> dt.datetime | None:
     if not value:
         return None
@@ -353,7 +427,7 @@ def _send_telegram(message: str) -> bool:
     return ok
 
 
-def main() -> int:
+def main(now: dt.datetime | None = None) -> int:
     project = (
         os.environ.get("RUNTIME_HEARTBEAT_GCP_PROJECT_ID")
         or os.environ.get("GCP_PROJECT_ID")
@@ -365,7 +439,12 @@ def main() -> int:
     fail_workflow = _env_bool("RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT", True)
     required_services = _load_required_services()
 
-    now = dt.datetime.now(dt.timezone.utc)
+    now = now or dt.datetime.now(dt.timezone.utc)
+    schedule_skip_reason = _heartbeat_skip_reason_for_schedule(now)
+    if schedule_skip_reason:
+        print(f"Execution report heartbeat skipped for {name}: {schedule_skip_reason}")
+        return 0
+
     since = now - dt.timedelta(hours=lookback_hours)
     globs = _report_globs(since, now)
     if not globs:

tests/test_execution_report_heartbeat.py

@@ -52,3 +52,36 @@ def fake_run_gcloud(command):
         "firstradequant",
     ]
 
+
+
+def test_heartbeat_skips_outside_runtime_target_scheduler_day(monkeypatch, capsys):
+    monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "Firstrade monthly runtime")
+    monkeypatch.setenv(
+        "RUNTIME_TARGET_JSON",
+        '{"scheduler":{"timezone":"America/New_York","main_time":"45 15 25-28 * *"}}',
+    )
+    monkeypatch.setattr(
+        heartbeat,
+        "_list_gcs_objects",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("GCS should not be queried")),
+    )
+
+    result = heartbeat.main(now=dt.datetime(2026, 6, 20, 23, 10, tzinfo=dt.timezone.utc))
+
+    assert result == 0
+    output = capsys.readouterr().out
+    assert "Execution report heartbeat skipped for Firstrade monthly runtime" in output
+    assert "expected day(s)=25,26,27,28" in output
+
+
+def test_heartbeat_does_not_skip_inside_runtime_target_scheduler_day(monkeypatch):
+    monkeypatch.setenv(
+        "RUNTIME_TARGET_JSON",
+        '{"scheduler":{"timezone":"America/New_York","main_time":"45 15 25-28 * *"}}',
+    )
+
+    reason = heartbeat._heartbeat_skip_reason_for_schedule(
+        dt.datetime(2026, 6, 25, 23, 10, tzinfo=dt.timezone.utc)
+    )
+
+    assert reason is None