From df85e241a45cfb3c87ee28c6339cfe50defd3871 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 16 Oct 2026 06:41:51 +0800 Subject: [PATCH 1/3] feat(notifications): wire opt-in unified digest without trading pin changes Co-Authored-By: Codex --- .../workflows/execution-report-heartbeat.yml | 16 +++ .github/workflows/sync-cloud-run-env.yml | 11 ++ application/rebalance_service.py | 15 ++- docs/unified-daily-notification.md | 32 +++++ ...-notification-trading-diff-review.zh-CN.md | 25 ++++ main.py | 11 +- notifications/digest_snapshot/__init__.py | 1 + .../digest_snapshot/digest_archive.py | 127 ++++++++++++++++++ .../digest_snapshot/digest_candidates.py | 105 +++++++++++++++ .../digest_snapshot/digest_routing.py | 92 +++++++++++++ scripts/emit_unified_digest_candidates.py | 8 ++ tests/test_unified_digest_candidates.py | 107 +++++++++++++++ tests/test_unified_digest_routing.py | 75 +++++++++++ tests/test_unified_notification_wiring.py | 16 +++ 14 files changed, 639 insertions(+), 2 deletions(-) create mode 100644 docs/unified-daily-notification.md create mode 100644 docs/unified-notification-trading-diff-review.zh-CN.md create mode 100644 notifications/digest_snapshot/__init__.py create mode 100644 notifications/digest_snapshot/digest_archive.py create mode 100644 notifications/digest_snapshot/digest_candidates.py create mode 100644 notifications/digest_snapshot/digest_routing.py create mode 100644 scripts/emit_unified_digest_candidates.py create mode 100644 tests/test_unified_digest_candidates.py create mode 100644 tests/test_unified_digest_routing.py create mode 100644 tests/test_unified_notification_wiring.py diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml index f5a7f21..c130199 100644 --- a/.github/workflows/execution-report-heartbeat.yml +++ b/.github/workflows/execution-report-heartbeat.yml @@ -286,6 +286,22 @@ jobs: if: ${{ !cancelled() && env.INSPECT_SG_PROBE_PERMISSIONS != 'true' && steps.account_history.outcome == 'failure' }} run: exit 1 + + - name: Publish existing reports as unified digest candidates + if: ${{ success() && vars.QSL_ROUTINE_NOTIFICATION_MODE == 'daily_digest' }} + env: + QSL_ROUTINE_NOTIFICATION_MODE: ${{ vars.QSL_ROUTINE_NOTIFICATION_MODE || 'immediate' }} + RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }} + RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON || secrets.RUNTIME_TARGET_JSON }} + EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI || secrets.EXECUTION_REPORT_GCS_URI }} + DIGEST_CANDIDATES_GCS_URI: ${{ vars.DIGEST_CANDIDATES_GCS_URI || secrets.DIGEST_CANDIDATES_GCS_URI }} + DIGEST_STORE_TARGET: ${{ vars.DIGEST_STORE_TARGET }} + DIGEST_OPAQUE_ACCOUNT_UID: ${{ secrets.DIGEST_OPAQUE_ACCOUNT_UID }} + DIGEST_TARGET_ID: ${{ secrets.DIGEST_TARGET_ID }} + DIGEST_TARGET_BINDINGS_JSON: ${{ secrets.DIGEST_TARGET_BINDINGS_JSON }} + DIGEST_NOTIFICATION_SCHEDULE_JSON: ${{ vars.DIGEST_NOTIFICATION_SCHEDULE_JSON }} + run: uv run --no-sync python scripts/emit_unified_digest_candidates.py + inspect-account-snapshot: needs: resolve-matrix if: ${{ github.event_name == 'workflow_dispatch' && ((inputs.inspect_sg_account_snapshot == true && inputs.target == 'sg') || (inputs.inspect_paper_account_snapshot == true && inputs.target == 'paper')) }} diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 77276f8..6b92743 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -683,6 +683,8 @@ jobs: IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW: ${{ vars.IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW }} RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }} NOTIFY_LANG: ${{ vars.NOTIFY_LANG }} + QSL_ROUTINE_NOTIFICATION_MODE: ${{ vars.QSL_ROUTINE_NOTIFICATION_MODE }} + QSL_DIGEST_ROUTINE_ORDERS: ${{ vars.QSL_DIGEST_ROUTINE_ORDERS }} EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI }} LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET }} LONGBRIDGE_DRY_RUN_ONLY: ${{ vars.LONGBRIDGE_DRY_RUN_ONLY }} @@ -1241,6 +1243,15 @@ jobs: } env_pairs=("${target_env_pairs[@]}") + for notification_key in QSL_ROUTINE_NOTIFICATION_MODE QSL_DIGEST_ROUTINE_ORDERS; do + if [ -n "${!notification_key:-}" ]; then + case "${notification_key}:${!notification_key}" in + QSL_ROUTINE_NOTIFICATION_MODE:immediate|QSL_ROUTINE_NOTIFICATION_MODE:daily_digest|QSL_DIGEST_ROUTINE_ORDERS:true|QSL_DIGEST_ROUTINE_ORDERS:false) ;; + *) echo "Invalid notification routing configuration" >&2; exit 2 ;; + esac + env_pairs+=("${notification_key}=${!notification_key}") + fi + done secret_pairs=() remove_env_vars=("${target_remove_env_vars[@]}") remove_env_vars+=( diff --git a/application/rebalance_service.py b/application/rebalance_service.py index e40a2ea..b9f45ab 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -633,6 +633,16 @@ def _record_execution_marker( +def _defer_cycle_notification(result, *, config): + from notifications.digest_snapshot.digest_routing import defer_routine_cycle_to_digest + evidence = {**dict(result.allocation or {}), **dict(result.execution or {})} + evidence.setdefault("execution_status", "completed" if result.action_done else "no_action") + evidence["orders_pending_count"] = len(tuple(getattr(result, "pending_orders", ()) or ())) + orders = tuple(getattr(result, "submitted_orders", ()) or ()) + return not config.dry_run_only and defer_routine_cycle_to_digest( + evidence, submitted_order_count=len(orders), submitted_orders=orders) + + def run_strategy( *, runtime: LongBridgeRebalanceRuntime, @@ -1227,7 +1237,10 @@ def submit_claimed_order(order_intent): if isinstance(position_currency_by_symbol, dict): execution["position_currency_by_symbol"] = dict(position_currency_by_symbol) - if pending_orders: + routine_to_digest = _defer_cycle_notification(execution_result, config=config) + if routine_to_digest: + print(config.with_prefix("notification_suppressed reason=daily_digest"), flush=True) + elif pending_orders: try: notification_publisher.publish( notification_renderers.render_rebalance_notification( diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md new file mode 100644 index 0000000..971cf76 --- /dev/null +++ b/docs/unified-daily-notification.md @@ -0,0 +1,32 @@ +# Unified daily notification (code only) + +`QSL_ROUTINE_NOTIFICATION_MODE` defaults to `immediate`. Exact `daily_digest` +opts into one central bilingual digest. `QSL_DIGEST_ROUTINE_ORDERS=false` retains +immediate confirmed order notices. Failed, pending, uncertain, risk, persistence +and authentication outcomes retain immediate alerts. Healthy idle cycles and +routine plugin observations do not produce individual pushes. + +QPK source reference: `b0d10862cfc0ffe9e1b868861b937a9cac6e7201` (notification modules only). + +The notification-only QPK source snapshot in `notifications/digest_snapshot/` avoids an +unrelated trading dependency upgrade. Keep these three modules byte-identical +to the reviewed QPK notification modules. Existing QPK/strategy pins, order +execution, funding, risk, account enablement and durable report writes are unchanged. + +Before enabling routing, validate the exact active target's private archived +report, candidate store and QRS receiver. Existing cloud monitor wiring uses +`RUNTIME_TARGET_ENABLED=true`, exact protected `RUNTIME_TARGET_JSON`, +`EXECUTION_REPORT_GCS_URI`, `DIGEST_CANDIDATES_GCS_URI`, `DIGEST_STORE_TARGET`, +`DIGEST_OPAQUE_ACCOUNT_UID` and `DIGEST_TARGET_ID`. Missing configuration stops +projection; disabled targets stay disabled. Do not infer private identities. +The producer aggregates every unique cycle in the completed 06:00 SGT window; +missing evidence is unknown, never zero. No new collector or scheduler is added. + +No deployment, variable application, IAM, secret or live message is authorized +by this PR. Source adoption and delivery require separate approved release work. + +For multiple active targets, protected `DIGEST_TARGET_BINDINGS_JSON` supplies a +list of per-target environment mappings. `DIGEST_NOTIFICATION_SCHEDULE_JSON` +can reuse the existing heartbeat scheduler/calendar declaration to prove a +window was not due. Missing due reports stay errors; closed configured sessions +produce an `actually_ran=false` marker without inventing zero trading counts. diff --git a/docs/unified-notification-trading-diff-review.zh-CN.md b/docs/unified-notification-trading-diff-review.zh-CN.md new file mode 100644 index 0000000..2703bcb --- /dev/null +++ b/docs/unified-notification-trading-diff-review.zh-CN.md @@ -0,0 +1,25 @@ +# 交易差异审查：LongBridgePlatform + +审查方式：本次作者自审，辅以 AST 调用参数比较；不声称独立人工审查。 +基线：`a0de99314ebd15fcfc7f3e99de3b54dbb924cb7c`。范围为当前 PR 相对基线的全部改动。 + +结论：源码差异仅涉及通知路由、附加的通知报告事实、已有监测入口的候选投影、通知配置透传、测试和说明。交易依赖、锁文件、UES/QPK pin、信号、仓位、资金限额、订单构造、券商调用、状态 claim 和风险门禁保持原样。 + +现有文件中的券商、订单、风险、交易状态和报告持久化调用，已对调用表达式及全部参数做 AST 比较： + +| 文件 | 敏感调用数量 | 参数完全相同 | +|---|---:|---| +| `application/rebalance_service.py` | 5 | 是 | +| `main.py` | 12 | 是 | + +新增模块不调用券商。只有既有云监测器在 `daily_digest` 且目标明确启用时，才读已有私有报告并写候选；不启用账户、不修改订单或交易排程。关闭模式保持既有行为。失败、未决订单、风险、认证及持久化问题保留即时通知；已确认的普通订单可通过 `QSL_DIGEST_ROUTINE_ORDERS=false` 继续即时发送。 + +通知代码取自 QPK `b0d10862cfc0ffe9e1b868861b937a9cac6e7201`，有意不升级平台交易依赖。三个源文件逐字节相同： + +| 文件 | SHA256 | +|---|---| +| `digest_routing.py` | `60c3f258c4d1a03bc49c7d03b392c02c316541726dfacb0b1d4fae6dbd2e5017` | +| `digest_candidates.py` | `a6a0d10f90a17906c45393a84e819c8d71f430300e977fdb4fa1f824e26878f5` | +| `digest_archive.py` | `9a316072dc9e70ffb03c585c8c2f49876771ea5d722624fb5263ff037b27e3bc` | + +合并条件：相关验证通过、PR 当前精确 head 的 CI 全绿、当前远端主线的 push / workflow_run / 可复用 workflow 调用链无自动部署。否则保留草稿。此审查不批准从旧生产 revision 升级到整个当前主线；生产到候选的完整差异、UES 指纹与 baseline 仍须按后续发布方案单独审查。 diff --git a/main.py b/main.py index df12a5f..bbc5d08 100644 --- a/main.py +++ b/main.py @@ -426,6 +426,14 @@ def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict: account_identity = execution.get("account_identity") if isinstance(account_identity, dict): summary["account_identity"] = dict(account_identity) + from notifications.digest_snapshot.digest_routing import daily_digest_enabled + if daily_digest_enabled(): + from notifications.digest_snapshot.digest_candidates import cycle_evidence + evidence = {**dict(getattr(cycle_result, "allocation", {}) or {}), **execution} + evidence.setdefault("execution_status", "completed" if broker_submission_done else "no_action") + evidence["orders_pending_count"] = len(pending_orders) + summary["notification_digest"] = cycle_evidence(evidence, + tuple(getattr(cycle_result, "submitted_orders", ()) or ()), dry_run=dry_run) return summary @@ -716,8 +724,9 @@ def build_strategy_plugin_alert_context_label() -> str: def publish_strategy_plugin_alerts(signals, *, report=None): + from notifications.digest_snapshot.digest_routing import immediate_plugin_signals result = dispatch_strategy_plugin_alerts( - signals, + immediate_plugin_signals(signals), notification_settings=RUNTIME_SETTINGS, translator=t, strategy_label=STRATEGY_PROFILE, diff --git a/notifications/digest_snapshot/__init__.py b/notifications/digest_snapshot/__init__.py new file mode 100644 index 0000000..63deddf --- /dev/null +++ b/notifications/digest_snapshot/__init__.py @@ -0,0 +1 @@ +"""Notification-only QPK source snapshot; trading dependency pins stay unchanged.""" diff --git a/notifications/digest_snapshot/digest_archive.py b/notifications/digest_snapshot/digest_archive.py new file mode 100644 index 0000000..aa1b0b1 --- /dev/null +++ b/notifications/digest_snapshot/digest_archive.py @@ -0,0 +1,127 @@ +"""Publish notification candidates from existing archives; disabled by default. + +Runs only inside an existing cloud monitor, with its existing identity. It never +invokes a platform service or broker, reads secrets, or enables a runtime target. +""" +from __future__ import annotations + +import json +import os +import re +import subprocess +from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo + +from .digest_candidates import project_reports + +_SAFE = re.compile(r"[a-z0-9][a-z0-9_-]{0,40}\Z") + + +def _read(args): + result = subprocess.run(["gcloud", "storage", *args], capture_output=True, text=True, check=False) + if result.returncode: + raise ValueError("candidate_archive_read_failed") + return json.loads(result.stdout) + + +def emit(env, *, now=None, read=_read, upload=None): + if env.get("QSL_ROUTINE_NOTIFICATION_MODE", "immediate") != "daily_digest": + return {"status": "disabled", "uploaded": False} + if env.get("RUNTIME_TARGET_ENABLED") == "false": + return {"status": "target_disabled", "uploaded": False} + if env.get("RUNTIME_TARGET_ENABLED") != "true": + raise ValueError("candidate_target_enablement_unknown") + bindings = env.get("DIGEST_TARGET_BINDINGS_JSON", "").strip() + if bindings: + values = json.loads(bindings) + if not isinstance(values, list) or not values or len(values) > 100 or not all(isinstance(v, dict) for v in values): + raise ValueError("candidate_bindings_invalid") + results = [] + for value in values: + results.append(emit({**env, **value, "DIGEST_TARGET_BINDINGS_JSON": ""}, now=now, read=read, upload=upload)) + return {"status": "projected", "uploaded": any(v["uploaded"] for v in results), "targets": len(results)} + target = json.loads(env.get("RUNTIME_TARGET_JSON", "")) + platform = target.get("platform_id", "") + slot = env.get("DIGEST_STORE_TARGET", "") + if not _SAFE.fullmatch(platform) or not _SAFE.fullmatch(slot): + raise ValueError("candidate_store_identity_missing") + uid, tid = env.get("DIGEST_OPAQUE_ACCOUNT_UID", ""), env.get("DIGEST_TARGET_ID", "") + if not uid or not tid: + raise ValueError("candidate_identity_missing") + prefix = env.get("EXECUTION_REPORT_GCS_URI", "").rstrip("/") + destination = env.get("DIGEST_CANDIDATES_GCS_URI", "").rstrip("/") + if any(not re.fullmatch(r"gs://[a-z0-9][a-z0-9._-]+/[^\s?#*]+", value) for value in (prefix, destination)): + raise ValueError("candidate_store_configuration_missing") + current = now or datetime.now(timezone.utc) + if current.astimezone(ZoneInfo("Asia/Singapore")).hour < 6: + return {"status": "before_cutoff", "uploaded": False} + day = (current.astimezone(ZoneInfo("Asia/Singapore")) - timedelta(hours=6)).date() - timedelta(days=1) + # Monitor runs after the fixed 06:00 SGT cut-off. Reject early runs rather + # than overwriting the previous day's complete candidate with partial facts. + end = datetime(day.year, day.month, day.day, 6, tzinfo=ZoneInfo("Asia/Singapore")) + timedelta(days=1) + start = end - timedelta(days=1) + months = {start.astimezone(timezone.utc).strftime("%Y-%m"), end.astimezone(timezone.utc).strftime("%Y-%m")} + entries = [] + for month in sorted(months): + entries.extend(read(["ls", "--json", f"{prefix}/{platform}/**/{month}/*.json"])) + if len(entries) > 10000: + raise ValueError("candidate_archive_budget_exceeded") + reports, uris = [], set() + for entry in entries: + uri = str(entry.get("url", "")).split("#", 1)[0] + if uri in uris: + continue + if not uri.startswith(f"{prefix}/{platform}/"): + raise ValueError("candidate_source_locator_mismatch") + uris.add(uri) + raw_stamp = (entry.get("metadata") or {}).get("updated") or (entry.get("metadata") or {}).get("timeCreated") + stamp = datetime.fromisoformat(str(raw_stamp).replace("Z", "+00:00")) + if stamp.tzinfo is None: + raise ValueError("candidate_archive_time_missing") + if stamp < start or stamp > current: + continue + source = read(["cat", uri]) + source_target = source.get("runtime_target") or {} + keys = ("platform_id", "strategy_profile", "service_name", "account_scope", "account_selector", "deployment_selector", "dry_run_only") + if not isinstance(source_target, dict) or any(source_target.get(key) != target.get(key) for key in keys): + continue + source_stamp = datetime.fromisoformat(str(source.get("started_at", "")).replace("Z", "+00:00")) + if source_stamp.tzinfo is None: + raise ValueError("candidate_source_time_invalid") + if start <= source_stamp < end: + reports.append(source) + if not reports: + schedule_raw = env.get("DIGEST_NOTIFICATION_SCHEDULE_JSON", "").strip() + if not schedule_raw: + raise ValueError("candidate_source_absent") + from quant_platform_kit.common.runtime_heartbeat_policy import filter_due_targets + schedule = json.loads(schedule_raw) + due, evaluated = filter_due_targets([schedule], since=start, now=end - timedelta(microseconds=1), + publication_grace=timedelta(0), warning_logger=lambda message: None) + if not evaluated or due: + raise ValueError("candidate_due_source_absent") + payload = {"schema_version": "qsl.digest_candidates.v1", "runs": [{ + "platform_id": platform, "strategy_profile": target["strategy_profile"], + "opaque_account_uid": uid, "target_id": tid, "business_day": day.isoformat(), + "actually_ran": False, "status": "ok", "scheduled_idle": True, + "evidence_provenance": "candidates", + }]} + else: + payload = project_reports(reports, expected_target=target, opaque_account_uid=uid, target_id=tid, now=current) + if upload is None: + from google.cloud import storage + def upload(uri, body): + storage.Blob.from_string(uri, client=storage.Client()).upload_from_string( + body, content_type="application/json", retry=None, timeout=30) + # Never retry an unknown write result. Public output contains no identities. + upload(f"{destination}/{platform}/{day.isoformat()}/{slot}.json", json.dumps(payload, allow_nan=False)) + return {"status": "projected", "uploaded": True, "runs": len(reports)} + + +def main(): + try: + print(json.dumps(emit(os.environ), sort_keys=True)) + return 0 + except Exception: + print(json.dumps({"status": "blocked", "uploaded": False, "reason": "candidate_projection_failed"})) + return 1 diff --git a/notifications/digest_snapshot/digest_candidates.py b/notifications/digest_snapshot/digest_candidates.py new file mode 100644 index 0000000..997fd53 --- /dev/null +++ b/notifications/digest_snapshot/digest_candidates.py @@ -0,0 +1,105 @@ +"""Notification-only facts projected from existing, privately archived reports. + +No broker calls. Missing or contradictory evidence stays unknown. Callers must +supply the exact protected runtime target; public labels never establish it. +""" +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo + + +def cycle_evidence(execution: Mapping, orders: Sequence[Mapping], *, dry_run: bool = False) -> dict: + """Keep counts derived from actual order responses, never from log lines.""" + from .digest_routing import defer_routine_cycle_to_digest + + count = len(orders) + valid = all(isinstance(order, Mapping) and bool(order.get("status")) for order in orders) + confirmed = valid and all(str(order["status"]).lower() in {"filled", "completed", "executed"} for order in orders) + healthy = defer_routine_cycle_to_digest(execution, submitted_order_count=count, + submitted_orders=orders, mode="daily_digest", include_orders=True) + return { + "schema_version": "qsl.notification_cycle.v1", + "order_count": count if valid and not dry_run else None, + "fill_count": count if confirmed and not dry_run else None, + "routine": healthy and not dry_run, + "dry_run": dry_run, + } + + +def project_reports(reports: Sequence[Mapping], *, expected_target: Mapping, + opaque_account_uid: str, target_id: str, now: datetime) -> dict: + """Aggregate each unique persisted cycle for the same target and source day.""" + if any(not isinstance(v, str) or not v.strip() for v in (opaque_account_uid, target_id)) or now.tzinfo is None: + raise ValueError("candidate_identity_or_time_missing") + required = {"platform_id", "strategy_profile", "service_name", "account_scope", "account_selector", "deployment_selector", "dry_run_only"} + if not required.issubset(expected_target) or expected_target.get("dry_run_only") is not False: + raise ValueError("candidate_target_invalid") + if (any(not isinstance(expected_target[k], str) or not expected_target[k].strip() + for k in required - {"account_selector", "dry_run_only"}) + or not isinstance(expected_target["account_selector"], list) or not expected_target["account_selector"] + or any(not isinstance(v, str) or not v.strip() for v in expected_target["account_selector"])): + raise ValueError("candidate_target_invalid") + rows, seen = {}, set() + for report in reports: + target = report.get("runtime_target") or {} + if not isinstance(target, Mapping) or any(target.get(k) != expected_target[k] for k in required): + raise ValueError("candidate_source_identity_mismatch") + if (report.get("schema_version") != "runtime_report.v1" or report.get("dry_run") is not False + or report.get("platform") != expected_target["platform_id"] + or report.get("strategy_profile") != expected_target["strategy_profile"] + or report.get("service_name") != expected_target["service_name"]): + raise ValueError("candidate_source_invalid") + stamp = datetime.fromisoformat(str(report.get("started_at", "")).replace("Z", "+00:00")) + if stamp.tzinfo is None or not timedelta(0) <= now - stamp <= timedelta(hours=36): + raise ValueError("candidate_source_time_invalid") + if report.get("status") not in {"ok", "completed", "success", "error", "aborted", "failed"}: + raise ValueError("candidate_source_incomplete") + run_id = report.get("run_id") + if not isinstance(run_id, str) or not run_id: + raise ValueError("candidate_run_id_missing") + if run_id in seen: + raise ValueError("candidate_duplicate_run") + seen.add(run_id) + day = (stamp.astimezone(ZoneInfo("Asia/Singapore")) - timedelta(hours=6)).date().isoformat() + facts = (report.get("summary") or {}).get("notification_digest") or {} + known = (facts.get("schema_version") == "qsl.notification_cycle.v1" and facts.get("dry_run") is False + and all(type(facts.get(k)) is int and facts[k] >= 0 for k in ("order_count", "fill_count")) + and facts["fill_count"] <= facts["order_count"]) + error = report.get("status") in {"error", "failed", "aborted"} or bool((report.get("error_summary") or {}).get("errors")) + alert = error or not known or facts.get("routine") is not True + row = rows.setdefault(day, { + "platform_id": expected_target["platform_id"], "strategy_profile": expected_target["strategy_profile"], + "opaque_account_uid": opaque_account_uid, "target_id": target_id, "business_day": day, + "actually_ran": True, "as_of": stamp.isoformat(), "status": "ok", "fill_count": 0, + "order_count": 0, "cycle_count": 0, "field_status": {}, "evidence_provenance": "candidates", + "persistence_status": "persisted", "delivery_status": "pending_central", + }) + row["cycle_count"] += 1 + row["as_of"] = max(row["as_of"], stamp.isoformat()) + for key in ("fill_count", "order_count"): + row[key] = row[key] + facts[key] if known and row[key] is not None else None + row["field_status"][key] = "known" if row[key] is not None else "counts_unknown" + if alert: + row.update(status="alert", reason_code="source_cycle_alert" if error else "execution_evidence_missing_or_attention", + execution_status="execution_alert" if error else "execution_unknown") + return {"schema_version": "qsl.digest_candidates.v1", "producer_status": "projected", "runs": list(rows.values())} + + +def summary_evidence(summary: Mapping, *, dry_run: bool = False) -> dict: + """IBKR named order collections are broker facts, unlike formatted logs.""" + if not isinstance(summary.get("orders_submitted"), (list, tuple)): + return {"schema_version": "qsl.notification_cycle.v1", "order_count": None, + "fill_count": None, "routine": False, "dry_run": dry_run} + orders = list(summary.get("orders_submitted") or ()) + list(summary.get("option_orders_submitted") or ()) + filled = list(summary.get("orders_filled") or ()) + list(summary.get("option_orders_filled") or ()) + evidence = dict(summary) + if (orders and len(orders) == len(filled) and not summary.get("orders_pending") + and not summary.get("option_orders_pending") and not summary.get("orders_partially_filled") + and not summary.get("option_orders_partially_filled")): + # These collections are populated by the existing broker reconciliation. + facts = [{"status": "filled"} for _ in filled] + else: + facts = orders + return cycle_evidence(evidence, facts, dry_run=dry_run) diff --git a/notifications/digest_snapshot/digest_routing.py b/notifications/digest_snapshot/digest_routing.py new file mode 100644 index 0000000..3d6929d --- /dev/null +++ b/notifications/digest_snapshot/digest_routing.py @@ -0,0 +1,92 @@ +"""Opt-in routing for routine cycles whose persisted reports feed a daily digest. + +This decides notification delivery only. It never changes an execution result, +calls a broker, grants order permissions, or treats deferral as delivery. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping, Sequence + + +def daily_digest_enabled(mode: str | None = None) -> bool: + return (mode if mode is not None else os.getenv("QSL_ROUTINE_NOTIFICATION_MODE", "immediate")) == "daily_digest" + + +def defer_routine_cycle_to_digest( + execution: Mapping[str, object], + *, + submitted_order_count: int, + mode: str | None = None, + submitted_orders: Sequence[Mapping[str, object]] | None = None, + include_orders: bool | None = None, +) -> bool: + """Defer healthy cycles and confirmed routine orders; urgent states stay live. + + Default ``QSL_ROUTINE_NOTIFICATION_MODE=immediate`` preserves existing + callers. An owner can opt a verified producer/recipient into ``daily_digest`` + after validating the existing private report-to-digest replacement path. + """ + if not daily_digest_enabled(mode) or type(submitted_order_count) is not int or submitted_order_count < 0: + return False + if not isinstance(execution, Mapping): + return False + if str(execution.get("execution_status") or "").lower() not in {"no_action", "no_op", "completed", "executed"}: + return False + if execution.get("no_op_reason") not in ( + None, "", "no_action", "no_signal", "no_rebalance", "below_threshold", + "rebalance_diff_below_threshold", "threshold_not_met", + ): + return False + if any(execution.get(key) for key in ( + "error", "errors", "fail_reason", "execution_blocked_reason", + "plugin_error", "persistence_error", "report_persistence_error", "notification_error", + "pending_reconciliation", "reconciliation_required", "market_data_retry_required", + "notification_attention_reason_codes", "orders_pending_count", + "account_new_risk_buy_blocked", "small_account_buy_blocked", "risk_flags", + "execution_blocked", "execution_blocking_skips", "strategy_run_persistence_error", + "strategy_plugin_error", "strategy_plugin_alert_error", "auth_expired", + "strategy_plugin_alert_failed_count", "strategy_plugin_alert_telegram_failed_count", + "data_error", "blocked_reason", "failed_order_count", + )): + # Only an explicit healthy no-order fact is eligible. Even a failed + # routine notification or unclassified no-op reason stays immediate. + return False + if str(execution.get("risk_gate") or "").upper() in {"BLOCK", "BLOCKED", "REJECT", "REJECTED", "DENY", "DENIED"}: + return False + if submitted_order_count: + orders_enabled = include_orders if include_orders is not None else os.getenv("QSL_DIGEST_ROUTINE_ORDERS", "true") == "true" + if not orders_enabled or submitted_orders is None or len(submitted_orders) != submitted_order_count: + return False + # A submission acknowledgement is not a completed fill. Pending, + # rejected and unknown outcomes retain the immediate channel. + if any(not isinstance(order, Mapping) or str(order.get("status") or "").lower() not in {"filled", "completed", "executed"} + for order in submitted_orders): + return False + elif execution.get("broker_submission_done"): + return False + return True + + +def immediate_plugin_signals(signals): + """Filter notification inputs only; callers retain original strategy signals.""" + if not daily_digest_enabled(): + return signals + return tuple(signal for signal in signals if not ( + str(getattr(signal, "canonical_route", "") or "").lower() in {"watch", "opportunity_watch", "panic_reversal", "taco_rebound"} + and str(getattr(signal, "suggested_action", "") or "").lower() in {"notify_manual_review", "watch", "observe"} + and not any(getattr(signal, key, None) for key in ("error", "errors", "risk_halt", "auth_expired")) + )) + + +def routine_notification(sender, runtime, report, text, *, routine_order=True): + """Route only explicitly classified routine call sites, never message text.""" + orders_enabled = os.getenv("QSL_DIGEST_ROUTINE_ORDERS", "true") == "true" + if daily_digest_enabled() and (not routine_order or orders_enabled): + report.setdefault("notifications", []).append({ + "delivery_status": "deferred_digest", "transport_acknowledged": False, + "sink": "telegram", "routine_order": routine_order, + }) + return False + return sender(runtime, report, text) diff --git a/scripts/emit_unified_digest_candidates.py b/scripts/emit_unified_digest_candidates.py new file mode 100644 index 0000000..ba1716c --- /dev/null +++ b/scripts/emit_unified_digest_candidates.py @@ -0,0 +1,8 @@ +"""Project existing private reports; no service or broker invocation.""" +from __future__ import annotations +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from notifications.digest_snapshot.digest_archive import main # noqa: E402 +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_unified_digest_candidates.py b/tests/test_unified_digest_candidates.py new file mode 100644 index 0000000..fa9af4f --- /dev/null +++ b/tests/test_unified_digest_candidates.py @@ -0,0 +1,107 @@ +from copy import deepcopy +from datetime import datetime, timezone +import pytest +from notifications.digest_snapshot.digest_candidates import cycle_evidence, project_reports + +TARGET = dict(platform_id="synthetic", strategy_profile="core", service_name="test-service", + account_scope="test-scope", account_selector=["test-selector"], deployment_selector="test-slot", dry_run_only=False) +NOW = datetime(2026, 10, 16, 2, tzinfo=timezone.utc) + + +def report(run_id="cycle-1", orders=()): + return dict(schema_version="runtime_report.v1", platform="synthetic", strategy_profile="core", + service_name="test-service", runtime_target=deepcopy(TARGET), dry_run=False, status="ok", run_id=run_id, + started_at="2026-10-16T01:00:00Z", summary={"notification_digest": cycle_evidence( + {"execution_status": "completed", "broker_submission_done": bool(orders)}, orders)}) + + +def project(reports): + return project_reports(reports, expected_target=TARGET, opaque_account_uid="synthetic-uid", + target_id="synthetic-target", now=NOW)["runs"] + + +def test_all_cycles_aggregate_without_inventing_fills(): + source = [report(), report("cycle-2", [{"status": "filled"}])] + original = deepcopy(source) + row = project(source)[0] + assert (row["order_count"], row["fill_count"], row["cycle_count"], row["status"]) == (1, 1, 2, "ok") + assert source == original + assert "account_scope" not in row + + +def test_missing_evidence_is_unknown_not_zero(): + source = report() + source["summary"] = {} + row = project([source])[0] + assert row["fill_count"] is None and row["order_count"] is None and row["status"] == "alert" + + +@pytest.mark.parametrize("mutation", ["identity", "future", "duplicate", "error"]) +def test_bad_source_stays_visible_or_rejects(mutation): + source = report() + if mutation == "error": + source["status"] = "error" + assert project([source])[0]["status"] == "alert" + return + if mutation == "identity": source["runtime_target"]["account_scope"] = "other" + if mutation == "future": source["started_at"] = "2026-10-17T00:00:00Z" + with pytest.raises(ValueError): + project([source, deepcopy(source)] if mutation == "duplicate" else [source]) + + +def test_archive_default_off_and_disabled_target_never_read_or_upload(): + from notifications.digest_snapshot.digest_archive import emit + def forbidden(*args): raise AssertionError("unexpected I/O") + assert emit({}, read=forbidden, upload=forbidden)["status"] == "disabled" + assert emit({"QSL_ROUTINE_NOTIFICATION_MODE": "daily_digest", "RUNTIME_TARGET_ENABLED": "false"}, + read=forbidden, upload=forbidden)["status"] == "target_disabled" + + +def test_archive_aggregates_window_and_uploads_once_without_private_output(): + import json + from notifications.digest_snapshot.digest_archive import emit + env = dict(QSL_ROUTINE_NOTIFICATION_MODE="daily_digest", RUNTIME_TARGET_ENABLED="true", + RUNTIME_TARGET_JSON=json.dumps(TARGET), DIGEST_STORE_TARGET="synthetic-slot", + DIGEST_OPAQUE_ACCOUNT_UID="synthetic-uid", DIGEST_TARGET_ID="synthetic-target", + EXECUTION_REPORT_GCS_URI="gs://synthetic-reports/archive", + DIGEST_CANDIDATES_GCS_URI="gs://synthetic-reports/candidates") + reports = [report("first", [{"status": "filled"}]), report("second")] + for source in reports: source["started_at"] = "2026-10-15T20:00:00Z" + def read(args): + if args[0] == "ls": + return [{"url": f"gs://synthetic-reports/archive/synthetic/core/2026-10/{i}.json", + "metadata": {"updated": "2026-10-15T20:01:00Z"}} for i in range(2)] + return reports[int(args[1].rsplit("/", 1)[-1].split(".")[0])] + uploads = [] + result = emit(env, now=NOW, read=read, upload=lambda *args: uploads.append(args)) + assert result["runs"] == 2 and len(uploads) == 1 + assert "/2026-10-15/" in uploads[0][0] + row = json.loads(uploads[0][1])["runs"][0] + assert (row["order_count"], row["fill_count"], row["cycle_count"]) == (1, 1, 2) + calls = [] + def failed_upload(*args): + calls.append(args) + raise RuntimeError("unknown write outcome") + with pytest.raises(RuntimeError): emit(env, now=NOW, read=read, upload=failed_upload) + assert len(calls) == 1 + + +def test_configured_closed_session_proves_idle_without_inventing_runs(monkeypatch): + import json + import quant_platform_kit.common.runtime_heartbeat_policy as policy + from notifications.digest_snapshot.digest_archive import emit + env = dict(QSL_ROUTINE_NOTIFICATION_MODE="daily_digest", RUNTIME_TARGET_ENABLED="true", + RUNTIME_TARGET_JSON=json.dumps(TARGET), DIGEST_STORE_TARGET="synthetic-slot", + DIGEST_OPAQUE_ACCOUNT_UID="synthetic-uid", DIGEST_TARGET_ID="synthetic-target", + EXECUTION_REPORT_GCS_URI="gs://synthetic-reports/archive", + DIGEST_CANDIDATES_GCS_URI="gs://synthetic-reports/candidates", + DIGEST_NOTIFICATION_SCHEDULE_JSON=json.dumps({"scheduler": {"main_time": "0 20 * * 1-5", "timezone": "UTC"}})) + monkeypatch.setattr(policy, "filter_due_targets", lambda *args, **kwargs: ([], True)) + uploads = [] + emit(env, now=NOW, read=lambda args: [], upload=lambda *args: uploads.append(args)) + row = json.loads(uploads[0][1])["runs"][0] + assert row["actually_ran"] is False and "order_count" not in row and "fill_count" not in row + monkeypatch.setattr(policy, "filter_due_targets", lambda *args, **kwargs: ([{}], True)) + with pytest.raises(ValueError, match="candidate_due_source_absent"): + emit(env, now=NOW, read=lambda args: [], upload=lambda *args: uploads.append(args)) + assert len(uploads) == 1 diff --git a/tests/test_unified_digest_routing.py b/tests/test_unified_digest_routing.py new file mode 100644 index 0000000..09b42b6 --- /dev/null +++ b/tests/test_unified_digest_routing.py @@ -0,0 +1,75 @@ +"""Offline routing tests: no broker, storage, or Telegram calls.""" + +import copy + +import pytest + +from notifications.digest_snapshot.digest_routing import defer_routine_cycle_to_digest + + +def test_default_off_and_explicit_opt_in(monkeypatch): + monkeypatch.delenv("QSL_ROUTINE_NOTIFICATION_MODE", raising=False) + evidence = {"execution_status": "no_action", "orders_pending_count": 0} + original = copy.deepcopy(evidence) + assert not defer_routine_cycle_to_digest(evidence, submitted_order_count=0) + monkeypatch.setenv("QSL_ROUTINE_NOTIFICATION_MODE", "daily_digest") + assert defer_routine_cycle_to_digest(evidence, submitted_order_count=0) + assert evidence == original + + +@pytest.mark.parametrize("status", ["blocked", "unknown", "pending", "pending_reconciliation", "rejected", "", None]) +def test_urgent_or_unknown_results_are_never_deferred(status): + assert not defer_routine_cycle_to_digest( + {"execution_status": status}, submitted_order_count=0, mode="daily_digest", + ) + + +@pytest.mark.parametrize("field,value", [ + ("notification_attention_reason_codes", ["DAILY_LOSS_LIMIT_EXCEEDED"]), + ("errors", ["storage_failed"]), ("orders_pending_count", 1), + ("reconciliation_required", True), ("broker_submission_done", True), + ("no_op_reason", "RECONCILE_ONLY"), ("notification_error", "unknown_result"), +]) +def test_healthy_label_cannot_hide_guard_or_operational_evidence(field, value): + assert not defer_routine_cycle_to_digest( + {"execution_status": "no_action", field: value}, + submitted_order_count=0, mode="daily_digest", + ) + + +@pytest.mark.parametrize("count", [1, None, False, -1]) +def test_any_order_or_unknown_order_count_remains_immediate(count): + assert not defer_routine_cycle_to_digest( + {"execution_status": "completed"}, submitted_order_count=count, mode="daily_digest", + ) + + +def test_invalid_mode_preserves_immediate_delivery(): + assert not defer_routine_cycle_to_digest( + {"execution_status": "no_action"}, submitted_order_count=0, mode="typo", + ) + + +def test_below_threshold_is_routine_but_an_unclassified_reason_stays_immediate(): + evidence = {"execution_status": "no_op", "no_op_reason": "rebalance_diff_below_threshold"} + assert defer_routine_cycle_to_digest(evidence, submitted_order_count=0, mode="daily_digest") + evidence["no_op_reason"] = "unknown_order_result" + assert not defer_routine_cycle_to_digest(evidence, submitted_order_count=0, mode="daily_digest") + + +def test_confirmed_routine_orders_are_configurable_and_do_not_mutate(monkeypatch): + evidence = {"execution_status": "completed", "broker_submission_done": True} + orders = [{"status": "filled"}] + original = copy.deepcopy((evidence, orders)) + assert defer_routine_cycle_to_digest(evidence, submitted_order_count=1, + submitted_orders=orders, mode="daily_digest") + monkeypatch.setenv("QSL_DIGEST_ROUTINE_ORDERS", "false") + assert not defer_routine_cycle_to_digest(evidence, submitted_order_count=1, + submitted_orders=orders, mode="daily_digest") + assert (evidence, orders) == original + + +@pytest.mark.parametrize("status", ["pending", "submitted", "rejected", "unknown", "partially_filled", ""]) +def test_unresolved_orders_never_become_routine(status): + assert not defer_routine_cycle_to_digest({"execution_status": "completed"}, + submitted_order_count=1, submitted_orders=[{"status": status}], mode="daily_digest") diff --git a/tests/test_unified_notification_wiring.py b/tests/test_unified_notification_wiring.py new file mode 100644 index 0000000..9e4ae46 --- /dev/null +++ b/tests/test_unified_notification_wiring.py @@ -0,0 +1,16 @@ +from types import SimpleNamespace + +from application.rebalance_service import _defer_cycle_notification + + +def test_only_healthy_confirmed_cycles_defer_and_pending_remains_live(monkeypatch): + monkeypatch.setenv("QSL_ROUTINE_NOTIFICATION_MODE", "daily_digest") + result = SimpleNamespace(allocation={}, execution={"execution_status": "completed"}, + action_done=True, pending_orders=(), submitted_orders=({"status": "filled"},)) + config = SimpleNamespace(dry_run_only=False) + assert _defer_cycle_notification(result, config=config) + result.pending_orders = ({"status": "pending_reconciliation"},) + assert not _defer_cycle_notification(result, config=config) + result.pending_orders = () + result.execution["report_persistence_error"] = "synthetic_failure" + assert not _defer_cycle_notification(result, config=config) From 0b6aee2a96681b93b60b49271448309d04c74df9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 16 Oct 2026 06:44:18 +0800 Subject: [PATCH 2/3] fix(notifications): honor per-target enablement and preserve protected JSON boundary Co-Authored-By: Codex --- docs/unified-daily-notification.md | 2 +- docs/unified-notification-trading-diff-review.zh-CN.md | 4 ++-- notifications/digest_snapshot/digest_archive.py | 8 ++++---- tests/test_unified_digest_candidates.py | 10 ++++++++++ 4 files changed, 17 insertions(+), 7 deletions(-) diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index 971cf76..1835e3d 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -6,7 +6,7 @@ immediate confirmed order notices. Failed, pending, uncertain, risk, persistence and authentication outcomes retain immediate alerts. Healthy idle cycles and routine plugin observations do not produce individual pushes. -QPK source reference: `b0d10862cfc0ffe9e1b868861b937a9cac6e7201` (notification modules only). +QPK source reference: `783bcc464cef89abd05b63e765cad00490b012c5` (notification modules only). The notification-only QPK source snapshot in `notifications/digest_snapshot/` avoids an unrelated trading dependency upgrade. Keep these three modules byte-identical diff --git a/docs/unified-notification-trading-diff-review.zh-CN.md b/docs/unified-notification-trading-diff-review.zh-CN.md index 2703bcb..50b5acb 100644 --- a/docs/unified-notification-trading-diff-review.zh-CN.md +++ b/docs/unified-notification-trading-diff-review.zh-CN.md @@ -14,12 +14,12 @@ 新增模块不调用券商。只有既有云监测器在 `daily_digest` 且目标明确启用时，才读已有私有报告并写候选；不启用账户、不修改订单或交易排程。关闭模式保持既有行为。失败、未决订单、风险、认证及持久化问题保留即时通知；已确认的普通订单可通过 `QSL_DIGEST_ROUTINE_ORDERS=false` 继续即时发送。 -通知代码取自 QPK `b0d10862cfc0ffe9e1b868861b937a9cac6e7201`，有意不升级平台交易依赖。三个源文件逐字节相同： +通知代码取自 QPK `783bcc464cef89abd05b63e765cad00490b012c5`，有意不升级平台交易依赖。三个源文件逐字节相同： | 文件 | SHA256 | |---|---| | `digest_routing.py` | `60c3f258c4d1a03bc49c7d03b392c02c316541726dfacb0b1d4fae6dbd2e5017` | | `digest_candidates.py` | `a6a0d10f90a17906c45393a84e819c8d71f430300e977fdb4fa1f824e26878f5` | -| `digest_archive.py` | `9a316072dc9e70ffb03c585c8c2f49876771ea5d722624fb5263ff037b27e3bc` | +| `digest_archive.py` | `b9a92299adfddff42c62f7d493845e0cbde91df7e475a5cb6bf2114446045b49` | 合并条件：相关验证通过、PR 当前精确 head 的 CI 全绿、当前远端主线的 push / workflow_run / 可复用 workflow 调用链无自动部署。否则保留草稿。此审查不批准从旧生产 revision 升级到整个当前主线；生产到候选的完整差异、UES 指纹与 baseline 仍须按后续发布方案单独审查。 diff --git a/notifications/digest_snapshot/digest_archive.py b/notifications/digest_snapshot/digest_archive.py index aa1b0b1..436fd2a 100644 --- a/notifications/digest_snapshot/digest_archive.py +++ b/notifications/digest_snapshot/digest_archive.py @@ -27,10 +27,6 @@ def _read(args): def emit(env, *, now=None, read=_read, upload=None): if env.get("QSL_ROUTINE_NOTIFICATION_MODE", "immediate") != "daily_digest": return {"status": "disabled", "uploaded": False} - if env.get("RUNTIME_TARGET_ENABLED") == "false": - return {"status": "target_disabled", "uploaded": False} - if env.get("RUNTIME_TARGET_ENABLED") != "true": - raise ValueError("candidate_target_enablement_unknown") bindings = env.get("DIGEST_TARGET_BINDINGS_JSON", "").strip() if bindings: values = json.loads(bindings) @@ -40,6 +36,10 @@ def emit(env, *, now=None, read=_read, upload=None): for value in values: results.append(emit({**env, **value, "DIGEST_TARGET_BINDINGS_JSON": ""}, now=now, read=read, upload=upload)) return {"status": "projected", "uploaded": any(v["uploaded"] for v in results), "targets": len(results)} + if env.get("RUNTIME_TARGET_ENABLED") == "false": + return {"status": "target_disabled", "uploaded": False} + if env.get("RUNTIME_TARGET_ENABLED") != "true": + raise ValueError("candidate_target_enablement_unknown") target = json.loads(env.get("RUNTIME_TARGET_JSON", "")) platform = target.get("platform_id", "") slot = env.get("DIGEST_STORE_TARGET", "") diff --git a/tests/test_unified_digest_candidates.py b/tests/test_unified_digest_candidates.py index fa9af4f..714e526 100644 --- a/tests/test_unified_digest_candidates.py +++ b/tests/test_unified_digest_candidates.py @@ -105,3 +105,13 @@ def test_configured_closed_session_proves_idle_without_inventing_runs(monkeypatc with pytest.raises(ValueError, match="candidate_due_source_absent"): emit(env, now=NOW, read=lambda args: [], upload=lambda *args: uploads.append(args)) assert len(uploads) == 1 + + +def test_multi_target_bindings_use_each_explicit_enablement_without_a_global_target(): + import json + from notifications.digest_snapshot.digest_archive import emit + result = emit({"QSL_ROUTINE_NOTIFICATION_MODE": "daily_digest", + "DIGEST_TARGET_BINDINGS_JSON": json.dumps([{"RUNTIME_TARGET_ENABLED": "false"}])}, + read=lambda args: pytest.fail("disabled target must not be read"), + upload=lambda *args: pytest.fail("disabled target must not be uploaded")) + assert result == {"status": "projected", "uploaded": False, "targets": 1} From 97968002ef4d84e178ccd7d19bc291f79f28464e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 16 Oct 2026 06:48:30 +0800 Subject: [PATCH 3/3] fix(notifications): keep execution fill counts distinct from completed orders Co-Authored-By: Codex --- docs/unified-daily-notification.md | 6 ++++- ...-notification-trading-diff-review.zh-CN.md | 4 +-- .../digest_snapshot/digest_candidates.py | 26 +++++++++++++++---- tests/test_unified_digest_candidates.py | 13 ++++++++-- 4 files changed, 39 insertions(+), 10 deletions(-) diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index 1835e3d..f9ef19b 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -6,7 +6,7 @@ immediate confirmed order notices. Failed, pending, uncertain, risk, persistence and authentication outcomes retain immediate alerts. Healthy idle cycles and routine plugin observations do not produce individual pushes. -QPK source reference: `783bcc464cef89abd05b63e765cad00490b012c5` (notification modules only). +QPK source reference: `4cc5e55251269923771f90cd661bb62a5cfdf0cd` (notification modules only). The notification-only QPK source snapshot in `notifications/digest_snapshot/` avoids an unrelated trading dependency upgrade. Keep these three modules byte-identical @@ -30,3 +30,7 @@ list of per-target environment mappings. `DIGEST_NOTIFICATION_SCHEDULE_JSON` can reuse the existing heartbeat scheduler/calendar declaration to prove a window was not due. Missing due reports stay errors; closed configured sessions produce an `actually_ran=false` marker without inventing zero trading counts. + +Fill counts require individual execution identifiers. A completed order response +proves an order completed, not how many fills it received. Missing fill counts +remain unknown; confirmed routine orders still appear without a false diagnostic. diff --git a/docs/unified-notification-trading-diff-review.zh-CN.md b/docs/unified-notification-trading-diff-review.zh-CN.md index 50b5acb..e1d0010 100644 --- a/docs/unified-notification-trading-diff-review.zh-CN.md +++ b/docs/unified-notification-trading-diff-review.zh-CN.md @@ -14,12 +14,12 @@ 新增模块不调用券商。只有既有云监测器在 `daily_digest` 且目标明确启用时，才读已有私有报告并写候选；不启用账户、不修改订单或交易排程。关闭模式保持既有行为。失败、未决订单、风险、认证及持久化问题保留即时通知；已确认的普通订单可通过 `QSL_DIGEST_ROUTINE_ORDERS=false` 继续即时发送。 -通知代码取自 QPK `783bcc464cef89abd05b63e765cad00490b012c5`，有意不升级平台交易依赖。三个源文件逐字节相同： +通知代码取自 QPK `4cc5e55251269923771f90cd661bb62a5cfdf0cd`，有意不升级平台交易依赖。三个源文件逐字节相同： | 文件 | SHA256 | |---|---| | `digest_routing.py` | `60c3f258c4d1a03bc49c7d03b392c02c316541726dfacb0b1d4fae6dbd2e5017` | -| `digest_candidates.py` | `a6a0d10f90a17906c45393a84e819c8d71f430300e977fdb4fa1f824e26878f5` | +| `digest_candidates.py` | `6272c0ca16ee63eecd9bcb25870df7f575d8fac3f487d3ab1ac69cc3831042a0` | | `digest_archive.py` | `b9a92299adfddff42c62f7d493845e0cbde91df7e475a5cb6bf2114446045b49` | 合并条件：相关验证通过、PR 当前精确 head 的 CI 全绿、当前远端主线的 push / workflow_run / 可复用 workflow 调用链无自动部署。否则保留草稿。此审查不批准从旧生产 revision 升级到整个当前主线；生产到候选的完整差异、UES 指纹与 baseline 仍须按后续发布方案单独审查。 diff --git a/notifications/digest_snapshot/digest_candidates.py b/notifications/digest_snapshot/digest_candidates.py index 997fd53..ee086ff 100644 --- a/notifications/digest_snapshot/digest_candidates.py +++ b/notifications/digest_snapshot/digest_candidates.py @@ -19,10 +19,24 @@ def cycle_evidence(execution: Mapping, orders: Sequence[Mapping], *, dry_run: bo confirmed = valid and all(str(order["status"]).lower() in {"filled", "completed", "executed"} for order in orders) healthy = defer_routine_cycle_to_digest(execution, submitted_order_count=count, submitted_orders=orders, mode="daily_digest", include_orders=True) + fill_ids = [] + fills_known = confirmed and healthy + for order in orders: + records = order.get("fills") if isinstance(order, Mapping) else None + if not isinstance(records, (list, tuple)) or not records: + fills_known = False + break + for record in records: + identity = next((record.get(k) for k in ("execution_id", "execId", "tradeId") if record.get(k) is not None), None) if isinstance(record, Mapping) else None + if identity is None: + fills_known = False + break + fill_ids.append(str(identity)) + fills_known = fills_known and len(fill_ids) == len(set(fill_ids)) return { "schema_version": "qsl.notification_cycle.v1", "order_count": count if valid and not dry_run else None, - "fill_count": count if confirmed and not dry_run else None, + "fill_count": len(fill_ids) if fills_known and not dry_run else None, "routine": healthy and not dry_run, "dry_run": dry_run, } @@ -65,8 +79,8 @@ def project_reports(reports: Sequence[Mapping], *, expected_target: Mapping, day = (stamp.astimezone(ZoneInfo("Asia/Singapore")) - timedelta(hours=6)).date().isoformat() facts = (report.get("summary") or {}).get("notification_digest") or {} known = (facts.get("schema_version") == "qsl.notification_cycle.v1" and facts.get("dry_run") is False - and all(type(facts.get(k)) is int and facts[k] >= 0 for k in ("order_count", "fill_count")) - and facts["fill_count"] <= facts["order_count"]) + and type(facts.get("order_count")) is int and facts["order_count"] >= 0 + and (facts.get("fill_count") is None or (type(facts["fill_count"]) is int and facts["fill_count"] >= 0))) error = report.get("status") in {"error", "failed", "aborted"} or bool((report.get("error_summary") or {}).get("errors")) alert = error or not known or facts.get("routine") is not True row = rows.setdefault(day, { @@ -79,8 +93,10 @@ def project_reports(reports: Sequence[Mapping], *, expected_target: Mapping, row["cycle_count"] += 1 row["as_of"] = max(row["as_of"], stamp.isoformat()) for key in ("fill_count", "order_count"): - row[key] = row[key] + facts[key] if known and row[key] is not None else None + row[key] = row[key] + facts[key] if known and type(facts.get(key)) is int and row[key] is not None else None row["field_status"][key] = "known" if row[key] is not None else "counts_unknown" + if not alert and row["status"] == "ok": + row["execution_status"] = "executed" if row["order_count"] else "no_action" if alert: row.update(status="alert", reason_code="source_cycle_alert" if error else "execution_evidence_missing_or_attention", execution_status="execution_alert" if error else "execution_unknown") @@ -99,7 +115,7 @@ def summary_evidence(summary: Mapping, *, dry_run: bool = False) -> dict: and not summary.get("option_orders_pending") and not summary.get("orders_partially_filled") and not summary.get("option_orders_partially_filled")): # These collections are populated by the existing broker reconciliation. - facts = [{"status": "filled"} for _ in filled] + facts = [{**order, "status": "filled"} for order in filled] if all(isinstance(order, Mapping) for order in filled) else orders else: facts = orders return cycle_evidence(evidence, facts, dry_run=dry_run) diff --git a/tests/test_unified_digest_candidates.py b/tests/test_unified_digest_candidates.py index 714e526..e880a27 100644 --- a/tests/test_unified_digest_candidates.py +++ b/tests/test_unified_digest_candidates.py @@ -24,7 +24,7 @@ def test_all_cycles_aggregate_without_inventing_fills(): source = [report(), report("cycle-2", [{"status": "filled"}])] original = deepcopy(source) row = project(source)[0] - assert (row["order_count"], row["fill_count"], row["cycle_count"], row["status"]) == (1, 1, 2, "ok") + assert (row["order_count"], row["fill_count"], row["cycle_count"], row["status"]) == (1, None, 2, "ok") assert source == original assert "account_scope" not in row @@ -77,7 +77,7 @@ def read(args): assert result["runs"] == 2 and len(uploads) == 1 assert "/2026-10-15/" in uploads[0][0] row = json.loads(uploads[0][1])["runs"][0] - assert (row["order_count"], row["fill_count"], row["cycle_count"]) == (1, 1, 2) + assert (row["order_count"], row["fill_count"], row["cycle_count"]) == (1, None, 2) calls = [] def failed_upload(*args): calls.append(args) @@ -115,3 +115,12 @@ def test_multi_target_bindings_use_each_explicit_enablement_without_a_global_tar read=lambda args: pytest.fail("disabled target must not be read"), upload=lambda *args: pytest.fail("disabled target must not be uploaded")) assert result == {"status": "projected", "uploaded": False, "targets": 1} + + +def test_completed_order_count_is_not_used_as_execution_fill_count(): + source = report(orders=[{"status": "filled"}]) + row = project([source])[0] + assert row["order_count"] == 1 and row["fill_count"] is None and row["status"] == "ok" + fills = [{"execution_id": f"synthetic-fill-{i}"} for i in range(3)] + row = project([report(orders=[{"status": "filled", "fills": fills}])])[0] + assert row["order_count"] == 1 and row["fill_count"] == 3