diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml index 4e29345..12af2e0 100644 --- a/.github/workflows/execution-report-heartbeat.yml +++ b/.github/workflows/execution-report-heartbeat.yml @@ -1,6 +1,5 @@ name: Execution Report Heartbeat -# Schedule disabled; trade/error notifications own unattended alerting. on: workflow_dispatch: inputs: @@ -17,6 +16,8 @@ on: options: - "true" - "false" + schedule: + - cron: "20 22 * * *" env: GCP_PROJECT_ID: charlesschwabquant @@ -44,9 +45,14 @@ jobs: RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT: ${{ inputs.fail_workflow_on_alert || vars.RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT || 'true' }} RUNTIME_HEARTBEAT_ACCEPT_STATUSES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STATUSES }} RUNTIME_HEARTBEAT_REJECT_STATUSES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STATUSES }} + RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }} + RUNTIME_HEARTBEAT_MARKET_CALENDAR: ${{ vars.SCHWAB_MARKET_CALENDAR }} + RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.SCHWAB_MARKET_TIMEZONE }} RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }} RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON }} CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }} + CLOUD_RUN_SERVICES: ${{ vars.CLOUD_RUN_SERVICES }} + CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }} GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }} steps: @@ -54,6 +60,19 @@ jobs: uses: actions/checkout@v6 - name: Authenticate to Google Cloud + id: gcp_auth_primary + continue-on-error: true + uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + + - name: Wait before Google Cloud authentication retry + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} + run: sleep 10 + + - name: Retry Google Cloud authentication + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} uses: google-github-actions/auth@v3 with: workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} @@ -62,5 +81,11 @@ jobs: - name: Set up gcloud uses: google-github-actions/setup-gcloud@v3 + - name: Install market calendar + continue-on-error: true + run: >- + python -m pip install --disable-pip-version-check + --retries 3 --timeout 30 "pandas-market-calendars==5.4.0" + - name: Check recent execution report run: python scripts/execution_report_heartbeat.py diff --git a/.github/workflows/runtime-guard.yml b/.github/workflows/runtime-guard.yml index b630849..90561cd 100644 --- a/.github/workflows/runtime-guard.yml +++ b/.github/workflows/runtime-guard.yml @@ -64,6 +64,19 @@ jobs: uses: actions/checkout@v6 - name: Authenticate to Google Cloud + id: gcp_auth_primary + continue-on-error: true + uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + + - name: Wait before Google Cloud authentication retry + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} + run: sleep 10 + + - name: Retry Google Cloud authentication + if: ${{ steps.gcp_auth_primary.outcome == 'failure' }} uses: google-github-actions/auth@v3 with: workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 9b31b96..846707e 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -52,6 +52,32 @@ def _record_platform_execution_telemetry( ) +def _build_notification_delivery_summary(delivery_events: list[dict]) -> dict: + safe_fields = ( + "sink", + "delivery_status", + "transport_acknowledged", + "error_type", + "compact_text_sha256", + "compact_text_length", + ) + events = [ + {key: event[key] for key in safe_fields if key in event} + for event in (dict(item) for item in delivery_events) + ] + if not events: + return {} + sent_count = sum(event.get("delivery_status") == "sent" for event in events) + failed_count = sum(event.get("delivery_status") == "failed" for event in events) + return { + "attempted_count": len(events), + "sent_count": sent_count, + "failed_count": failed_count, + "all_acknowledged": failed_count == 0 and sent_count == len(events), + "delivery_events": events, + } + + def _plan_portfolio(plan): return dict(plan.get("portfolio") or {}) @@ -596,5 +622,10 @@ def load_plan(current_snapshot): + json.dumps({"reason": "no_trade_or_error", "trade_logs_count": len(trade_logs)}, ensure_ascii=False), flush=True, ) + notification_delivery_summary = _build_notification_delivery_summary( + getattr(runtime, "notification_delivery_events", []) or [] + ) + if notification_delivery_summary: + execution["notification_delivery_summary"] = notification_delivery_summary _record_platform_execution_telemetry(config, execution_result) return execution_result diff --git a/application/runtime_composer.py b/application/runtime_composer.py index 4c7a3a6..638b7fd 100644 --- a/application/runtime_composer.py +++ b/application/runtime_composer.py @@ -57,26 +57,25 @@ class SchwabRuntimeComposer: printer: Callable[..., Any] = print notification_channel: str = "telegram" webhook_url: str | None = None - sender_builder: Callable[..., Callable[[str], None]] | None = None + sender_builder: Callable[..., Callable[[str], bool | None]] | None = None notification_builder: Callable[..., Any] = build_runtime_notification_adapters reporting_builder: Callable[..., Any] = build_runtime_reporting_adapters runtime_target: RuntimeTarget | None = None limit_buy_premium_by_symbol: dict[str, float] | None = None extra_reporting_fields: dict[str, Any] = field(default_factory=dict) - def send_message(self, message: str) -> None: + def send_message(self, message: str) -> bool: """Send a cycle notification through the configured channel.""" if self.sender_builder is not None: sender = self.sender_builder(self.tg_token, self.tg_chat_id) - sender(message) - return + return sender(message) is not False sender = build_cycle_sender( channel=self.notification_channel, telegram_token=self.tg_token, telegram_chat_id=self.tg_chat_id, webhook_url=self.webhook_url, ) - sender(message) + return sender(message) send_tg_message = send_message # backward-compat alias @@ -152,6 +151,11 @@ def build_rebalance_runtime(self, client, *, silent_cycle_notifications: bool = client, account_hash, ), + notification_delivery_events=getattr( + notification_adapters, + "delivery_events", + [], + ), ) def build_rebalance_config( diff --git a/application/runtime_dependencies.py b/application/runtime_dependencies.py index b6b3a2e..3b10e5d 100644 --- a/application/runtime_dependencies.py +++ b/application/runtime_dependencies.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Callable, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any from quant_platform_kit.common.ports import ExecutionPort, MarketDataPort, NotificationPort, PortfolioPort @@ -42,3 +42,4 @@ class SchwabRebalanceRuntime: execution_port_factory: Callable[[str], ExecutionPort] | None = None order_status_fetcher_factory: Callable[[str], Callable[[str], Any] | None] | None = None submit_equity_order: Callable[..., Any] | None = None + notification_delivery_events: list[dict[str, Any]] = field(default_factory=list) diff --git a/application/runtime_notification_adapters.py b/application/runtime_notification_adapters.py index 9a350de..b5d8dea 100644 --- a/application/runtime_notification_adapters.py +++ b/application/runtime_notification_adapters.py @@ -2,7 +2,9 @@ from __future__ import annotations +import hashlib from dataclasses import dataclass +from typing import Any from notifications.events import NotificationPublisher, RenderedNotification from quant_platform_kit.common.port_adapters import CallableNotificationPort @@ -13,14 +15,20 @@ class SchwabNotificationAdapters: notification_port: NotificationPort cycle_publisher: NotificationPublisher + delivery_events: list[dict[str, Any]] - def publish_cycle_notification(self, *, detailed_text: str, compact_text: str) -> None: - self.cycle_publisher.publish( + def publish_cycle_notification(self, *, detailed_text: str, compact_text: str) -> bool: + before_count = len(self.delivery_events) + outcome = self.cycle_publisher.publish( RenderedNotification( detailed_text=detailed_text, compact_text=compact_text, ) ) + deliveries = self.delivery_events[before_count:] + if deliveries: + return all(event.get("delivery_status") == "sent" for event in deliveries) + return outcome is not False def build_runtime_notification_adapters( @@ -28,11 +36,44 @@ def build_runtime_notification_adapters( send_message, notification_channel: str = "telegram", log_message=None, + delivery_events: list[dict[str, Any]] | None = None, ) -> SchwabNotificationAdapters: + recorded_delivery_events = delivery_events if delivery_events is not None else [] + + def send_recorded_message(message: str) -> bool: + compact = str(message or "") + event = { + "sink": notification_channel, + "compact_text_sha256": hashlib.sha256(compact.encode("utf-8")).hexdigest(), + "compact_text_length": len(compact), + } + try: + outcome = send_message(message) + except Exception as exc: + event.update( + { + "delivery_status": "failed", + "transport_acknowledged": False, + "error_type": type(exc).__name__, + } + ) + recorded_delivery_events.append(event) + return False + acknowledged = outcome is not False + event.update( + { + "delivery_status": "sent" if acknowledged else "failed", + "transport_acknowledged": acknowledged, + } + ) + recorded_delivery_events.append(event) + return acknowledged + return SchwabNotificationAdapters( - notification_port=CallableNotificationPort(send_message), + notification_port=CallableNotificationPort(send_recorded_message), cycle_publisher=NotificationPublisher( log_message=log_message or (lambda message: print(message, flush=True)), - send_message=send_message, + send_message=send_recorded_message, ), + delivery_events=recorded_delivery_events, ) diff --git a/application/runtime_report_summary.py b/application/runtime_report_summary.py index a8b4876..3bb9eed 100644 --- a/application/runtime_report_summary.py +++ b/application/runtime_report_summary.py @@ -52,4 +52,7 @@ def summarize_execution_cycle_result(result: object, *, dry_run: bool) -> dict[s value = execution.get(field_name) if value is not None and value != "": summary[field_name] = value + notification_delivery_summary = _as_mapping(execution.get("notification_delivery_summary")) + if notification_delivery_summary: + summary["notification_delivery_summary"] = notification_delivery_summary return summary diff --git a/main.py b/main.py index 1057b5b..34c54ae 100644 --- a/main.py +++ b/main.py @@ -407,7 +407,7 @@ def send_tg_message(message): def publish_notification(*, detailed_text, compact_text): - build_composer().build_notification_adapters().publish_cycle_notification( + return build_composer().build_notification_adapters().publish_cycle_notification( detailed_text=detailed_text, compact_text=compact_text, ) @@ -469,22 +469,35 @@ def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool: print("Schwab runtime error notification skipped: no Telegram target configured.", flush=True) return False message = _runtime_error_notification_message(exc, route_label=route_label) + outcomes = [] for token, chat_id in targets: try: - requests.post( + response = requests.post( f"https://api.telegram.org/bot{token}/sendMessage", json={"chat_id": chat_id, "text": message}, timeout=15, ) + status_code = int(getattr(response, "status_code", 200) or 200) + acknowledged = 200 <= status_code < 300 + load_payload = getattr(response, "json", None) + payload = load_payload() if acknowledged and callable(load_payload) else None + if isinstance(payload, dict) and payload.get("ok") is False: + acknowledged = False + outcomes.append(acknowledged) except Exception as send_exc: - print(f"Schwab runtime error Telegram send failed: {send_exc}", flush=True) - return True + print( + f"Schwab runtime error Telegram send failed: {type(send_exc).__name__}", + flush=True, + ) + outcomes.append(False) + return bool(outcomes) and all(outcomes) def _publish_runtime_failure_notification(*, detailed_text: str, compact_text: str, exc: Exception) -> bool: try: - publish_notification(detailed_text=detailed_text, compact_text=compact_text) - return True + if publish_notification(detailed_text=detailed_text, compact_text=compact_text): + return True + return _notify_runtime_error(exc, route_label="strategy_cycle") except Exception as notification_exc: print(f"Schwab runtime error notification fallback: {notification_exc}", flush=True) return _notify_runtime_error(exc, route_label="strategy_cycle") diff --git a/notifications/telegram.py b/notifications/telegram.py index 834f4d0..085af39 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -417,17 +417,28 @@ def build_sender(token, chat_id, *, requests_module=None): if requests_module is None: import requests as requests_module - def send_tg_message(message): + def send_tg_message(message) -> bool: if not token or not chat_id: - return + return False url = f"https://api.telegram.org/bot{token}/sendMessage" try: - requests_module.post( + response = requests_module.post( url, json={"chat_id": chat_id, "text": _break_telegram_market_symbol_auto_links(message)}, timeout=15, ) + status_code = int(getattr(response, "status_code", 200) or 200) + if status_code < 200 or status_code >= 300: + print(f"Telegram send failed: HTTP {status_code}", flush=True) + return False + load_payload = getattr(response, "json", None) + payload = load_payload() if callable(load_payload) else None + if isinstance(payload, dict) and payload.get("ok") is False: + print("Telegram send failed: negative API acknowledgement", flush=True) + return False except Exception as exc: print(f"Telegram send failed: {type(exc).__name__}", flush=True) + return False + return True return send_tg_message diff --git a/scripts/cloud_run_runtime_guard.py b/scripts/cloud_run_runtime_guard.py index 9fbf51b..7c846af 100644 --- a/scripts/cloud_run_runtime_guard.py +++ b/scripts/cloud_run_runtime_guard.py @@ -23,6 +23,7 @@ "URL_ERROR", "URL_UNREACHABLE", ) +SCHEDULER_CLOUD_RUN_DEDUP_SECONDS = 120 def _split_values(raw: str | None) -> list[str]: @@ -135,6 +136,41 @@ def _scheduler_entry_since( return max(matches) if matches else fallback +def _is_duplicate_scheduler_failure( + entry: dict[str, Any], + cloud_run_failures_by_service: dict[str, list[dict[str, Any]]], +) -> bool: + scheduler_timestamp = _parse_timestamp(entry.get("timestamp")) + job_name = _entry_job_name(entry) + if scheduler_timestamp is None or not job_name: + return False + + tolerance = dt.timedelta(seconds=SCHEDULER_CLOUD_RUN_DEDUP_SECONDS) + for service, failures in cloud_run_failures_by_service.items(): + if not any(alias and alias in job_name for alias in _service_job_aliases(service)): + continue + for failure in failures: + cloud_run_timestamp = _parse_timestamp(failure.get("timestamp")) + if ( + cloud_run_timestamp is not None + and abs(scheduler_timestamp - cloud_run_timestamp) <= tolerance + ): + return True + return False + + +def _services_without_success( + services: list[str], + success_count_by_service: dict[str, int], + queried_services: set[str], +) -> list[str]: + return [ + service + for service in services + if service in queried_services and success_count_by_service.get(service, 0) == 0 + ] + + def _run_gcloud(args: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(args, text=True, capture_output=True, check=False) @@ -504,6 +540,9 @@ def main() -> int: issues: list[str] = [] details: list[str] = [] success_count = 0 + success_count_by_service: dict[str, int] = {} + queried_services: set[str] = set() + cloud_run_failures_by_service: dict[str, list[dict[str, Any]]] = {} service_since_by_name: dict[str, dt.datetime] = {} try: @@ -526,16 +565,26 @@ def main() -> int: except RuntimeError as exc: issues.append(f"Cloud Run log query failed for {service}: {exc}") continue + queried_services.add(service) failures = [entry for entry in entries if _is_failure(entry)] - success_count += sum(1 for entry in entries if _is_success(entry)) + cloud_run_failures_by_service[service] = failures + service_success_count = sum(1 for entry in entries if _is_success(entry)) + success_count_by_service[service] = service_success_count + success_count += service_success_count if failures: issues.append(f"{len(failures)} Cloud Run failure log(s) for {service}") details.extend(_summarize(entry) for entry in failures[:5]) - if services and require_success and success_count == 0: - issues.append( - f"no successful Cloud Run request found for {', '.join(services)} in the last {lookback_minutes} minutes" - ) + if services and require_success: + for service in _services_without_success( + services, + success_count_by_service, + queried_services, + ): + issues.append( + f"no successful Cloud Run request found for {service} " + f"in the last {lookback_minutes} minutes" + ) if check_scheduler and scheduler_pattern: log_filter = f'resource.type="cloud_scheduler_job" AND timestamp >= "{since_text}"' @@ -556,6 +605,11 @@ def main() -> int: entry_since = _scheduler_entry_since(entry, service_since_by_name, since) if entry_timestamp and entry_timestamp < entry_since: continue + if _is_duplicate_scheduler_failure( + entry, + cloud_run_failures_by_service, + ): + continue failures.append(entry) if failures: issues.append(f"{len(failures)} Cloud Scheduler failure log(s)") diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index 1649394..f3b4bae 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -14,6 +14,27 @@ from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +try: + from scripts.runtime_heartbeat_policy import ( + filter_due_targets, + filter_services_for_targets, + load_runtime_targets, + match_payload_target, + target_key, + target_label, + target_latest_due_at, + ) +except ModuleNotFoundError: + from runtime_heartbeat_policy import ( # type: ignore[no-redef] + filter_due_targets, + filter_services_for_targets, + load_runtime_targets, + match_payload_target, + target_key, + target_label, + target_latest_due_at, + ) + DEFAULT_ACCEPT_STATUSES = {"ok", "skipped", "success", "completed", "no_action"} DEFAULT_REJECT_STATUSES = {"error", "failed", "failure", "cancelled", "canceled", "timed_out"} @@ -194,6 +215,8 @@ def _base_report_uris() -> list[str]: def _load_required_services() -> list[str]: services = [] + enabled_target_services = [] + disabled_target_services = [] for name in ( "RUNTIME_HEARTBEAT_REQUIRED_SERVICES", "CLOUD_RUN_SERVICES", @@ -216,20 +239,31 @@ def _load_required_services() -> list[str]: runtime_target = json.loads(runtime_target) except json.JSONDecodeError: runtime_target = {} + enabled_value = target.get("runtime_target_enabled") + if enabled_value is None: + enabled_value = target.get("RUNTIME_TARGET_ENABLED") + if enabled_value is None and isinstance(runtime_target, dict): + enabled_value = runtime_target.get("runtime_target_enabled") for key in ("service", "service_name", "cloud_run_service"): value = target.get(key) or ( runtime_target.get(key) if isinstance(runtime_target, dict) else None ) if value: - services.extend(_split_values(str(value))) + target_services = _split_values(str(value)) + if _enabled_value(enabled_value, default=True): + enabled_target_services.extend(target_services) + else: + disabled_target_services.extend(target_services) break except json.JSONDecodeError: pass + services.extend(enabled_target_services) + disabled = set(disabled_target_services) - set(enabled_target_services) seen = set() unique = [] for service in services: - if service not in seen: + if service not in seen and service not in disabled: seen.add(service) unique.append(service) return unique @@ -318,6 +352,20 @@ def _report_status(payload: dict[str, Any]) -> tuple[str, str]: return status, stage +def _report_notification_failure(payload: dict[str, Any]) -> str: + scopes = [payload] + summary = payload.get("summary") + if isinstance(summary, dict): + scopes.append(summary) + for scope in scopes: + delivery_summary = scope.get("notification_delivery_summary") + if isinstance(delivery_summary, dict) and delivery_summary.get("all_acknowledged") is False: + return "notification delivery not acknowledged" + if scope.get("notification_error") and scope.get("notification_suppressed") is not True: + return "notification delivery failed" + return "" + + def _payload_service_name(payload: dict[str, Any]) -> str: runtime_target = payload.get("runtime_target") service = payload.get("service_name") @@ -334,7 +382,12 @@ def _payload_account_scope(payload: dict[str, Any]) -> str: return "" -def _payload_matches(payload: dict[str, Any], required_services: list[str]) -> tuple[bool, str, str]: +def _payload_matches( + payload: dict[str, Any], + required_services: list[str], + *, + required_targets: list[dict[str, Any]] | None = None, +) -> tuple[bool, str, str]: expected_platform = (os.environ.get("RUNTIME_HEARTBEAT_REPORT_PLATFORM") or "").strip().lower() if expected_platform: platform = str(payload.get("platform") or "").strip().lower() @@ -350,6 +403,16 @@ def _payload_matches(payload: dict[str, Any], required_services: list[str]) -> t service_name = _payload_service_name(payload) if required_services and service_name not in required_services: return False, service_name, f"service_name={service_name or '-'}" + if required_targets: + matched_target, reason = match_payload_target(payload, required_targets) + if matched_target: + return True, matched_target, reason + target_services = { + str(target.get("service") or "").strip() + for target in required_targets + } + if service_name in target_services: + return False, service_name, reason return True, service_name, "matched filters" @@ -376,8 +439,11 @@ def _is_accepted_report(payload: dict[str, Any]) -> tuple[bool, str]: status_key = status.lower() stage_key = stage.upper() errors = _report_errors(payload) + notification_failure = _report_notification_failure(payload) if errors and not allow_errors: return False, f"errors={len(errors)} status={status or '-'} stage={stage or '-'}" + if notification_failure: + return False, notification_failure if status_key in reject_statuses or stage_key in reject_stages: return False, f"rejected status={status or '-'} stage={stage or '-'}" if status_key and status_key in accepted_statuses: @@ -437,10 +503,61 @@ 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) - schedule_skip_reason = _heartbeat_skip_reason_for_schedule(since, now) - if schedule_skip_reason: - print(f"Execution report heartbeat skipped for {name}: {schedule_skip_reason}") + runtime_targets = load_runtime_targets(os.environ) + due_targets, target_schedule_evaluated = filter_due_targets( + runtime_targets, + since=since, + now=now, + market_aware=_env_bool("RUNTIME_HEARTBEAT_MARKET_AWARE", True), + ) + 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) + print( + f"Execution report heartbeat skipped for {name}: " + + ( + schedule_reason + or "no runtime target main schedule was due on an open market session " + f"({target_names})" + ) + ) return 0 + if not runtime_targets: + schedule_skip_reason = _heartbeat_skip_reason_for_schedule(since, now) + if schedule_skip_reason: + print(f"Execution report heartbeat skipped for {name}: {schedule_skip_reason}") + return 0 + if due_targets: + required_services = filter_services_for_targets( + required_services, + due_targets, + all_targets=runtime_targets, + ) + required_targets = [ + target + for target in due_targets + if not required_services or str(target.get("service") or "").strip() in required_services + ] + required_keys = [target_key(target) for target in required_targets] + required_labels = { + target_key(target): target_label(target) + for target in required_targets + } + required_due_at = { + target_key(target): due_at + for target in required_targets + if (due_at := target_latest_due_at(target)) is not None + } + target_services = { + str(target.get("service") or "").strip() + for target in required_targets + } + for service in required_services: + if service not in target_services: + required_keys.append(service) + required_labels[service] = service + if required_keys: + max_reports = max(max_reports, min(200, len(required_keys) * 4)) globs = _report_globs(since, now) if not globs: @@ -467,23 +584,35 @@ def main(now: dt.datetime | None = None) -> int: if payload is None: inspected.append(f"- {updated.isoformat()} {uri} unreadable") continue - matches, service_name, filter_reason = _payload_matches(payload, required_services) + matches, service_name, filter_reason = _payload_matches( + payload, + required_services, + required_targets=required_targets, + ) if not matches: inspected.append(f"- {updated.isoformat()} {uri} skipped {filter_reason}") continue + latest_due_at = required_due_at.get(service_name) + if latest_due_at is not None and updated < latest_due_at: + inspected.append( + f"- {updated.isoformat()} {uri} skipped " + f"predates latest due schedule {latest_due_at.isoformat()}" + ) + continue ok, reason = _is_accepted_report(payload) inspected.append(f"- {updated.isoformat()} {uri} {reason}") if ok: - if required_services: + if required_keys: accepted_by_service[service_name] = (uri, updated, reason) else: accepted.append((uri, updated, reason)) - if required_services: - missing = [service for service in required_services if service not in accepted_by_service] + if required_keys: + missing = [key for key in required_keys if key not in accepted_by_service] if not missing: details = ", ".join( - f"{service}@{accepted_by_service[service][1].isoformat()}" for service in required_services + 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 @@ -499,9 +628,12 @@ def main(now: dt.datetime | None = None) -> int: issues.extend(f"list failed: {item}" for item in list_errors[:3]) if not sorted_objects: issues.append(f"no report object updated in the last {lookback_hours:g} hours") - elif required_services: - missing = [service for service in required_services if service not in accepted_by_service] - issues.append(f"missing acceptable report for service(s): {', '.join(missing)}") + elif required_keys: + missing = [key for key in required_keys if key not in accepted_by_service] + issues.append( + "missing acceptable report for runtime target(s): " + + ", ".join(required_labels[key] for key in missing) + ) else: issues.append(f"no acceptable report among {min(len(sorted_objects), max_reports)} recent report object(s)") diff --git a/scripts/runtime_heartbeat_policy.py b/scripts/runtime_heartbeat_policy.py new file mode 100644 index 0000000..fd13fea --- /dev/null +++ b/scripts/runtime_heartbeat_policy.py @@ -0,0 +1,525 @@ +"""Per-runtime-target schedule and market-session policy for heartbeat checks.""" + +from __future__ import annotations + +import datetime as dt +import json +import sys +from collections.abc import Callable, Mapping +from typing import Any +from zoneinfo import ZoneInfo + + +SessionDatesLoader = Callable[..., set[dt.date]] +WarningLogger = Callable[[str], None] + +_LATEST_DUE_AT_KEY = "_heartbeat_latest_due_at" +_MARKET_DEFAULTS = { + "US": ("NYSE", "America/New_York"), + "HK": ("XHKG", "Asia/Hong_Kong"), + "CN": ("SSE", "Asia/Shanghai"), + "SG": ("XSES", "Asia/Singapore"), + "CRYPTO": ("24/7", "UTC"), +} +_TIMEZONE_MARKETS = { + "America/New_York": "US", + "Asia/Hong_Kong": "HK", + "Asia/Shanghai": "CN", + "Asia/Singapore": "SG", +} +_PLATFORM_MARKETS = { + "schwab": "US", + "firstrade": "US", + "qmt": "CN", + "binance": "CRYPTO", +} + + +def _split_values(raw: str | None) -> list[str]: + if not raw: + return [] + values = str(raw).replace(";", ",").replace("\n", ",").split(",") + return [value.strip() for value in values if value.strip()] + + +def _enabled(value: Any, *, default: bool = True) -> bool: + if value is None or not str(value).strip(): + return default + return str(value).strip().lower() not in {"0", "false", "no", "n", "off"} + + +def _runtime_target(item: Mapping[str, Any]) -> dict[str, Any]: + value = item.get("runtime_target") or item.get("runtime_target_json") + if isinstance(value, str): + try: + value = json.loads(value) + except json.JSONDecodeError: + value = {} + if isinstance(value, dict): + return value + return dict(item) + + +def _first_value(sources: list[Mapping[str, Any]], keys: tuple[str, ...]) -> str: + for source in sources: + for key in keys: + value = source.get(key) + if value is not None and str(value).strip(): + return str(value).strip() + return "" + + +def _suffix_value(sources: list[Mapping[str, Any]], suffix: str) -> str: + for source in sources: + for key, value in source.items(): + if str(key).upper().endswith(suffix) and value is not None and str(value).strip(): + return str(value).strip() + return "" + + +def _service_values( + item: Mapping[str, Any], + runtime_target: Mapping[str, Any], + environ: Mapping[str, str], +) -> list[str]: + value = _first_value( + [item, runtime_target], + ("service", "service_name", "cloud_run_service"), + ) + if value: + return _split_values(value) + services = _split_values(environ.get("CLOUD_RUN_SERVICES")) + services.extend(_split_values(environ.get("CLOUD_RUN_SERVICE"))) + return list(dict.fromkeys(services)) + + +def _normalize_target( + item: Mapping[str, Any], + runtime_target: Mapping[str, Any], + service: str, + environ: Mapping[str, str], + *, + use_global_market_fallback: bool, +) -> dict[str, Any]: + sources = [runtime_target, item] + scheduler = next( + ( + value + for source in sources + if isinstance((value := source.get("scheduler")), dict) + ), + {}, + ) + account_scope = _first_value( + sources, + ( + "account_scope", + "account_group", + "account_region", + "ACCOUNT_GROUP", + "ACCOUNT_REGION", + ), + ) + target_market_timezone = ( + _first_value(sources, ("market_timezone", "MARKET_TIMEZONE")) + or _suffix_value(sources, "_MARKET_TIMEZONE") + ) + global_market_timezone = ( + str(environ.get("RUNTIME_HEARTBEAT_MARKET_TIMEZONE") or "").strip() + or _suffix_value([environ], "_MARKET_TIMEZONE") + ) + market_timezone = target_market_timezone or ( + global_market_timezone if use_global_market_fallback else "" + ) + market = ( + _first_value(sources, ("market", "MARKET")) + or _suffix_value(sources, "_MARKET") + ).upper() + if not market: + market = _TIMEZONE_MARKETS.get(target_market_timezone, "") + if market not in _MARKET_DEFAULTS: + market = _TIMEZONE_MARKETS.get(str(scheduler.get("timezone") or "").strip(), "") + if not market: + platform_id = _first_value(sources, ("platform_id",)).lower() + market = _PLATFORM_MARKETS.get(platform_id, "") + if not market and account_scope.upper() in {"US", "HK", "CN"}: + market = account_scope.upper() + if market not in _MARKET_DEFAULTS and use_global_market_fallback: + market = str(environ.get("RUNTIME_HEARTBEAT_MARKET") or "").strip().upper() + target_market_calendar = ( + _first_value(sources, ("market_calendar", "MARKET_CALENDAR")) + or _suffix_value(sources, "_MARKET_CALENDAR") + ) + global_market_calendar = ( + str(environ.get("RUNTIME_HEARTBEAT_MARKET_CALENDAR") or "").strip() + or _suffix_value([environ], "_MARKET_CALENDAR") + ) + default_calendar, default_timezone = _MARKET_DEFAULTS.get(market, ("", "")) + market_calendar = ( + target_market_calendar + or (global_market_calendar if use_global_market_fallback else "") + or default_calendar + ) + market_timezone = ( + market_timezone + or default_timezone + or str(scheduler.get("timezone") or "").strip() + ) + return { + "service": str(service).strip(), + "strategy_profile": _first_value( + sources, + ("strategy_profile", "strategy", "profile"), + ), + "account_scope": account_scope, + "scheduler": dict(scheduler), + "market": market, + "market_calendar": market_calendar, + "market_timezone": market_timezone, + } + + +def load_runtime_targets(environ: Mapping[str, str]) -> list[dict[str, Any]]: + raw_targets = str(environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() + items: list[Mapping[str, Any]] = [] + if raw_targets: + try: + payload = json.loads(raw_targets) + except json.JSONDecodeError: + payload = {} + targets = payload.get("targets") if isinstance(payload, dict) else payload + if isinstance(targets, list): + items = [target for target in targets if isinstance(target, dict)] + + if not items: + raw_runtime_target = str(environ.get("RUNTIME_TARGET_JSON") or "").strip() + if raw_runtime_target: + try: + runtime_target = json.loads(raw_runtime_target) + except json.JSONDecodeError: + runtime_target = {} + if isinstance(runtime_target, dict): + items = [runtime_target] + + expected_scope = str(environ.get("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE") or "").strip().lower() + eligible: list[tuple[Mapping[str, Any], dict[str, Any]]] = [] + for item in items: + runtime_target = _runtime_target(item) + enabled_value = item.get("runtime_target_enabled") + if enabled_value is None: + enabled_value = item.get("RUNTIME_TARGET_ENABLED") + if enabled_value is None: + enabled_value = runtime_target.get("runtime_target_enabled") + if not _enabled(enabled_value): + continue + target_scope = _first_value( + [item, runtime_target], + ( + "account_scope", + "account_group", + "account_region", + "ACCOUNT_GROUP", + "ACCOUNT_REGION", + ), + ) + if expected_scope and target_scope and target_scope.lower() != expected_scope: + continue + eligible.append((item, runtime_target)) + + normalized: list[dict[str, Any]] = [] + for item, runtime_target in eligible: + for service in _service_values(item, runtime_target, environ): + target = _normalize_target( + item, + runtime_target, + service, + environ, + use_global_market_fallback=len(eligible) == 1, + ) + key = target_key(target) + if key and all(target_key(existing) != key for existing in normalized): + normalized.append(target) + return normalized + + +def target_key(target: Mapping[str, Any]) -> str: + service = str(target.get("service") or "").strip().lower() + strategy = str(target.get("strategy_profile") or "").strip().lower() + scope = str(target.get("account_scope") or "").strip().lower() + return f"{service}|{strategy or '*'}|{scope or '*'}" if service else "" + + +def target_label(target: Mapping[str, Any]) -> str: + service = str(target.get("service") or "").strip() or "" + strategy = str(target.get("strategy_profile") or "").strip() + scope = str(target.get("account_scope") or "").strip() + qualifiers = "/".join(value for value in (strategy, scope) if value) + return f"{service}[{qualifiers}]" if qualifiers else service + + +def target_latest_due_at(target: Mapping[str, Any]) -> dt.datetime | None: + value = target.get(_LATEST_DUE_AT_KEY) + return value if isinstance(value, dt.datetime) else None + + +def _payload_value(payload: Mapping[str, Any], keys: tuple[str, ...]) -> str: + runtime_target = payload.get("runtime_target") + sources = [payload] + if isinstance(runtime_target, Mapping): + sources.append(runtime_target) + return _first_value(sources, keys) + + +def match_payload_target( + payload: Mapping[str, Any], + targets: list[dict[str, Any]], +) -> tuple[str | None, str]: + service = _payload_value( + payload, + ("service_name", "service", "cloud_run_service"), + ).lower() + strategy = _payload_value( + payload, + ("strategy_profile", "strategy", "profile"), + ).lower() + scope = _payload_value( + payload, + ("account_scope", "account_group", "account_region"), + ).lower() + for target in targets: + expected_service = str(target.get("service") or "").strip().lower() + expected_strategy = str(target.get("strategy_profile") or "").strip().lower() + expected_scope = str(target.get("account_scope") or "").strip().lower() + if service != expected_service: + continue + if expected_strategy and strategy != expected_strategy: + continue + if expected_scope and scope != expected_scope: + continue + return target_key(target), "matched runtime target" + return None, ( + f"runtime_target={service or '-'}/{strategy or '-'}/{scope or '-'}" + ) + + +def _cron_token_value(token: str, *, names: dict[str, int] | None = None) -> int: + normalized = token.strip().lower() + if names and normalized in names: + return names[normalized] + return int(normalized) + + +def _cron_field_values( + field: str, + *, + minimum: int, + maximum: int, + names: dict[str, int] | None = None, +) -> set[int] | None: + text = str(field or "").strip().lower() + if text in {"", "*"}: + return None + values: set[int] = set() + for raw_part in text.split(","): + part = raw_part.strip() + if not part: + continue + base, raw_step = part, "1" + if "/" in part: + base, raw_step = part.split("/", 1) + step = max(1, int(raw_step)) + if base == "*": + start, end = minimum, maximum + elif "-" in base: + raw_start, raw_end = base.split("-", 1) + start = _cron_token_value(raw_start, names=names) + end = _cron_token_value(raw_end, names=names) + else: + start = end = _cron_token_value(base, names=names) + for value in range(start, end + 1, step): + if minimum <= value <= maximum: + values.add(value) + elif maximum == 6 and value == 7: + values.add(0) + return values + + +def cron_matches(schedule: str, value: dt.datetime) -> bool: + fields = str(schedule or "").split() + if len(fields) == 2: + fields.extend(("*", "*", "*")) + if len(fields) != 5: + return False + minute, hour, day_of_month, month, day_of_week = fields + dow_names = { + "sun": 0, + "mon": 1, + "tue": 2, + "wed": 3, + "thu": 4, + "fri": 5, + "sat": 6, + } + minute_values = _cron_field_values(minute, minimum=0, maximum=59) + hour_values = _cron_field_values(hour, minimum=0, maximum=23) + dom_values = _cron_field_values(day_of_month, minimum=1, maximum=31) + month_values = _cron_field_values(month, minimum=1, maximum=12) + dow_values = _cron_field_values(day_of_week, minimum=0, maximum=6, names=dow_names) + if minute_values is not None and value.minute not in minute_values: + return False + if hour_values is not None and value.hour not in hour_values: + return False + if month_values is not None and value.month not in month_values: + return False + dom_matches = dom_values is None or value.day in dom_values + dow_matches = dow_values is None or value.isoweekday() % 7 in dow_values + if dom_values is not None and dow_values is not None: + return dom_matches or dow_matches + return dom_matches and dow_matches + + +def _market_session_dates( + calendar: str, + *, + start_date: dt.date, + end_date: dt.date, +) -> set[dt.date]: + import pandas_market_calendars as mcal + + schedule = mcal.get_calendar(calendar).schedule( + start_date=start_date, + end_date=end_date, + ) + return {value.date() for value in schedule.index} + + +def _target_due_status( + target: Mapping[str, Any], + *, + since: dt.datetime, + now: dt.datetime, + market_aware: bool, + session_dates_loader: SessionDatesLoader, + warning_logger: WarningLogger, +) -> tuple[bool | None, dt.datetime | None]: + scheduler = target.get("scheduler") + if not isinstance(scheduler, Mapping): + return None, None + schedule = str(scheduler.get("main_time") or "").strip() + fields = schedule.split() + if len(fields) == 2: + schedule = f"{schedule} * * *" + elif len(fields) != 5: + return None, None + timezone_name = str(scheduler.get("timezone") or "UTC").strip() or "UTC" + try: + scheduler_timezone = ZoneInfo(timezone_name) + except Exception as exc: # noqa: BLE001 + warning_logger( + f"Unable to evaluate heartbeat scheduler timezone {timezone_name}: " + f"{type(exc).__name__}; keeping target required" + ) + return None, None + + since_utc = since.astimezone(dt.timezone.utc) + now_utc = now.astimezone(dt.timezone.utc) + session_dates: set[dt.date] | None = None + market_calendar = str(target.get("market_calendar") or "").strip() + if market_aware and market_calendar: + market_timezone_name = ( + str(target.get("market_timezone") or "").strip() or timezone_name + ) + try: + market_timezone = ZoneInfo(market_timezone_name) + session_dates = session_dates_loader( + market_calendar, + start_date=since_utc.astimezone(market_timezone).date(), + end_date=now_utc.astimezone(market_timezone).date(), + ) + except Exception as exc: # noqa: BLE001 + warning_logger( + f"Unable to evaluate heartbeat market calendar {market_calendar}: " + f"{type(exc).__name__}; keeping target required" + ) + return None, None + + cursor = since_utc.replace(second=0, microsecond=0) + if cursor < since_utc: + cursor += dt.timedelta(minutes=1) + latest_due_at: dt.datetime | None = None + while cursor <= now_utc: + local_time = cursor.astimezone(scheduler_timezone) + try: + matches = cron_matches(schedule, local_time) + except (TypeError, ValueError) as exc: + warning_logger( + f"Unable to evaluate heartbeat cron for {target_label(target)}: " + f"{type(exc).__name__}; keeping target required" + ) + return None, None + if matches: + if session_dates is None: + latest_due_at = cursor + else: + market_timezone = ZoneInfo( + str(target.get("market_timezone") or "").strip() or timezone_name + ) + if cursor.astimezone(market_timezone).date() in session_dates: + latest_due_at = cursor + cursor += dt.timedelta(minutes=1) + return latest_due_at is not None, latest_due_at + + +def filter_due_targets( + targets: list[dict[str, Any]], + *, + since: dt.datetime, + now: dt.datetime, + market_aware: bool = True, + session_dates_loader: SessionDatesLoader = _market_session_dates, + warning_logger: WarningLogger = lambda message: print(message, file=sys.stderr), +) -> tuple[list[dict[str, Any]], bool]: + due: list[dict[str, Any]] = [] + evaluated = False + for target in targets: + status, latest_due_at = _target_due_status( + target, + since=since, + now=now, + market_aware=market_aware, + session_dates_loader=session_dates_loader, + warning_logger=warning_logger, + ) + if status is not None: + evaluated = True + if status is not False: + due_target = dict(target) + if latest_due_at is not None: + due_target[_LATEST_DUE_AT_KEY] = latest_due_at + due.append(due_target) + return due, evaluated + + +def filter_services_for_targets( + services: list[str], + targets: list[dict[str, Any]], + *, + all_targets: list[dict[str, Any]] | None = None, +) -> list[str]: + if not targets: + return services + target_services = { + str(target.get("service") or "").strip() + for target in targets + if str(target.get("service") or "").strip() + } + configured_services = { + str(target.get("service") or "").strip() + for target in (all_targets or targets) + if str(target.get("service") or "").strip() + } + return [ + service + for service in services + if service not in configured_services or service in target_services + ] diff --git a/tests/test_cloud_run_runtime_guard.py b/tests/test_cloud_run_runtime_guard.py index a96b04b..1889d0a 100644 --- a/tests/test_cloud_run_runtime_guard.py +++ b/tests/test_cloud_run_runtime_guard.py @@ -139,6 +139,54 @@ def test_scheduler_entry_since_uses_matching_service_revision_window(): ) +def test_scheduler_failure_matching_cloud_run_failure_is_duplicate(): + scheduler_entry = { + "timestamp": "2026-07-29T19:45:03Z", + "resource": {"labels": {"job_id": "test-scheduler"}}, + } + cloud_run_failures = { + "test-service": [ + { + "timestamp": "2026-07-29T19:45:01Z", + "resource": {"labels": {"service_name": "test-service"}}, + } + ] + } + + assert guard._is_duplicate_scheduler_failure( + scheduler_entry, + cloud_run_failures, + ) + + +def test_scheduler_failure_for_other_service_is_not_duplicate(): + scheduler_entry = { + "timestamp": "2026-07-29T19:45:03Z", + "resource": {"labels": {"job_id": "other-platform-scheduler"}}, + } + cloud_run_failures = { + "test-service": [ + { + "timestamp": "2026-07-29T19:45:01Z", + "resource": {"labels": {"service_name": "test-service"}}, + } + ] + } + + assert not guard._is_duplicate_scheduler_failure( + scheduler_entry, + cloud_run_failures, + ) + + +def test_services_without_success_are_reported_individually(): + assert guard._services_without_success( + ["healthy-service", "silent-service"], + {"healthy-service": 1, "silent-service": 0}, + {"healthy-service", "silent-service"}, + ) == ["silent-service"] + + def test_monitor_dispatch_capacity_warning_is_not_failure_by_default(monkeypatch): monkeypatch.delenv("RUNTIME_GUARD_IGNORE_MONITOR_DISPATCH_CAPACITY_WARNINGS", raising=False) entry = { diff --git a/tests/test_execution_report_heartbeat.py b/tests/test_execution_report_heartbeat.py index e895317..66b30b8 100644 --- a/tests/test_execution_report_heartbeat.py +++ b/tests/test_execution_report_heartbeat.py @@ -1,10 +1,33 @@ from __future__ import annotations import datetime as dt +import json from scripts import execution_report_heartbeat as heartbeat +def test_required_services_skip_disabled_runtime_targets(monkeypatch): + for name in ( + "RUNTIME_HEARTBEAT_REQUIRED_SERVICES", + "CLOUD_RUN_SERVICES", + "CLOUD_RUN_SERVICE", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "targets": [ + {"service": "schwab-enabled-service", "RUNTIME_TARGET_ENABLED": "true"}, + {"service": "schwab-disabled-service", "RUNTIME_TARGET_ENABLED": "false"}, + ] + } + ), + ) + + assert heartbeat._load_required_services() == ["schwab-enabled-service"] + + def test_heartbeat_skips_when_runtime_target_is_disabled(monkeypatch, capsys): monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "CharlesSchwab disabled runtime") monkeypatch.setenv("RUNTIME_TARGET_ENABLED", "false") @@ -87,3 +110,22 @@ def test_heartbeat_does_not_skip_when_lookback_includes_scheduler_day(monkeypatc ) assert reason is None + + +def test_report_with_failed_notification_delivery_is_rejected(): + accepted, reason = heartbeat._is_accepted_report( + { + "status": "ok", + "summary": { + "notification_delivery_summary": { + "event_count": 1, + "sent_count": 0, + "failed_count": 1, + "all_acknowledged": False, + } + }, + } + ) + + assert accepted is False + assert "notification delivery not acknowledged" in reason diff --git a/tests/test_notifications.py b/tests/test_notifications.py index cd741cb..c8f0be7 100644 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -20,12 +20,23 @@ class FakeRequests: - def __init__(self): + def __init__(self, *, status_code=200, payload=None): self.calls = [] + self.status_code = status_code + self.payload = {"ok": True} if payload is None else payload def post(self, url, json, timeout): self.calls.append((url, json, timeout)) - return object() + return FakeResponse(self.status_code, self.payload) + + +class FakeResponse: + def __init__(self, status_code, payload): + self.status_code = status_code + self._payload = payload + + def json(self): + return self._payload class NotificationTests(unittest.TestCase): @@ -341,7 +352,7 @@ def test_build_signal_text_formats_icon_and_label(self): def test_build_sender_posts_to_telegram(self): fake_requests = FakeRequests() sender = build_sender("token-1", "chat-1", requests_module=fake_requests) - sender("SOXL.US and 00700.HK") + self.assertTrue(sender("SOXL.US and 00700.HK")) self.assertEqual(len(fake_requests.calls), 1) url, payload, timeout = fake_requests.calls[0] self.assertIn("token-1", url) @@ -349,6 +360,12 @@ def test_build_sender_posts_to_telegram(self): self.assertEqual(payload["text"], "SOXL.\u2060US and 00700.\u2060HK") self.assertEqual(timeout, 15) + def test_build_sender_returns_false_when_telegram_rejects_message(self): + fake_requests = FakeRequests(payload={"ok": False, "description": "chat not found"}) + sender = build_sender("token-1", "chat-1", requests_module=fake_requests) + + self.assertIs(sender("rebalance"), False) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 1700453..a0bc04f 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -21,6 +21,26 @@ class RebalanceServiceTests(unittest.TestCase): + def test_notification_delivery_summary_keeps_failed_transport_receipt(self): + payload = rebalance_service._build_notification_delivery_summary( + [ + { + "sink": "telegram", + "delivery_status": "failed", + "transport_acknowledged": False, + "error_type": "RuntimeError", + "compact_text_sha256": "a" * 64, + "compact_text_length": 42, + } + ] + ) + + self.assertEqual(payload["attempted_count"], 1) + self.assertEqual(payload["sent_count"], 0) + self.assertEqual(payload["failed_count"], 1) + self.assertFalse(payload["all_acknowledged"]) + self.assertNotIn("compact_text", payload["delivery_events"][0]) + def test_sell_uses_position_quantity_when_market_value_is_stale_below_quote(self): submitted_orders = [] plan = { diff --git a/tests/test_runtime_composer.py b/tests/test_runtime_composer.py index 609c440..c8837a8 100644 --- a/tests/test_runtime_composer.py +++ b/tests/test_runtime_composer.py @@ -80,15 +80,19 @@ def fake_reporting_builder(**kwargs): }.get(name, default), sleeper=lambda _seconds: None, printer=lambda *_args, **_kwargs: None, - sender_builder=lambda token, chat_id: lambda message: observed.setdefault( - "sent_message", - (token, chat_id, message), - ), + sender_builder=lambda token, chat_id: lambda message: ( + observed.setdefault( + "sent_message", + (token, chat_id, message), + ), + False, + )[1], notification_builder=fake_notification_builder, reporting_builder=fake_reporting_builder, ) - composer.send_tg_message("hello") + sent = composer.send_tg_message("hello") + assert sent is False runtime = composer.build_rebalance_runtime("client") config = composer.build_rebalance_config( strategy_plugin_signals=("plugin-line",), diff --git a/tests/test_runtime_heartbeat_policy.py b/tests/test_runtime_heartbeat_policy.py new file mode 100644 index 0000000..258bac6 --- /dev/null +++ b/tests/test_runtime_heartbeat_policy.py @@ -0,0 +1,358 @@ +from __future__ import annotations + +import datetime as dt +import json + +from scripts.runtime_heartbeat_policy import ( + filter_due_targets, + load_runtime_targets, + match_payload_target, + target_key, + target_latest_due_at, +) +from scripts import execution_report_heartbeat as heartbeat + + +def _target( + *, + service: str, + strategy: str, + scope: str, + timezone: str, + calendar: str, +) -> dict[str, object]: + return { + "service": service, + "runtime_target": { + "service_name": service, + "strategy_profile": strategy, + "account_scope": scope, + "scheduler": { + "timezone": timezone, + "main_time": "45 15 * * *", + }, + "market_calendar": calendar, + "market_timezone": timezone, + }, + } + + +def test_due_targets_use_each_strategy_market_calendar() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="svc-hk", + strategy="hk-strategy", + scope="HK", + timezone="Asia/Hong_Kong", + calendar="XHKG", + ), + ] + } + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 3, 0, 0, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 3, 22, 0, tzinfo=dt.timezone.utc), + session_dates_loader=lambda calendar, **_kwargs: ( + {dt.date(2026, 7, 3)} if calendar == "XHKG" else set() + ), + ) + + assert evaluated is True + assert [target["strategy_profile"] for target in due] == ["hk-strategy"] + + +def test_real_exchange_calendars_distinguish_us_holiday_from_hk_session() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="svc-hk", + strategy="hk-strategy", + scope="HK", + timezone="Asia/Hong_Kong", + calendar="XHKG", + ), + ] + } + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 3, 0, 0, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 3, 22, 0, tzinfo=dt.timezone.utc), + ) + + assert evaluated is True + assert [target["strategy_profile"] for target in due] == ["hk-strategy"] + + +def test_july_29_us_month_end_target_is_due_at_1545_eastern() -> None: + raw_target = _target( + service="svc-us-monthly", + strategy="us-monthly", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ) + raw_target["runtime_target"]["scheduler"]["main_time"] = "45 15 25-29 * *" + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + {"targets": [raw_target]} + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 29, 19, 40, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 29, 20, 10, tzinfo=dt.timezone.utc), + ) + + assert evaluated is True + assert [target["strategy_profile"] for target in due] == ["us-monthly"] + assert target_latest_due_at(due[0]) == dt.datetime( + 2026, + 7, + 29, + 19, + 45, + tzinfo=dt.timezone.utc, + ) + + +def test_neutral_daily_heartbeat_tracks_latest_due_time_per_market() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="svc-hk", + strategy="hk-strategy", + scope="HK", + timezone="Asia/Hong_Kong", + calendar="XHKG", + ), + ] + } + ) + } + ) + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 28, 10, 20, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 29, 22, 20, tzinfo=dt.timezone.utc), + session_dates_loader=lambda _calendar, **_kwargs: { + dt.date(2026, 7, 28), + dt.date(2026, 7, 29), + }, + ) + + assert evaluated is True + assert { + target["strategy_profile"]: target_latest_due_at(target) + for target in due + } == { + "us-strategy": dt.datetime( + 2026, + 7, + 29, + 19, + 45, + tzinfo=dt.timezone.utc, + ), + "hk-strategy": dt.datetime( + 2026, + 7, + 29, + 7, + 45, + tzinfo=dt.timezone.utc, + ), + } + + +def test_same_service_strategies_require_distinct_reports() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + _target( + service="shared-service", + strategy="strategy-a", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + _target( + service="shared-service", + strategy="strategy-b", + scope="US", + timezone="America/New_York", + calendar="NYSE", + ), + ] + } + ) + } + ) + + matched, reason = match_payload_target( + { + "service_name": "shared-service", + "strategy_profile": "strategy-a", + "account_scope": "US", + }, + targets, + ) + + assert matched == target_key(targets[0]) + assert reason == "matched runtime target" + missing_strategy, _ = match_payload_target( + {"service_name": "shared-service", "account_scope": "US"}, + targets, + ) + assert missing_strategy is None + + matched_by_heartbeat, matched_key, _ = heartbeat._payload_matches( + { + "service_name": "shared-service", + "strategy_profile": "strategy-a", + "account_scope": "US", + }, + ["shared-service"], + required_targets=targets, + ) + assert matched_by_heartbeat is True + assert matched_key == target_key(targets[0]) + missing_by_heartbeat, _, _ = heartbeat._payload_matches( + {"service_name": "shared-service", "account_scope": "US"}, + ["shared-service"], + required_targets=targets, + ) + assert missing_by_heartbeat is False + + +def test_scheduler_timezone_beats_account_region_when_market_is_not_explicit() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + { + "service": "longbridge-sg-us-service", + "account_scope": "SG", + "runtime_target": { + "service_name": "longbridge-sg-us-service", + "strategy_profile": "us-strategy", + "account_scope": "SG", + "scheduler": { + "timezone": "America/New_York", + "main_time": "45 15 * * *", + }, + }, + } + ] + } + ) + } + ) + + assert targets[0]["market"] == "US" + assert targets[0]["market_calendar"] == "NYSE" + assert targets[0]["market_timezone"] == "America/New_York" + + +def test_ambiguous_sg_account_does_not_guess_a_stock_exchange() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "targets": [ + { + "service": "longbridge-sg-service", + "account_scope": "SG", + "runtime_target": { + "service_name": "longbridge-sg-service", + "strategy_profile": "unknown-strategy", + "account_scope": "SG", + "scheduler": { + "timezone": "UTC", + "main_time": "0 12 * * *", + }, + }, + } + ] + } + ) + } + ) + + assert targets[0]["market"] == "" + assert targets[0]["market_calendar"] == "" + + +def test_calendar_failure_keeps_target_due_fail_closed() -> None: + targets = load_runtime_targets( + { + "RUNTIME_TARGET_JSON": json.dumps( + _target( + service="svc-us", + strategy="us-strategy", + scope="US", + timezone="America/New_York", + calendar="INVALID", + )["runtime_target"] + ) + } + ) + + def fail_calendar(_calendar: str, **_kwargs: object) -> set[dt.date]: + raise RuntimeError("calendar unavailable") + + due, evaluated = filter_due_targets( + targets, + since=dt.datetime(2026, 7, 3, 0, 0, tzinfo=dt.timezone.utc), + now=dt.datetime(2026, 7, 3, 22, 0, tzinfo=dt.timezone.utc), + session_dates_loader=fail_calendar, + warning_logger=lambda _message: None, + ) + + assert due == targets + assert evaluated is False diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py new file mode 100644 index 0000000..a2a65a8 --- /dev/null +++ b/tests/test_runtime_monitor_workflows.py @@ -0,0 +1,23 @@ +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_execution_report_heartbeat_has_market_neutral_daily_schedule() -> None: + workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text() + + assert 'cron: "20 22 * * *"' in workflow + assert 'cron: "20 22 * * 1-5"' not in workflow + assert "RUNTIME_HEARTBEAT_MARKET_AWARE:" in workflow + assert "pandas-market-calendars==5.4.0" in workflow + + +def test_runtime_monitor_workflows_retry_gcp_authentication() -> None: + for name in ("execution-report-heartbeat.yml", "runtime-guard.yml"): + workflow = (ROOT / ".github/workflows" / name).read_text() + + assert workflow.count("google-github-actions/auth@v3") == 2 + assert "id: gcp_auth_primary" in workflow + assert "continue-on-error: true" in workflow + assert "steps.gcp_auth_primary.outcome == 'failure'" in workflow diff --git a/tests/test_runtime_notification_adapters.py b/tests/test_runtime_notification_adapters.py new file mode 100644 index 0000000..1c34b04 --- /dev/null +++ b/tests/test_runtime_notification_adapters.py @@ -0,0 +1,42 @@ +from application.runtime_notification_adapters import build_runtime_notification_adapters + + +def test_runtime_notification_adapter_records_negative_ack_as_failed(): + events = [] + adapters = build_runtime_notification_adapters( + send_message=lambda _message: False, + delivery_events=events, + log_message=lambda _message: None, + ) + + sent = adapters.publish_cycle_notification( + detailed_text="details", + compact_text="rebalance", + ) + + assert sent is False + assert events[0]["delivery_status"] == "failed" + assert events[0]["transport_acknowledged"] is False + assert "compact_text" not in events[0] + + +def test_runtime_notification_adapter_records_sender_exception_without_raising(): + events = [] + + def fail(_message): + raise RuntimeError("transport unavailable") + + adapters = build_runtime_notification_adapters( + send_message=fail, + delivery_events=events, + log_message=lambda _message: None, + ) + + sent = adapters.publish_cycle_notification( + detailed_text="details", + compact_text="rebalance", + ) + + assert sent is False + assert events[0]["delivery_status"] == "failed" + assert events[0]["error_type"] == "RuntimeError" diff --git a/tests/test_runtime_report_summary.py b/tests/test_runtime_report_summary.py index fb2af6a..d700d11 100644 --- a/tests/test_runtime_report_summary.py +++ b/tests/test_runtime_report_summary.py @@ -59,3 +59,21 @@ def test_summarize_execution_cycle_result_handles_empty_result() -> None: assert summary["submitted_order_side_counts"] == {} assert summary["submitted_order_type_counts"] == {} assert summary["no_op_reason"] == "market_closed" + + +def test_summarize_execution_cycle_result_preserves_notification_delivery_summary() -> None: + delivery_summary = { + "event_count": 1, + "sent_count": 0, + "failed_count": 1, + "all_acknowledged": False, + } + result = SimpleNamespace( + execution={"notification_delivery_summary": delivery_summary}, + submitted_orders=(), + trade_logs=(), + ) + + summary = summarize_execution_cycle_result(result, dry_run=False) + + assert summary["notification_delivery_summary"] == delivery_summary