diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml index c173e30..a8826e8 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: interactivebrokersquant @@ -44,6 +45,10 @@ 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.IBKR_MARKET_CALENDAR }} + RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.IBKR_MARKET_TIMEZONE }} + RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES: ${{ vars.RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES || '30' }} RUNTIME_HEARTBEAT_SCHEDULER_AWARE: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_AWARE || 'true' }} RUNTIME_HEARTBEAT_SCHEDULER_LOCATION: ${{ vars.RUNTIME_HEARTBEAT_SCHEDULER_LOCATION || vars.CLOUD_RUN_REGION || 'us-central1' }} RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }} @@ -51,6 +56,7 @@ jobs: 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 }} + CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }} GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }} TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }} @@ -59,6 +65,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 }} @@ -67,5 +86,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 9702a1e..1618bb1 100644 --- a/.github/workflows/runtime-guard.yml +++ b/.github/workflows/runtime-guard.yml @@ -63,6 +63,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/runtime_notification_adapters.py b/application/runtime_notification_adapters.py index fcd58e7..be270e0 100644 --- a/application/runtime_notification_adapters.py +++ b/application/runtime_notification_adapters.py @@ -17,13 +17,18 @@ class IBKRNotificationAdapters: 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( @@ -35,17 +40,34 @@ def build_runtime_notification_adapters( ) -> IBKRNotificationAdapters: recorded_delivery_events = delivery_events if delivery_events is not None else [] - def send_recorded_message(message: str) -> None: - send_message(message) + def send_recorded_message(message: str) -> bool: compact = str(message or "") - recorded_delivery_events.append( + 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( { - "sink": notification_channel, - "delivery_status": "sent", - "compact_text_sha256": hashlib.sha256(compact.encode("utf-8")).hexdigest(), - "compact_text_length": len(compact), + "delivery_status": "sent" if acknowledged else "failed", + "transport_acknowledged": acknowledged, } ) + recorded_delivery_events.append(event) + return acknowledged return IBKRNotificationAdapters( notification_port=CallableNotificationPort(send_recorded_message), diff --git a/entrypoints/cloud_run.py b/entrypoints/cloud_run.py index ae1a39b..93d1401 100644 --- a/entrypoints/cloud_run.py +++ b/entrypoints/cloud_run.py @@ -15,7 +15,7 @@ def _load_market_calendar(calendar_name: str, *, logger) -> object | None: except Exception as exc: logger( f"pandas_market_calendars unavailable for {calendar_name}: {exc}; " - "falling back to weekday-only market-open check" + "failing closed as market closed" ) return None @@ -31,7 +31,7 @@ def is_market_open_now( now_ny = datetime.now(tz_ny) calendar = _load_market_calendar(calendar_name, logger=logger) if calendar is None: - return now_ny.weekday() < 5 + return False schedule = calendar.schedule(start_date=now_ny.date(), end_date=now_ny.date()) if len(getattr(schedule, "index", ())) == 0: return False @@ -54,6 +54,6 @@ def is_market_open_today( now_ny = datetime.now(tz_ny) calendar = _load_market_calendar(calendar_name, logger=logger) if calendar is None: - return now_ny.weekday() < 5 + return False schedule = calendar.schedule(start_date=now_ny.date(), end_date=now_ny.date()) return len(getattr(schedule, "index", ())) > 0 diff --git a/main.py b/main.py index a9d011d..7c45791 100644 --- a/main.py +++ b/main.py @@ -701,7 +701,7 @@ def send_tg_message(message): telegram_chat_id=TG_CHAT_ID, webhook_url=_NOTIFICATION_WEBHOOK_URL, ) - sender(_with_platform_notification_prefix(message)) + return sender(_with_platform_notification_prefix(message)) def _platform_notification_prefix() -> str: @@ -739,7 +739,7 @@ def _with_platform_notification_prefix(message: str) -> str: 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, ) @@ -796,20 +796,22 @@ def _notify_runtime_error(exc: Exception, *, route_label: str | None = None) -> print("IBKR 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: - send_telegram_message( + outcomes.append(send_telegram_message( _with_platform_notification_prefix(message), token=token, chat_id=chat_id, requests_module=requests, - ) - return True + )) + 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) except Exception as notification_exc: print(f"IBKR runtime error notification fallback: {notification_exc}", flush=True) return _notify_runtime_error(exc) @@ -1025,6 +1027,32 @@ def _build_notification_delivery_log_for_report( } +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 publish_strategy_plugin_alerts(signals, *, report=None): result = dispatch_strategy_plugin_alerts( signals, @@ -1329,6 +1357,11 @@ def _handle_request( ) if notification_delivery_log: report_summary["notification_delivery_log"] = notification_delivery_log + notification_delivery_summary = _build_notification_delivery_summary( + notification_delivery_events + ) + if notification_delivery_summary: + report_summary["notification_delivery_summary"] = notification_delivery_summary execution_status = str(report_summary.get("execution_status") or "").strip().lower() execution_failed = _is_execution_failure( execution_status, diff --git a/notifications/telegram.py b/notifications/telegram.py index e489d6e..b22c308 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -523,7 +523,7 @@ def send_telegram_message( printer=print, ): if not token or not chat_id: - return + return False url = f"https://api.telegram.org/bot{token}/sendMessage" try: @@ -539,5 +539,13 @@ def send_telegram_message( f"{response.status_code}: {_safe_telegram_error_text(response.text, token=token)}", 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: + printer("Telegram send failed: negative API acknowledgement", flush=True) + return False except Exception as exc: printer(f"Telegram send failed: {type(exc).__name__}", flush=True) + return False + return True diff --git a/runtime_config_support.py b/runtime_config_support.py index 67787ba..7fb8281 100644 --- a/runtime_config_support.py +++ b/runtime_config_support.py @@ -257,6 +257,21 @@ class PlatformRuntimeSettings: execution_backend: str = EXECUTION_BACKEND_GATEWAY +def _runtime_target_market_value(runtime_target: RuntimeTarget, field: str) -> str | None: + value = getattr(runtime_target, field, None) + if value is not None and str(value).strip(): + return str(value).strip() + raw = os.getenv("QSL_RUNTIME_TARGET_JSON") or os.getenv("RUNTIME_TARGET_JSON") + if not raw: + return None + try: + payload = json.loads(raw) + except json.JSONDecodeError: + return None + value = payload.get(field) if isinstance(payload, dict) else None + return str(value).strip() if value is not None and str(value).strip() else None + + def load_platform_runtime_settings( *, project_id_resolver: Callable[[], str | None], @@ -336,7 +351,13 @@ def load_platform_runtime_settings( ) ib_client_id = group_config.ib_client_id or 0 - market = resolve_market(os.getenv("IBKR_MARKET"), account_group=account_group) + market = resolve_market( + first_non_empty( + os.getenv("IBKR_MARKET"), + _runtime_target_market_value(runtime_target, "market"), + ), + account_group=account_group, + ) market_defaults = market_default_settings(market) return PlatformRuntimeSettings( project_id=project_id, @@ -386,6 +407,7 @@ def load_platform_runtime_settings( market=market, market_calendar=first_non_empty( os.getenv("IBKR_MARKET_CALENDAR"), + _runtime_target_market_value(runtime_target, "market_calendar"), market_defaults["market_calendar"], ), market_currency=first_non_empty( @@ -404,6 +426,7 @@ def load_platform_runtime_settings( ).upper(), market_timezone=first_non_empty( os.getenv("IBKR_MARKET_TIMEZONE"), + _runtime_target_market_value(runtime_target, "market_timezone"), market_defaults["market_timezone"], ), quantity_step=1.0, diff --git a/scripts/cloud_run_runtime_guard.py b/scripts/cloud_run_runtime_guard.py index 7a1768c..7bc893a 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]: @@ -40,6 +41,8 @@ def _env_bool(name: str, default: bool = False) -> bool: def _load_services() -> list[str]: services = [] + enabled_target_services = [] + disabled_target_services = [] for name in ( "RUNTIME_GUARD_CLOUD_RUN_SERVICES", "CLOUD_RUN_SERVICES", @@ -51,37 +54,27 @@ def _load_services() -> list[str]: if raw_targets: try: payload = json.loads(raw_targets) + defaults = payload.get("defaults") if isinstance(payload, dict) else {} + defaults = defaults if isinstance(defaults, dict) else {} targets = payload.get("targets") if isinstance(payload, dict) else payload if isinstance(targets, list): for target in targets: if not isinstance(target, dict): continue - if not _target_enabled(target): - continue - runtime_target = target.get("runtime_target") or target.get( - "runtime_target_json" - ) - if isinstance(runtime_target, str): - try: - runtime_target = json.loads(runtime_target) - except json.JSONDecodeError: - runtime_target = {} - 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))) - break + target_services = _target_service_names(target, defaults) + if _target_enabled(target, defaults): + enabled_target_services.extend(target_services) + else: + disabled_target_services.extend(target_services) except json.JSONDecodeError as exc: raise RuntimeError(f"CLOUD_RUN_SERVICE_TARGETS_JSON is invalid: {exc}") from exc + 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 @@ -111,9 +104,29 @@ def _service_job_aliases(service: str) -> list[str]: def _scheduler_job_pattern_for_services(services: list[str]) -> str: candidates: list[str] = [] for service in services: - candidates.extend(_service_job_aliases(service)) + candidates.extend(_scheduler_job_names(service)) unique = list(dict.fromkeys(candidates)) - return "|".join(re.escape(candidate) for candidate in unique) + if not unique: + return "" + return r"^(?:" + "|".join(re.escape(candidate) for candidate in unique) + r")\Z" + + +def _scheduler_job_names(service: str) -> list[str]: + names = [] + for alias in _service_job_aliases(service): + names.extend( + ( + f"{alias}-scheduler", + f"{alias}-probe-scheduler", + f"{alias}-precheck-scheduler", + ) + ) + return list(dict.fromkeys(names)) + + +def _job_matches_service(job_name: str, service: str) -> bool: + normalized = str(job_name or "").strip().rsplit("/", 1)[-1] + return normalized in _scheduler_job_names(service) def _entry_job_name(entry: dict[str, Any]) -> str: @@ -130,11 +143,46 @@ def _scheduler_entry_since( matches = [ service_since for service, service_since in service_since_by_name.items() - if any(alias and alias in job_name for alias in _service_job_aliases(service)) + if _job_matches_service(job_name, service) ] 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 _job_matches_service(job_name, 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) @@ -197,22 +245,51 @@ def _format_timestamp(value: dt.datetime) -> str: return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") -def _target_payloads() -> list[dict[str, Any]]: +def _target_configuration() -> tuple[list[dict[str, Any]], dict[str, Any]]: raw_targets = (os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() if not raw_targets: - return [] + return [], {} try: payload = json.loads(raw_targets) except json.JSONDecodeError: - return [] + return [], {} + defaults = payload.get("defaults") if isinstance(payload, dict) else {} + defaults = defaults if isinstance(defaults, dict) else {} targets = payload.get("targets") if isinstance(payload, dict) else payload if not isinstance(targets, list): - return [] - return [target for target in targets if isinstance(target, dict)] + return [], defaults + return [target for target in targets if isinstance(target, dict)], defaults -def _runtime_target(target: dict[str, Any]) -> dict[str, Any]: - runtime_target = target.get("runtime_target") or target.get("runtime_target_json") +def _target_payloads() -> list[dict[str, Any]]: + targets, _defaults = _target_configuration() + return targets + + +def _target_field( + target: dict[str, Any], + defaults: dict[str, Any], + *names: str, +) -> Any: + target_env = target.get("env") if isinstance(target.get("env"), dict) else {} + defaults_env = defaults.get("env") if isinstance(defaults.get("env"), dict) else {} + for source in (target, target_env, defaults, defaults_env): + for name in names: + if name in source: + return source[name] + return None + + +def _runtime_target( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> dict[str, Any]: + runtime_target = _target_field( + target, + defaults or {}, + "runtime_target", + "runtime_target_json", + ) if isinstance(runtime_target, str): try: runtime_target = json.loads(runtime_target) @@ -232,32 +309,57 @@ def _coerce_bool(value: Any, default: bool) -> bool: return text in {"1", "true", "yes", "y", "on"} -def _target_enabled(target: dict[str, Any]) -> bool: - runtime_target = _runtime_target(target) +def _target_enabled( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> bool: + defaults = defaults or {} + runtime_target = _runtime_target(target, defaults) + value = _target_field( + target, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) + if value is not None: + return _coerce_bool(value, True) for key in ("runtime_target_enabled", "RUNTIME_TARGET_ENABLED"): - if key in target: - return _coerce_bool(target.get(key), True) if key in runtime_target: return _coerce_bool(runtime_target.get(key), True) return True -def _target_service_names(target: dict[str, Any]) -> list[str]: - runtime_target = _runtime_target(target) - for key in ("service", "service_name", "cloud_run_service"): - value = target.get(key) or runtime_target.get(key) - if value: - return _split_values(str(value)) +def _target_service_names( + target: dict[str, Any], + defaults: dict[str, Any] | None = None, +) -> list[str]: + defaults = defaults or {} + runtime_target = _runtime_target(target, defaults) + value = _target_field( + target, + defaults, + "service", + "service_name", + "cloud_run_service", + ) + if value is None: + for key in ("service", "service_name", "cloud_run_service"): + if runtime_target.get(key): + value = runtime_target[key] + break + if value: + return _split_values(str(value)) return [] def _region_for_service(service: str) -> str: - for target in _target_payloads(): - if service not in _target_service_names(target): + targets, defaults = _target_configuration() + for target in targets: + if service not in _target_service_names(target, defaults): continue - runtime_target = _runtime_target(target) + runtime_target = _runtime_target(target, defaults) for key in ("region", "cloud_run_region", "location"): - value = target.get(key) or runtime_target.get(key) + value = _target_field(target, defaults, key) or runtime_target.get(key) if value: return str(value).strip() return ( @@ -505,6 +607,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: @@ -527,16 +632,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}"' @@ -547,7 +662,7 @@ def main() -> int: entries = [ entry for entry in entries - if regex.search(str(_labels(entry).get("job_id") or _labels(entry).get("job_name") or "")) + if regex.search(_entry_job_name(entry).rsplit("/", 1)[-1]) ] failures = [] for entry in entries: @@ -557,6 +672,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 fe6182b..d467e4f 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -14,6 +14,31 @@ 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, + runtime_target_configuration_has_enabled_targets, + runtime_target_configuration_present, + 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, + runtime_target_configuration_has_enabled_targets, + runtime_target_configuration_present, + 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"} @@ -313,30 +338,17 @@ def _base_report_uris() -> list[str]: def _load_target_service_candidates() -> tuple[list[str], list[str]]: - disabled_target_services: list[str] = [] - enabled_target_services: list[str] = [] - raw_targets = (os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() - if not raw_targets: - return enabled_target_services, disabled_target_services - try: - payload = json.loads(raw_targets) - targets = payload.get("targets") if isinstance(payload, dict) else payload - if isinstance(targets, list): - for target in targets: - if not isinstance(target, dict): - continue - runtime_target = _target_runtime_target(target) - if not _target_matches_expected_scope(target, runtime_target): - continue - target_services = _target_service_values(target, runtime_target) - if not target_services: - continue - if _target_enabled(target, runtime_target): - enabled_target_services.extend(target_services) - else: - disabled_target_services.extend(target_services) - except json.JSONDecodeError: - pass + targets = load_runtime_targets(os.environ, include_disabled=True) + enabled_target_services = [ + str(target.get("service") or "").strip() + for target in targets + if target.get("enabled") is True + ] + disabled_target_services = [ + str(target.get("service") or "").strip() + for target in targets + if target.get("enabled") is False + ] return _unique_values(enabled_target_services), _unique_values(disabled_target_services) @@ -528,6 +540,41 @@ def _scheduler_job_name_candidates(service: str) -> list[str]: return _unique_values(candidates) +def _hydrate_runtime_target_schedules( + targets: list[dict[str, Any]], + *, + project: str | None, +) -> list[dict[str, Any]]: + hydrated: list[dict[str, Any]] = [] + for target in targets: + scheduler = target.get("scheduler") + effective_scheduler = dict(scheduler) if isinstance(scheduler, dict) else {} + if len(str(effective_scheduler.get("main_time") or "").split()) == 5: + hydrated.append(target) + continue + if not _env_bool("RUNTIME_HEARTBEAT_SCHEDULER_AWARE", True): + raise RuntimeError( + "runtime target schedule is incomplete and scheduler lookup is disabled" + ) + service = str(target.get("service") or "").strip() + job = None + for job_name in _scheduler_job_name_candidates(service): + job = _describe_scheduler_job(job_name, project=project) + if job is not None: + break + schedule = str((job or {}).get("schedule") or "").strip() + if len(schedule.split()) != 5: + raise RuntimeError( + f"unable to resolve effective five-field scheduler cron for {service or ''}" + ) + timezone_name = str((job or {}).get("timeZone") or "").strip() + effective_scheduler["main_time"] = schedule + if timezone_name: + effective_scheduler["timezone"] = timezone_name + hydrated.append({**target, "scheduler": effective_scheduler}) + return hydrated + + def _scheduler_job_targets_strategy_run(job: dict[str, Any], service: str) -> bool: if str(job.get("state") or "").strip().upper() not in {"", "ENABLED"}: return False @@ -755,6 +802,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 _report_execution_status(payload: dict[str, Any]) -> str: summary = payload.get("summary") nested_status = summary.get("execution_status") if isinstance(summary, dict) else None @@ -789,7 +850,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() @@ -805,6 +871,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" @@ -834,8 +910,11 @@ def _is_accepted_report(payload: dict[str, Any]) -> tuple[bool, str]: execution_status_key = execution_status.lower() no_op_reason = _report_no_op_reason(payload) 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 execution_status_key in DEFAULT_REJECT_EXECUTION_STATUSES or ( @@ -926,6 +1005,15 @@ def main(now: dt.datetime | None = None) -> int: if not _runtime_target_enabled(): print(f"Execution report heartbeat skipped for {name}: runtime target is disabled") return 0 + if ( + runtime_target_configuration_present(os.environ) + and not runtime_target_configuration_has_enabled_targets(os.environ) + ): + print( + f"Execution report heartbeat skipped for {name}: " + "no enabled runtime target matches this heartbeat" + ) + return 0 lookback_hours = float(os.environ.get("RUNTIME_HEARTBEAT_LOOKBACK_HOURS") or "36") max_reports = int(os.environ.get("RUNTIME_HEARTBEAT_MAX_REPORTS_TO_READ") or "20") fail_workflow = _env_bool("RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT", True) @@ -935,10 +1023,46 @@ 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) - runtime_target_skip_reason = _runtime_target_scheduler_skip_reason(since, now) - if runtime_target_skip_reason: - print(f"Execution report heartbeat skipped for {name}: {runtime_target_skip_reason}") + runtime_targets = load_runtime_targets(os.environ) + try: + runtime_targets = _hydrate_runtime_target_schedules( + runtime_targets, + project=project, + ) + except RuntimeError as exc: + message = f"[Execution Report Heartbeat] {name}\nScheduler policy error: {exc}" + print(message) + _send_telegram(message) + return 1 if fail_workflow else 0 + publication_grace_minutes = float( + os.environ.get("RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES") or "30" + ) + if publication_grace_minutes < 0: + raise ValueError("RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES must be non-negative") + due_targets, target_schedule_evaluated = filter_due_targets( + runtime_targets, + since=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 = _runtime_target_scheduler_skip_reason(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: + runtime_target_skip_reason = _runtime_target_scheduler_skip_reason(since, now) + if runtime_target_skip_reason: + print(f"Execution report heartbeat skipped for {name}: {runtime_target_skip_reason}") + return 0 required_services, scheduler_skip_reason, _scheduler_checked = _resolve_required_services( project=project, @@ -948,6 +1072,37 @@ def main(now: dt.datetime | None = None) -> int: if scheduler_skip_reason: print(f"Execution report heartbeat skipped for {name}: {scheduler_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: raise SystemExit("No heartbeat GCS report URI configured") @@ -973,23 +1128,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 @@ -1005,9 +1172,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..4baf3cb --- /dev/null +++ b/scripts/runtime_heartbeat_policy.py @@ -0,0 +1,677 @@ +"""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] +ProfileResolver = Callable[[str], str] + +_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 _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _target_field( + item: Mapping[str, Any], + defaults: Mapping[str, Any], + *names: str, +) -> Any: + for source in ( + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ): + for name in names: + if name in source: + return source[name] + return None + + +def _runtime_target( + item: Mapping[str, Any], + defaults: Mapping[str, Any], +) -> dict[str, Any]: + value = _target_field(item, defaults, "runtime_target", "runtime_target_json") + if isinstance(value, str): + try: + value = json.loads(value) + except json.JSONDecodeError as exc: + raise ValueError(f"runtime target JSON is invalid: {exc}") from exc + if isinstance(value, Mapping): + return dict(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], + defaults: Mapping[str, Any], + environ: Mapping[str, str], +) -> list[str]: + value = _target_field( + item, + defaults, + "service", + "service_name", + "cloud_run_service", + ) + if value is None: + value = _first_value( + [runtime_target], + ("service", "service_name", "cloud_run_service"), + ) + if value: + return _split_values(str(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], + defaults: Mapping[str, Any], + service: str, + environ: Mapping[str, str], + *, + use_global_market_fallback: bool, + profile_resolver: ProfileResolver | None, +) -> dict[str, Any]: + sources = [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ] + scheduler = next( + ( + value + for source in sources + if isinstance((value := source.get("scheduler")), dict) + ), + {}, + ) + scheduler = dict(scheduler) + if not str(scheduler.get("main_time") or "").strip(): + main_time = _target_field( + item, + defaults, + "CLOUD_SCHEDULER_MAIN_TIME", + "cloud_scheduler_main_time", + ) + if main_time is None: + main_time = environ.get("CLOUD_SCHEDULER_MAIN_TIME") + if main_time is not None and str(main_time).strip(): + scheduler["main_time"] = str(main_time).strip() + 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() + ) + if scheduler and not str(scheduler.get("timezone") or "").strip() and market_timezone: + scheduler["timezone"] = market_timezone + strategy_profile = _first_value( + sources, + ("strategy_profile", "strategy", "profile"), + ) + if strategy_profile and profile_resolver is not None: + strategy_profile = profile_resolver(strategy_profile) + return { + "service": str(service).strip(), + "strategy_profile": strategy_profile, + "account_scope": account_scope, + "scheduler": scheduler, + "market": market, + "market_calendar": market_calendar, + "market_timezone": market_timezone, + } + + +def _load_runtime_target_items( + environ: Mapping[str, str], +) -> tuple[list[Mapping[str, Any]], Mapping[str, Any]]: + raw_targets = str(environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() + items: list[Mapping[str, Any]] = [] + defaults: Mapping[str, Any] = {} + if raw_targets: + try: + payload = json.loads(raw_targets) + except json.JSONDecodeError as exc: + raise ValueError(f"CLOUD_RUN_SERVICE_TARGETS_JSON is invalid: {exc}") from exc + if isinstance(payload, Mapping): + targets = payload.get("targets") + defaults = _mapping(payload.get("defaults")) + else: + targets = payload + if isinstance(targets, list): + items = [target for target in targets if isinstance(target, Mapping)] + else: + raise ValueError( + "CLOUD_RUN_SERVICE_TARGETS_JSON must be an array or object with targets" + ) + + 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 as exc: + raise ValueError(f"RUNTIME_TARGET_JSON is invalid: {exc}") from exc + if isinstance(runtime_target, Mapping): + items = [runtime_target] + else: + raise ValueError("RUNTIME_TARGET_JSON must decode to an object") + return items, defaults + + +def runtime_target_configuration_present(environ: Mapping[str, str]) -> bool: + return bool( + str(environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() + or str(environ.get("RUNTIME_TARGET_JSON") or "").strip() + ) + + +def runtime_target_configuration_has_enabled_targets( + environ: Mapping[str, str], +) -> bool: + items, defaults = _load_runtime_target_items(environ) + if not items: + return False + expected_scope = str(environ.get("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE") or "").strip().lower() + for item in items: + runtime_target = _runtime_target(item, defaults) + target_scope = _first_value( + [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ], + ( + "account_scope", + "account_group", + "account_region", + "ACCOUNT_GROUP", + "ACCOUNT_REGION", + ), + ) + if expected_scope and target_scope and target_scope.lower() != expected_scope: + continue + enabled_value = _target_field( + item, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) + if enabled_value is None: + enabled_value = runtime_target.get("runtime_target_enabled") + if _enabled(enabled_value): + return True + return False + + +def load_runtime_targets( + environ: Mapping[str, str], + *, + include_disabled: bool = False, + profile_resolver: ProfileResolver | None = None, +) -> list[dict[str, Any]]: + items, defaults = _load_runtime_target_items(environ) + + expected_scope = str(environ.get("RUNTIME_HEARTBEAT_ACCOUNT_SCOPE") or "").strip().lower() + eligible: list[tuple[Mapping[str, Any], dict[str, Any], bool]] = [] + for item in items: + runtime_target = _runtime_target(item, defaults) + enabled_value = _target_field( + item, + defaults, + "runtime_target_enabled", + "RUNTIME_TARGET_ENABLED", + ) + if enabled_value is None: + enabled_value = runtime_target.get("runtime_target_enabled") + enabled = _enabled(enabled_value) + if not enabled and not include_disabled: + continue + target_scope = _first_value( + [ + runtime_target, + item, + _mapping(item.get("env")), + defaults, + _mapping(defaults.get("env")), + ], + ( + "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, enabled)) + + normalized: list[dict[str, Any]] = [] + enabled_count = sum(1 for _item, _runtime_target_value, enabled in eligible if enabled) + for item, runtime_target, enabled in eligible: + for service in _service_values(item, runtime_target, defaults, environ): + target = _normalize_target( + item, + runtime_target, + defaults, + service, + environ, + use_global_market_fallback=enabled_count == 1, + profile_resolver=profile_resolver, + ) + if include_disabled: + target["enabled"] = enabled + 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, + publication_grace: dt.timedelta, +) -> 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) != 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) + cursor = since_utc.replace(second=0, microsecond=0) + if cursor < since_utc: + cursor += dt.timedelta(minutes=1) + matured_at = now_utc - max(publication_grace, dt.timedelta()) + cron_due_at: list[dt.datetime] = [] + 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 and cursor <= matured_at: + cron_due_at.append(cursor) + cursor += dt.timedelta(minutes=1) + latest_cron_due_at = cron_due_at[-1] if cron_due_at else None + + session_dates: set[dt.date] | None = None + market_calendar = str(target.get("market_calendar") or "").strip() + market_timezone_name = ( + str(target.get("market_timezone") or "").strip() or timezone_name + ) + if market_aware and market_calendar: + 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, latest_cron_due_at + + latest_due_at = latest_cron_due_at + if session_dates is not None: + market_timezone = ZoneInfo(market_timezone_name) + latest_due_at = next( + ( + due_at + for due_at in reversed(cron_due_at) + if due_at.astimezone(market_timezone).date() in session_dates + ), + None, + ) + 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, + publication_grace: dt.timedelta = dt.timedelta(minutes=30), + 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, + publication_grace=publication_grace, + ) + 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_entrypoint.py b/tests/test_cloud_run_entrypoint.py index 2331c46..a078a51 100644 --- a/tests/test_cloud_run_entrypoint.py +++ b/tests/test_cloud_run_entrypoint.py @@ -7,7 +7,7 @@ from entrypoints.cloud_run import is_market_open_now, is_market_open_today -def test_is_market_open_today_falls_back_to_weekday_when_calendar_import_fails(monkeypatch): +def test_is_market_open_today_fails_closed_when_calendar_import_fails(monkeypatch): monkeypatch.setattr( "entrypoints.cloud_run.import_module", lambda _name: (_ for _ in ()).throw(TypeError("broken calendar")), @@ -25,10 +25,10 @@ def test_is_market_open_today_falls_back_to_weekday_when_calendar_import_fails(m ), ) - assert is_market_open_today() is True + assert is_market_open_today() is False -def test_is_market_open_now_falls_back_to_weekday_when_calendar_import_fails(monkeypatch): +def test_is_market_open_now_fails_closed_when_calendar_import_fails(monkeypatch): monkeypatch.setattr( "entrypoints.cloud_run.import_module", lambda _name: (_ for _ in ()).throw(TypeError("broken calendar")), @@ -46,7 +46,7 @@ def test_is_market_open_now_falls_back_to_weekday_when_calendar_import_fails(mon ), ) - assert is_market_open_now() is True + assert is_market_open_now() is False def test_is_market_open_now_returns_false_before_regular_session(monkeypatch): diff --git a/tests/test_cloud_run_runtime_guard.py b/tests/test_cloud_run_runtime_guard.py index 1166db9..113d596 100644 --- a/tests/test_cloud_run_runtime_guard.py +++ b/tests/test_cloud_run_runtime_guard.py @@ -157,6 +157,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 = { @@ -197,3 +245,49 @@ def test_strategy_request_capacity_warning_still_fails(monkeypatch): } assert guard._is_failure(entry) is True + + +def test_scheduler_job_matching_rejects_prefixed_service_names(): + pattern = guard._scheduler_job_pattern_for_services(["test-service"]) + entry = { + "timestamp": "2026-07-29T19:45:03Z", + "resource": { + "labels": {"job_id": "test-secondary-service-scheduler"} + }, + } + failures = { + "test-service": [{"timestamp": "2026-07-29T19:45:01Z"}], + } + fallback = dt.datetime(2026, 7, 29, 19, 0, tzinfo=dt.timezone.utc) + service_since = dt.datetime(2026, 7, 29, 19, 30, tzinfo=dt.timezone.utc) + + assert not re.search(pattern, "test-secondary-service-scheduler") + assert guard._scheduler_entry_since( + entry, + {"test-service": service_since}, + fallback, + ) == fallback + assert guard._is_duplicate_scheduler_failure(entry, failures) is False + + +def test_explicit_disabled_service_is_removed_using_target_defaults(monkeypatch): + monkeypatch.delenv("RUNTIME_GUARD_CLOUD_RUN_SERVICES", raising=False) + monkeypatch.delenv("CLOUD_RUN_SERVICE", raising=False) + monkeypatch.setenv("CLOUD_RUN_SERVICES", "enabled-service,disabled-service") + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [ + { + "service": "enabled-service", + "runtime_target_enabled": True, + }, + {"service": "disabled-service"}, + ], + } + ), + ) + + assert guard._load_services() == ["enabled-service"] diff --git a/tests/test_execution_report_heartbeat.py b/tests/test_execution_report_heartbeat.py index fb15bf8..1f11924 100644 --- a/tests/test_execution_report_heartbeat.py +++ b/tests/test_execution_report_heartbeat.py @@ -107,6 +107,7 @@ def test_explicit_required_services_skip_disabled_targets(monkeypatch): "CLOUD_RUN_SERVICE_TARGETS_JSON", json.dumps( { + "defaults": {"RUNTIME_TARGET_ENABLED": "false"}, "targets": [ { "service": "interactive-brokers-enabled-service", @@ -114,7 +115,6 @@ def test_explicit_required_services_skip_disabled_targets(monkeypatch): }, { "service": "interactive-brokers-disabled-service", - "RUNTIME_TARGET_ENABLED": "false", }, ] } @@ -492,6 +492,63 @@ def test_runtime_target_scheduler_does_not_skip_when_lookback_includes_scheduler assert reason is None +def test_main_rejects_previous_session_report_after_a_new_session_is_due( + monkeypatch, + capsys, +): + _clear_runtime_env(monkeypatch) + monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "IBKR US runtime") + monkeypatch.setenv("RUNTIME_HEARTBEAT_REQUIRED_SERVICES", "svc-us") + monkeypatch.setenv("RUNTIME_HEARTBEAT_GCS_URIS", "gs://bucket/reports") + monkeypatch.setenv( + "RUNTIME_TARGET_JSON", + json.dumps( + { + "service_name": "svc-us", + "strategy_profile": "us-strategy", + "account_scope": "US", + "scheduler": { + "timezone": "America/New_York", + "main_time": "45 15 * * *", + }, + "market": "US", + "market_calendar": "NYSE", + "market_timezone": "America/New_York", + } + ), + ) + monkeypatch.setattr( + heartbeat, + "_list_gcs_objects", + lambda *_args, **_kwargs: [ + { + "url": "gs://bucket/reports/previous.json", + "metadata": {"updated": "2026-07-28T19:46:00Z"}, + } + ], + ) + monkeypatch.setattr( + heartbeat, + "_cat_gcs_json", + lambda *_args, **_kwargs: { + "status": "ok", + "service_name": "svc-us", + "strategy_profile": "us-strategy", + "account_scope": "US", + }, + ) + monkeypatch.setattr(heartbeat, "_send_telegram", lambda _message: True) + + result = heartbeat.main( + now=dt.datetime(2026, 7, 29, 22, 20, tzinfo=dt.timezone.utc) + ) + + assert result == 1 + output = capsys.readouterr().out + assert "missing acceptable report for runtime target(s)" in output + assert "predates latest due schedule" in output + + def test_report_with_blocked_execution_status_is_rejected_even_when_top_level_is_ok(): accepted, reason = heartbeat._is_accepted_report( { @@ -530,6 +587,25 @@ def test_report_with_expected_execution_guard_is_accepted_when_top_level_is_ok(n assert reason == "status=ok" +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 + + def test_telegram_token_falls_back_to_secret_manager(monkeypatch): monkeypatch.delenv("TELEGRAM_TOKEN", raising=False) monkeypatch.delenv("TG_TOKEN", raising=False) @@ -555,3 +631,59 @@ def fake_run_gcloud(command): "--project", "interactivebrokersquant", ] + + +def test_incomplete_target_schedule_uses_deployed_scheduler_cron(monkeypatch): + targets = [ + { + "service": "interactive-brokers-service", + "scheduler": { + "main_time": "45 15", + "timezone": "America/New_York", + }, + } + ] + monkeypatch.setattr( + heartbeat, + "_describe_scheduler_job", + lambda job_name, **_kwargs: ( + { + "schedule": "45 15 25-29 * *", + "timeZone": "America/New_York", + } + if job_name == "interactive-brokers-service-scheduler" + else None + ), + ) + + hydrated = heartbeat._hydrate_runtime_target_schedules( + targets, + project="test-project", + ) + + assert hydrated[0]["scheduler"]["main_time"] == "45 15 25-29 * *" + + +def test_main_skips_when_all_configured_targets_are_disabled(monkeypatch, capsys): + monkeypatch.delenv("RUNTIME_TARGET_ENABLED", raising=False) + monkeypatch.delenv("RUNTIME_TARGET_JSON", raising=False) + monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "IBKR disabled targets") + monkeypatch.setenv( + "CLOUD_RUN_SERVICE_TARGETS_JSON", + json.dumps( + { + "defaults": {"runtime_target_enabled": False}, + "targets": [{"service": "disabled-service"}], + } + ), + ) + monkeypatch.setattr( + heartbeat, + "_list_gcs_objects", + lambda *_args, **_kwargs: pytest.fail("GCS should not be queried"), + ) + + assert heartbeat.main( + now=dt.datetime(2026, 6, 20, 23, 10, tzinfo=dt.timezone.utc) + ) == 0 + assert "no enabled runtime target matches this heartbeat" in capsys.readouterr().out diff --git a/tests/test_notifications.py b/tests/test_notifications.py index df74f9b..af5c2e3 100644 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -272,13 +272,14 @@ class FakeRequests: def post(*args, **kwargs): return FakeResponse() - send_telegram_message( + sent = send_telegram_message( "hello", token="token", chat_id="chat-id", requests_module=FakeRequests, ) + assert sent is False captured = capsys.readouterr() assert "Telegram send failed with status 401: unauthorized" in captured.out @@ -296,13 +297,14 @@ def post(*args, **kwargs): calls.append((args, kwargs)) return FakeResponse() - send_telegram_message( + sent = send_telegram_message( "SOXL.US 预计；00700.HK 持仓；https://example.com 保持原样", token="token", chat_id="chat-id", requests_module=FakeRequests, ) + assert sent is True payload = calls[0][1]["json"] assert payload["text"] == "SOXL.\u2060US 预计；00700.\u2060HK 持仓；https://example.com 保持原样" diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 68ed279..d4b0a47 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -938,6 +938,28 @@ def test_notification_delivery_log_summary_stays_empty_without_sent_event(strate assert payload == {} +def test_notification_delivery_summary_keeps_failed_transport_receipt(strategy_module): + payload = strategy_module._build_notification_delivery_summary( + [ + { + "sink": "telegram", + "delivery_status": "failed", + "transport_acknowledged": False, + "error_type": "RuntimeError", + "compact_text_sha256": "a" * 64, + "compact_text_length": 42, + } + ] + ) + + assert payload["attempted_count"] == 1 + assert payload["sent_count"] == 0 + assert payload["failed_count"] == 1 + assert payload["all_acknowledged"] is False + assert payload["delivery_events"][0]["error_type"] == "RuntimeError" + assert "compact_text" not in payload["delivery_events"][0] + + def test_handle_request_post_returns_market_closed_when_schedule_empty(strategy_module, monkeypatch): observed = {} @@ -1144,12 +1166,13 @@ def test_send_tg_message_uses_cycle_channel_sender(strategy_module, monkeypatch) "quant_platform_kit.notifications.cycle_channel.build_cycle_sender", lambda **kwargs: ( observed.setdefault("sender_config", kwargs), - lambda message: observed.setdefault("message", message), + lambda message: (observed.setdefault("message", message), False)[1], )[1], ) - strategy_module.send_tg_message("hello") + sent = strategy_module.send_tg_message("hello") + assert sent is False assert observed == { "sender_config": { "channel": "telegram", diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index 6f145f0..a2c3491 100644 --- a/tests/test_runtime_config_support.py +++ b/tests/test_runtime_config_support.py @@ -97,6 +97,9 @@ def runtime_target_json( account_selector: list[str] | tuple[str, ...] | None = None, account_scope: str = "default", service_name: str | None = None, + market: str | None = None, + market_calendar: str | None = None, + market_timezone: str | None = None, ) -> str: payload: dict[str, object] = { "platform_id": platform_id, @@ -109,6 +112,12 @@ def runtime_target_json( payload["account_selector"] = list(account_selector) if service_name is not None: payload["service_name"] = service_name + if market is not None: + payload["market"] = market + if market_calendar is not None: + payload["market_calendar"] = market_calendar + if market_timezone is not None: + payload["market_timezone"] = market_timezone payload["execution_mode"] = "paper" if dry_run_only else "live" return json.dumps(payload, separators=(",", ":")) @@ -240,6 +249,26 @@ def test_load_platform_runtime_settings_uses_minimal_group_config(monkeypatch): assert settings.strategy_plugin_alert_telegram_body_max_chars is None +def test_runtime_target_market_overrides_account_group_defaults(monkeypatch): + monkeypatch.setenv( + "RUNTIME_TARGET_JSON", + runtime_target_json( + SAMPLE_STRATEGY_PROFILE, + market="HK", + market_calendar="XHKG", + market_timezone="Asia/Hong_Kong", + ), + ) + monkeypatch.setenv("ACCOUNT_GROUP", "paper") + monkeypatch.setenv("IB_ACCOUNT_GROUP_CONFIG_JSON", MINIMAL_GROUP_JSON) + + settings = load_platform_runtime_settings(project_id_resolver=lambda: "project-1") + + assert settings.market == HK_MARKET + assert settings.market_calendar == "XHKG" + assert settings.market_timezone == "Asia/Hong_Kong" + + def test_load_platform_runtime_settings_prefers_runtime_target_json(monkeypatch): monkeypatch.setenv("RUNTIME_TARGET_JSON", runtime_target_json(SAMPLE_STRATEGY_PROFILE)) monkeypatch.setenv("ACCOUNT_GROUP", "paper") diff --git a/tests/test_runtime_heartbeat_policy.py b/tests/test_runtime_heartbeat_policy.py new file mode 100644 index 0000000..c9af229 --- /dev/null +++ b/tests/test_runtime_heartbeat_policy.py @@ -0,0 +1,314 @@ +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, + runtime_target_configuration_present, + target_key, + target_latest_due_at, +) +from scripts import execution_report_heartbeat as heartbeat + + +def _july(day: int, hour: int, minute: int) -> dt.datetime: + return dt.datetime(2026, 7, day, hour, minute, tzinfo=dt.timezone.utc) + + +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_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, 20, 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]) == _july(29, 19, 45) + + + +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_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 len(due) == 1 + assert target_latest_due_at(due[0]) == _july(3, 19, 45) + assert evaluated is False + + +def test_target_defaults_and_scheduler_aliases_are_normalized() -> None: + targets = load_runtime_targets( + { + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps( + { + "defaults": { + "env": { + "RUNTIME_TARGET_ENABLED": "false", + "CLOUD_SCHEDULER_MAIN_TIME": "45 15 25-29 * *", + }, + "market": "US", + }, + "targets": [ + { + "service": "disabled-service", + "runtime_target": { + "strategy_profile": "disabled-strategy", + }, + }, + { + "service": "enabled-service", + "RUNTIME_TARGET_ENABLED": "true", + "runtime_target": { + "strategy_profile": "enabled-strategy", + }, + }, + ], + } + ) + } + ) + + assert len(targets) == 1 + assert targets[0]["service"] == "enabled-service" + assert targets[0]["market"] == "US" + assert targets[0]["scheduler"] == { + "main_time": "45 15 25-29 * *", + "timezone": "America/New_York", + } + + + +def test_publication_grace_uses_previous_matured_schedule_cutoff() -> None: + targets = load_runtime_targets( + { + "RUNTIME_TARGET_JSON": json.dumps( + { + "service_name": "grace-service", + "strategy_profile": "grace-strategy", + "scheduler": { + "main_time": "0 12 * * *", + "timezone": "UTC", + }, + } + ) + } + ) + since = dt.datetime(2026, 7, 28, 11, 30, tzinfo=dt.timezone.utc) + + within_grace, evaluated = filter_due_targets( + targets, + since=since, + now=dt.datetime(2026, 7, 29, 12, 5, tzinfo=dt.timezone.utc), + market_aware=False, + publication_grace=dt.timedelta(minutes=30), + ) + after_grace, _ = filter_due_targets( + targets, + since=since, + now=dt.datetime(2026, 7, 29, 12, 31, tzinfo=dt.timezone.utc), + market_aware=False, + publication_grace=dt.timedelta(minutes=30), + ) + + assert evaluated is True + assert target_latest_due_at(within_grace[0]) == _july(28, 12, 0) + assert target_latest_due_at(after_grace[0]) == _july(29, 12, 0) diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py new file mode 100644 index 0000000..01dc2f5 --- /dev/null +++ b/tests/test_runtime_monitor_workflows.py @@ -0,0 +1,26 @@ +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 "RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES:" in workflow + assert "RUNTIME_HEARTBEAT_SCHEDULER_LOCATION:" in workflow + assert "CLOUD_SCHEDULER_MAIN_TIME:" 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"