diff --git a/docs/README.md b/docs/README.md index 81c64bb3..c017f9ca 100644 --- a/docs/README.md +++ b/docs/README.md @@ -18,6 +18,7 @@ Read documents at the library revision used by your consumer. These are source-l - [Platform notification outcomes](platform-notification-outcomes.md): credential-free intents, receipts and log redaction. +- [Unified operator format](unified-operator-format.md): five sections, count semantics and display-only rollout. - [Notification routing policy](notification-routing-policy.md): durable daily facts and urgent incidents. - [Alpaca public market data collection](alpaca-public-market-data.md): manual, read-only workflow and its secrets. - [Platform repository boundaries](platform-repo-boundaries.md) ([中文](platform-repo-boundaries.zh-CN.md)). diff --git a/docs/unified-operator-format.md b/docs/unified-operator-format.md new file mode 100644 index 00000000..3c26aa17 --- /dev/null +++ b/docs/unified-operator-format.md @@ -0,0 +1,64 @@ +# Unified operator notification format + +This display contract extends [routine delivery](routine-daily-digest.md), +[notification routing](notification-routing-policy.md) and the existing +`NotificationEvent`, digest candidate and execution-receipt contracts. It does +not replace their schemas, classification, accounting or delivery policy. + +Every strategy-risk alert, daily trading summary, run-guard alert, +execution-receipt check and candidate-confirmation warning uses one locale and +five sections, in this order. `notifications.presentation` owns the section +names and the pure renderer; QRS consumes a byte-verified snapshot so its digest +image does not need another runtime dependency. + +| Section | Existing facts and display rule | +| --- | --- | +| Result / 结果 | Caller-proven outcome or review requirement. Keep `reason_code` with its existing localized explanation. A guard failure is not a broker outcome. | +| Scope / 范围 | Supplied `occurred_at`, `observed_at`, `due_at`, lookback or business window. A missing time is unknown. Never substitute render time. | +| Evidence / 证据 | Actual run link, risk observations, receipt categories, `order_count`, `fill_count`, known state and per-run diagnostics. Urgent diagnostics precede optional snapshots. | +| Coverage / 覆盖 | The scope actually inspected, expected/covered target counts and source status. Daily reports always name IBKR, Schwab, Firstrade, LongBridge, Binance and management console. | +| Next step / 下一步 | Review the source run, receipt and missing sources when evidence needs attention; retain execution restrictions. A verified routine window needs no action. Existing risk-specific next steps remain more specific. | + +## Field semantics + +- Keep `platform_id`, `opaque_account_uid`, `target_id`, `strategy_profile` and + `source_run_id` as existing protected identity fields. Never derive missing + identities, run links or account names from each other. +- `order_count` and `fill_count` retain their independent known/unknown status. + Zero requires affirmative count evidence. Accepted receipts, inspected report + counts, skipped reports, workflow success and explicit non-execution do not + create zero orders or fills. Missing category counts remain unknown. +- Whole-window totals cover only the existing bound targets and require their + existing complete evidence gate. No data connection is shown as unknown; + an explicitly unbound source is not included, with run state unknown. +- `actually_ran=false` remains non-execution evidence. Unknown, missing, stale, + invalid and read failure states remain distinct in existing diagnostics. +- A candidate warning establishes unconfirmed persistence only. It does not + establish a failed trade, successful daily inclusion or delivery receipt. +- Preserve the producer's window. QRS consolidated display uses its existing + half-open `[D 06:00, D+1 06:00)` Singapore window. Formatters do not reinterpret + legacy date contracts or move observations across windows. +- Links must be caller-supplied HTTPS URLs. Drop query and fragment; unavailable + or invalid links are explicitly unavailable. Private storage URIs, account + identifiers and raw exception payloads are not link fallbacks. + +## Compatibility and rollout + +Existing QPK formatter signatures remain compatible. Risk, guard and receipt +message copy changes to these sections when consumers adopt this revision. +Healthy heartbeat routing and its existing renderer remain unchanged. Daily +idle copy now says “No trades in verified targets” inside a complete report, +not “No trades across all accounts today”. Unknown unbound sources still appear +in coverage. This is a deliberate display change, not a new data connection. + +The renderer reserves result, scope, coverage and next step before truncating +optional evidence. It reports omitted evidence explicitly. Normal production +uses the existing 3900 UTF-16-unit single-message budget. An artificial budget +that cannot hold the mandatory fields raises a fixed error instead of silently +omitting coverage. No extra send is created. + +Downstream source adoption must be coordinated with platform fault fixes. +QPK consumers remain at their current pins until a separately reviewed adoption; +QRS may snapshot only the pure presentation module with recorded source bytes. +Source PRs do not change runtime flags, schedules, recipient/date reservations, +receipt acceptance, deployment or execution authority. diff --git a/src/quant_platform_kit/common/operational_notification_localization.py b/src/quant_platform_kit/common/operational_notification_localization.py index 7f070ea5..4a20842f 100644 --- a/src/quant_platform_kit/common/operational_notification_localization.py +++ b/src/quant_platform_kit/common/operational_notification_localization.py @@ -78,6 +78,9 @@ "attention_reason_drift_watch": "生产偏离观察（WATCH）", "attention_reason_drift_status_unknown": "生产偏离状态未知", "attention_reason_operational_uncertain": "运行状态不确定 / 故障", + "attention_reason_blocked_run": "本轮已运行，执行被阻断", + "attention_reason_strategy_risk_rejected": "本轮策略风控未通过", + "attention_reason_empty_position_decision": "本轮策略目标仓位为空", "attention_reason_hard_park": "硬停（需授权恢复）", "attention_reason_mandate_dd_exhausted": "授权回撤额度已用尽", "attention_reason_mandate_dd_elevated": "授权回撤接近上限", @@ -153,6 +156,9 @@ "attention_reason_drift_watch": "production drift watch", "attention_reason_drift_status_unknown": "production drift status unknown", "attention_reason_operational_uncertain": "operational uncertainty / fault", + "attention_reason_blocked_run": "Run observed; execution blocked", + "attention_reason_strategy_risk_rejected": "Strategy risk checks rejected this run", + "attention_reason_empty_position_decision": "Strategy target positions are empty for this run", "attention_reason_hard_park": "hard park (authorized recovery required)", "attention_reason_mandate_dd_exhausted": "mandate drawdown budget exhausted", "attention_reason_mandate_dd_elevated": "mandate drawdown elevated", @@ -238,31 +244,36 @@ def format_operational_alert( ``recent_reports`` retains its existing labelled receipt-summary format. """ + from quant_platform_kit.notifications.presentation import ( + next_step_text, render_operator_message, safe_run_url, + ) + + lang = resolve_operational_notification_locale(locale) + zh = lang == "zh" title_key = f"{str(alert_type).strip()}_title" - lines = [operational_notification_text(locale, title_key, name=str(name or "runtime"))] - for key, value in (context or {}).items(): - lines.append(operational_notification_text(locale, str(key), value=value)) - if issues: - lines.extend([operational_notification_text(locale, "issues"), *(f"- {item}" for item in issues)]) + title = operational_notification_text(locale, title_key, name=str(name or "runtime")) + scope = [operational_notification_text(locale, str(key), value=value) + for key, value in (context or {}).items()] + evidence = [operational_notification_text(locale, "issues"), *(f"- {item}" for item in issues)] if issues else [] if recent_reports: - lines.extend( - [ - operational_notification_text(locale, "recent_reports"), - *(str(item) for item in recent_reports), - ] - ) + evidence.extend([operational_notification_text(locale, "recent_reports"), *map(str, recent_reports)]) if latest_runtime_run: - lines.extend( - [ - operational_notification_text(locale, "latest_runtime_run"), - *(str(item) for item in latest_runtime_run), - ] - ) + evidence.extend([operational_notification_text(locale, "latest_runtime_run"), *map(str, latest_runtime_run)]) if technical_details: - lines.append(operational_notification_text(locale, "technical_details_in_logs")) - if workflow_url: - lines.append(operational_notification_text(locale, "workflow", value=str(workflow_url))) - return "\n".join(line for line in lines if line) + evidence.append(operational_notification_text(locale, "technical_details_in_logs")) + link = safe_run_url(workflow_url) + evidence.append(operational_notification_text(locale, "workflow", value=link) if link else + ("运行记录：未提供运行链接" if zh else "Run record: run link not provided")) + return render_operator_message( + title=title, locale=lang, + result=["需核对；告警不代表订单或成交结果" if zh else + "Needs review; an alert does not establish order or fill outcomes"], + scope=scope, + evidence=evidence, + coverage=["范围仅限本次检查；订单：未知；成交：未知" if zh else + "This check only; orders: unknown; fills: unknown"], + next_step=[next_step_text(lang, needs_review=True)], + ) def format_operational_heartbeat_status( @@ -380,4 +391,20 @@ def format_receipt_alert( # Links are supplied by the caller, never reconstructed from account IDs. safe_url = urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", "")) lines.append(text["run"].format(value=safe_url)) - return "\n".join(lines) + from quant_platform_kit.notifications.presentation import next_step_text, render_operator_message + + # A category count is a count of inspected reports, never an order/fill total. + attention = not categories or any( + category not in {"accepted", "skipped_before_due", "skipped_outside_window"} + and (count is None or count > 0) + for category, count in categories.items() + ) or categories.get("accepted") in (None, 0) + return render_operator_message( + title=lines[0], locale=normalized, + result=[("回执证据需核对" if attention else "已观察到符合要求的回执") if normalized == "zh" else + ("Receipt evidence needs review" if attention else "Acceptable receipt observed")], + scope=lines[1:3], evidence=lines[3:], + coverage=["仅列已检查回执；成交：未知" if normalized == "zh" else + "Inspected receipts only; fills: unknown"], + next_step=[next_step_text(normalized, needs_review=attention)], + ) diff --git a/src/quant_platform_kit/notifications/presentation.py b/src/quant_platform_kit/notifications/presentation.py new file mode 100644 index 00000000..b73178d9 --- /dev/null +++ b/src/quant_platform_kit/notifications/presentation.py @@ -0,0 +1,83 @@ +"""Shared display sections; no execution, routing, receipt or transport policy.""" +from __future__ import annotations + +from collections.abc import Sequence +from urllib.parse import urlsplit, urlunsplit + +SECTION_NAMES = { + "zh": ("结果", "范围", "证据", "覆盖", "下一步"), + "en": ("Result", "Scope", "Evidence", "Coverage", "Next step"), +} + + +def safe_run_url(value: object) -> str | None: + """Use only caller-supplied HTTPS links, dropping query and fragment.""" + if not isinstance(value, str) or any(c.isspace() or ord(c) < 32 for c in value): + return None + try: + parsed = urlsplit(value) + if parsed.scheme == "https" and parsed.hostname and not parsed.username and not parsed.password: + return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", "")) + except ValueError: + pass + return None + + +def next_step_text(locale: str, *, needs_review: bool) -> str: + if locale == "zh": + return ("核对该次运行、回执和缺失来源；保持现有执行限制。" if needs_review else + "无需操作；等待下一统计窗口。") + return ("Review the source run, receipt and missing sources; retain existing execution restrictions." + if needs_review else "No action required; await the next reporting window.") + + +def render_operator_message(*, title: str, locale: str, result: Sequence[str], + scope: Sequence[str], evidence: Sequence[str], coverage: Sequence[str], + next_step: Sequence[str], max_units: int | None = None, + required_evidence: Sequence[str] = ()) -> str: + """Keep result, scope, coverage and next step before optional evidence. + + Truncation is explicit and cannot remove a section or imply full coverage. + Callers put urgent evidence before optional observations. Financial facts + must already have been validated by the producer; this helper infers none. + """ + names = SECTION_NAMES["zh" if locale == "zh" else "en"] + unknown = "未知" if locale == "zh" else "Unknown" + groups = [[str(line) for line in group if line] or [unknown] + for group in (result, scope, evidence, coverage, next_step)] + + def assemble(): + lines = [title] + for name, group in zip(names, groups): + lines.extend([f"【{name}】" if locale == "zh" else f"[{name}]", *group]) + return "\n".join(lines) + + if max_units is None: + return assemble() + if max_units < 256: + raise ValueError("notification_budget_too_small") + omitted = 0 + original = [line for line in groups[2] if line not in required_evidence] + required = list(dict.fromkeys(required_evidence)) + groups[2] = required + original + complete = assemble() + if len(complete.encode("utf-16-le")) // 2 <= max_units: + return complete + suffix = "部分证据已省略；请查看运行记录。" if locale == "zh" else "Some evidence omitted; inspect the source run." + groups[2] = required + [suffix] + if len(assemble().encode("utf-16-le")) // 2 > max_units: + raise ValueError("notification_required_fields_exceed_budget") + groups[2] = required.copy() + for line in original: + groups[2].append(line) + groups[2].append(suffix) + fits = len(assemble().encode("utf-16-le")) // 2 <= max_units + groups[2].pop() + if not fits: + groups[2].pop() + omitted += 1 + if omitted: + groups[2].append(suffix) + if not groups[2]: + groups[2] = [unknown] + return assemble() diff --git a/src/quant_platform_kit/risk/attention.py b/src/quant_platform_kit/risk/attention.py index a4d385e8..2e87a5b3 100644 --- a/src/quant_platform_kit/risk/attention.py +++ b/src/quant_platform_kit/risk/attention.py @@ -246,7 +246,16 @@ def render_attention_compact( locale=lang, daily_loss_context=context, occurred_at=occurred_at, run_url=run_url, ) - return "\n".join((f"{emoji} {title}", result, reason_line, *details, next_step)) + from quant_platform_kit.notifications.presentation import render_operator_message + + return render_operator_message( + title=f"{emoji} {title}", locale=lang, result=[result], + scope=[line for line in details if line.startswith(("运行时间：", "Run time:"))], + evidence=[reason_line, *[line for line in details if not line.startswith(("运行时间：", "Run time:"))]], + coverage=["单次运行；订单：未知；成交：未知" if lang == "zh" else + "Single run; orders: unknown; fills: unknown"], + next_step=[next_step], + ) # Feature marker for pinned consumers that retain a legacy renderer until rollout. diff --git a/tests/test_notification_presentation.py b/tests/test_notification_presentation.py new file mode 100644 index 00000000..2e51d9c1 --- /dev/null +++ b/tests/test_notification_presentation.py @@ -0,0 +1,104 @@ +"""Display boundaries retain unknown counts and all mandatory sections.""" +from datetime import datetime, timezone + +import pytest + +from quant_platform_kit.common.operational_notification_localization import format_operational_alert, format_receipt_alert +from quant_platform_kit.notifications.presentation import render_operator_message, safe_run_url +from quant_platform_kit.risk.attention import AttentionAxes, evaluate_attention, render_attention_compact + + +@pytest.mark.parametrize("locale", ["zh", "en"]) +def test_risk_guard_and_receipt_share_sections_without_inventing_counts(locale): + messages = [ + format_operational_alert(locale=locale, alert_type="runtime_guard", name="Synthetic platform", + context={"lookback_minutes": 60}, issues=["Synthetic fault"]), + format_receipt_alert(locale=locale, categories={"strategy_risk_rejected": 1, "skipped_before_due": 7}, + observed_at=datetime(2026, 1, 2, tzinfo=timezone.utc)), + render_attention_compact(locale=locale, platform="ibkr", account_alias="", + strategy_profile="synthetic_core", + decision=evaluate_attention(AttentionAxes(operational_uncertain=True))), + ] + sections = ("结果", "范围", "证据", "覆盖", "下一步") if locale == "zh" else ("Result", "Scope", "Evidence", "Coverage", "Next step") + for message in messages: + positions = [message.index(f"【{section}】" if locale == "zh" else f"[{section}]") for section in sections] + assert positions == sorted(positions) + assert "未知" in message if locale == "zh" else "unknown" in message.lower() + assert "成交：0" not in message and "fills: 0" not in message + + +@pytest.mark.parametrize("raw", ["https://example.test/run/1?secret=hidden#hidden", "https://example.test/run/1"]) +def test_only_supplied_safe_link_path_is_kept(raw): + assert safe_run_url(raw) == "https://example.test/run/1" + + +@pytest.mark.parametrize("raw", [None, "gs://private/object", "https://a:b@example.test/run", "https://example.test/\nsecret", "https://[broken"]) +def test_invalid_links_are_unavailable(raw): + assert safe_run_url(raw) is None + + +def test_budget_keeps_coverage_and_next_step_and_reports_omission(): + text = render_operator_message(title="Synthetic", locale="en", result=["Needs review"], scope=["One run"], + evidence=["urgent issue", "optional " * 1000], coverage=["Unknown"], next_step=["Review source"], max_units=256) + assert "urgent issue" in text and "omitted" in text + assert "[Coverage]\nUnknown" in text and "[Next step]\nReview source" in text + assert len(text.encode("utf-16-le")) // 2 <= 256 + with pytest.raises(ValueError, match="required_fields_exceed_budget"): + render_operator_message(title="Synthetic", locale="en", result=["x" * 300], scope=[], evidence=[], + coverage=[], next_step=[], max_units=256) + + +def test_required_diagnostics_survive_before_optional_context(): + text = render_operator_message(title="Synthetic", locale="en", result=["Needs review"], scope=["One run"], + evidence=["optional " * 1000, "Persistence failed"], coverage=["Unknown"], next_step=["Review source"], + required_evidence=["Persistence failed"], max_units=256) + assert "Persistence failed" in text + assert text.count("Persistence failed") == 1 + assert "omitted" in text + + +@pytest.mark.parametrize("code", ["blocked_run", "strategy_risk_rejected", "empty_position_decision"]) +@pytest.mark.parametrize("locale", ["zh", "en"]) +def test_block_cause_codes_have_canonical_explanations(code, locale): + from quant_platform_kit.common.operational_notification_localization import localize_attention_reason_code + text = localize_attention_reason_code(locale, code) + assert text != code + assert "unknown" not in text.lower() + + +@pytest.mark.parametrize("locale", ["zh", "en"]) +@pytest.mark.parametrize("category", ["skipped_before_due", "skipped_outside_window"]) +@pytest.mark.parametrize("count", [3, None]) +def test_accepted_receipt_with_routine_skips_needs_no_action(locale, category, count): + text = format_receipt_alert(locale=locale, categories={"accepted": 1, category: count}) + assert ("无需操作" if locale == "zh" else "No action required") in text + assert ("已观察到符合要求的回执" if locale == "zh" else "Acceptable receipt observed") in text + + +@pytest.mark.parametrize("category", ["receipt_missing", "receipt_stale", "blocked_run", + "strategy_risk_rejected", "receipt_unacceptable", "auth_failure", "unknown_category"]) +@pytest.mark.parametrize("count", [1, None]) +def test_accepted_receipt_with_real_or_unknown_fault_still_needs_review(category, count): + text = format_receipt_alert(locale="en", categories={"accepted": 1, category: count}) + assert "Receipt evidence needs review" in text + + +@pytest.mark.parametrize("accepted", [None, 0]) +def test_routine_skips_without_accepted_receipt_still_need_review(accepted): + text = format_receipt_alert(locale="en", categories={"accepted": accepted, "skipped_before_due": 3}) + assert "Receipt evidence needs review" in text + + +@pytest.mark.parametrize("required", [False, True]) +@pytest.mark.parametrize("message_units", [246, 256]) +def test_complete_fitting_message_keeps_diagnostic_without_omission(required, message_units): + fields = dict(title="Diagnostic", locale="en", result=["Needs review"], scope=["One run"], + evidence=["Diagnostic: "], coverage=["Unknown"], next_step=["Review source"]) + empty = render_operator_message(**fields) + diagnostic = "Diagnostic: " + "x" * (message_units - len(empty.encode("utf-16-le")) // 2) + fields["evidence"] = [diagnostic] + if required: + fields["required_evidence"] = [diagnostic] + complete = render_operator_message(**fields) + assert len(complete.encode("utf-16-le")) // 2 == message_units + assert render_operator_message(**fields, max_units=256) == complete diff --git a/tests/test_unified_notification_legacy_equivalence.py b/tests/test_unified_notification_legacy_equivalence.py index 731f320b..9ff98bc5 100644 --- a/tests/test_unified_notification_legacy_equivalence.py +++ b/tests/test_unified_notification_legacy_equivalence.py @@ -1,4 +1,4 @@ -"""Retain historical golden output, allowing only the diagnostic-display change.""" +"""Retain routing golden evidence while migrating operator display sections.""" import json from datetime import datetime from pathlib import Path @@ -45,7 +45,7 @@ def sender(runtime, report, text): @pytest.mark.parametrize('locale', ['zh', 'en']) -def test_existing_renderers_and_publisher_preserve_content_except_raw_diagnostics(monkeypatch, locale): +def test_existing_renderers_keep_facts_and_publisher_routing_in_unified_sections(monkeypatch, locale): for flag in FLAGS: monkeypatch.delenv(flag, raising=False) message = format_operational_alert(locale=locale, alert_type='execution_report_heartbeat', name='runtime', @@ -59,14 +59,19 @@ def test_existing_renderers_and_publisher_preserve_content_except_raw_diagnostic }[locale] assert expected.count(original) == 1 expected = expected.replace(original, replacement, 1) - assert message.encode() == expected.encode() + # Copy migration changes layout, while every legacy non-raw fact remains. + for line in expected.splitlines(): + assert line in message + sections = ("结果", "范围", "证据", "覆盖", "下一步") if locale == "zh" else ("Result", "Scope", "Evidence", "Coverage", "Next step") + assert all(section in message for section in sections) + assert "original detail" not in message heartbeat = format_operational_heartbeat_status(locale=locale, name='runtime', detail='existing detail') assert heartbeat.encode() == GOLDEN['observations'][f'heartbeat:{locale}'].encode() calls = [] publisher = NotificationPublisher(log_message=lambda text: calls.append(('log', text)), send_message=lambda text: calls.append(('send', text)) or True) assert publisher.publish(RenderedNotification(message, heartbeat)) - assert calls == [('log', expected), ('send', heartbeat)] + assert calls == [('log', message), ('send', heartbeat)] @pytest.mark.parametrize('kind', ['real_error', 'auth_failure', 'order_status_unknown', 'receipt_missing',