Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions .github/workflows/execution-report-heartbeat.yml
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ jobs:
RUNTIME_HEARTBEAT_ACCEPT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STAGES }}
RUNTIME_HEARTBEAT_REJECT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STAGES }}
RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }}
RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: ${{ vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS || 'false' }}
RUNTIME_HEARTBEAT_MARKET_CALENDAR: ${{ vars.FIRSTRADE_MARKET_CALENDAR }}
RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.FIRSTRADE_MARKET_TIMEZONE }}
RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES: ${{ vars.RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES || '30' }}
Expand Down
38 changes: 29 additions & 9 deletions scripts/execution_report_heartbeat.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from quant_platform_kit.common.operational_notification_localization import (
format_operational_alert,
format_operational_heartbeat_status,
operational_notification_text,
resolve_operational_notification_locale,
)
Expand Down Expand Up @@ -639,6 +640,16 @@ def _send_telegram(message: str) -> bool:
return ok


def _send_normal_heartbeat(name: str, detail: str) -> int:
if not _env_bool("RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS", False):
return 0
message = format_operational_heartbeat_status(
locale=_notification_locale(), name=name, detail=detail,
)
print(message)
return 0 if _send_telegram(message[:3900]) else 1


def main(now: dt.datetime | None = None) -> int:
project = (
os.environ.get("RUNTIME_HEARTBEAT_GCP_PROJECT_ID")
Expand Down Expand Up @@ -668,6 +679,7 @@ def main(now: dt.datetime | None = None) -> int:
now = now.replace(tzinfo=dt.timezone.utc)
now = now.astimezone(dt.timezone.utc)
since = now - dt.timedelta(hours=lookback_hours)
due_since = now - dt.timedelta(hours=24)
runtime_targets = load_runtime_targets(os.environ)
try:
runtime_targets = _hydrate_runtime_target_schedules(
Expand All @@ -693,14 +705,14 @@ def main(now: dt.datetime | None = None) -> int:
raise ValueError("RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES must be non-negative")
due_targets, target_schedule_evaluated = filter_due_targets(
runtime_targets,
since=since,
since=due_since,
now=now,
market_aware=_env_bool("RUNTIME_HEARTBEAT_MARKET_AWARE", True),
publication_grace=dt.timedelta(minutes=publication_grace_minutes),
)
if runtime_targets and target_schedule_evaluated and not due_targets:
target_names = ", ".join(target_label(target) for target in runtime_targets)
schedule_reason = _heartbeat_skip_reason_for_schedule(since, now)
schedule_reason = _heartbeat_skip_reason_for_schedule(due_since, now)
print(
f"Execution report heartbeat skipped for {name}: "
+ (
Expand All @@ -709,12 +721,12 @@ def main(now: dt.datetime | None = None) -> int:
f"({target_names})"
)
)
return 0
return _send_normal_heartbeat(name, _notice("heartbeat_no_scheduled_window_due"))
if not runtime_targets:
schedule_skip_reason = _heartbeat_skip_reason_for_schedule(since, now)
schedule_skip_reason = _heartbeat_skip_reason_for_schedule(due_since, now)
if schedule_skip_reason:
print(f"Execution report heartbeat skipped for {name}: {schedule_skip_reason}")
return 0
return _send_normal_heartbeat(name, _notice("heartbeat_no_scheduler_run_due"))
if due_targets:
required_services = filter_services_for_targets(
required_services,
Expand Down Expand Up @@ -780,6 +792,9 @@ def main(now: dt.datetime | None = None) -> int:
if not matches:
inspected.append(f"- {updated.isoformat()} {uri} skipped {filter_reason}")
continue
if updated < due_since:
inspected.append(f"- {updated.isoformat()} {uri} skipped predates current heartbeat window")
continue
latest_due_at = required_due_at.get(service_name)
if latest_due_at is not None and updated < latest_due_at:
inspected.append(
Expand All @@ -797,19 +812,24 @@ def main(now: dt.datetime | None = None) -> int:

if required_keys:
missing = [key for key in required_keys if key not in accepted_by_service]
if not missing:
if not missing and not list_errors:
details = ", ".join(
f"{required_labels[key]}@{accepted_by_service[key][1].isoformat()}"
for key in required_keys
)
print(f"Execution report heartbeat OK for {name}: {details}")
return 0
if accepted:
return _send_normal_heartbeat(
name, _notice("heartbeat_accepted_report", detail=details),
)
if accepted and not list_errors:
uri, updated, reason = accepted[0]
print(
f"Execution report heartbeat OK for {name}: {reason}, updated={updated.isoformat()}, uri={uri}"
)
return 0
return _send_normal_heartbeat(
name,
_notice("heartbeat_accepted_report", detail=f"{reason}, {updated.isoformat()}"),
)

issues = []
technical_details = []
Expand Down
Loading