From fd69fda0ba0115b50469bc6072988445abf67fe9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:14:27 +0800 Subject: [PATCH 1/3] feat: audit snapshot completeness and retain daily Binance detail Co-Authored-By: Codex --- docs/README.md | 1 + docs/snapshot-completeness.md | 147 +++++++ examples/snapshot-completeness.example.json | 134 +++++++ python/Dockerfile.snapshot-completeness | 6 + python/scripts/snapshot_completeness.py | 364 ++++++++++++++++++ python/tests/test_snapshot_completeness.py | 188 +++++++++ tests/binance_account_facts_validation.mjs | 29 +- .../binance_detail_archive.js | 44 +++ web/strategy-switch-console/worker.js | 25 ++ 9 files changed, 937 insertions(+), 1 deletion(-) create mode 100644 docs/snapshot-completeness.md create mode 100644 examples/snapshot-completeness.example.json create mode 100644 python/Dockerfile.snapshot-completeness create mode 100644 python/scripts/snapshot_completeness.py create mode 100644 python/tests/test_snapshot_completeness.py create mode 100644 web/strategy-switch-console/binance_detail_archive.js diff --git a/docs/README.md b/docs/README.md index 92df6757..c01d7cf0 100644 --- a/docs/README.md +++ b/docs/README.md @@ -20,6 +20,7 @@ The current package retains configuration and compatibility ownership. Broker cr - [English operator guide](../web/strategy-switch-console/README.md#reading-by-task) · [中文操作入口](../web/strategy-switch-console/README.zh-CN.md#按任务阅读): installation/configuration, interfaces and operations grouped without moving contract text. - [Console API](console-api.md): interface details. +- [Snapshot completeness](snapshot-completeness.md): cloud archive checks, daily Binance detail, and approval-bound activation. - [English fork guide](strategy-switch-fork-guide.md) · [中文 Fork 指南](strategy-switch-fork-guide.zh-CN.md): installation in a separately configured environment. - [English permission model](manual-strategy-switch-permission-control.md) · [中文权限说明](manual-strategy-switch-permission-control.zh-CN.md). diff --git a/docs/snapshot-completeness.md b/docs/snapshot-completeness.md new file mode 100644 index 00000000..c7aa7208 --- /dev/null +++ b/docs/snapshot-completeness.md @@ -0,0 +1,147 @@ +# Snapshot completeness + +`python/scripts/snapshot_completeness.py` audits private daily archives in a +Cloud Run Job. It checks the requested window independently of stored samples, +so an empty archive, a leading gap and a trailing outage remain visible. It does +not call a trading or runtime-control route. The CLI refuses to run outside a +Cloud Run Job; offline tests inject storage, history and notification clients. + +## Date and field contract + +Supply a private configuration file through `SNAPSHOT_COMPLETENESS_CONFIG`. +The [synthetic configuration example](../examples/snapshot-completeness.example.json) +shows the shape; its one-day calendar deliberately cannot pass a rolling +production audit. Replace it with verified private configuration before approval. +`window_days` defaults to 30 and is bounded to 366. The end is the previous UTC +date, preserving 30 complete dates rather than counting the unfinished current +day; only dates whose collection deadline has passed are expected. A continuous +UTC calendar is used for crypto. Exchange calendars must include `id`, `source`, +`timezone`, inclusive `start`/`end`, and the authoritative `sessions` list of +`date` and timezone-aware `close_at`. Set `grace_seconds` for the collector's +publication deadline. Published close instants carry holidays, early closes and +DST; the guard never substitutes weekdays for an unavailable exchange calendar. +Calendar coverage must span the requested window or the target is unknown. + +Account region does not identify its trading market. Select the calendar from +the broker account's actual covered instruments and collection contract. A +multi-market account needs its explicitly approved combined session calendar. +Session snapshots must be observed on that local session date after close. +Continuous-calendar snapshots retain their original UTC date. Use an actual +broker as-of timestamp for historical statements; do not relabel collection time. + +The protected `targets` inventory includes every account/binding to audit. +Platforms are `longbridge`, `firstrade`, `schwab`, `ibkr`, `binance`, `alpaca`, +`qmt` and `ftmo`. `status: active` requires a complete contract; other statuses +produce `not_configured`, never a green result. Omitted platforms are also +reported as `not_configured`. The operator must reconcile this inventory +with the protected account registry before activation. + +Each target declares: + +- `identity`: dotted payload paths and exact values, including schema, platform, + target and account/source-binding identity. These values stay out of Git and + public output. +- `as_of_path`: source observation or broker historical as-of timestamp. +- `archive`: a GCS `prefix` ending in `/`, with `{date}` or `{yyyy}/{mm}/{dd}`. + The guard lists bounded pages and reads each JSON object at its listed + generation. `payload_path` optionally selects a nested account snapshot. +- `fields`: rules for all four groups, `cash`, `positions`, `orders`, `fills`. + Each rule supplies `path`, `complete_path`, `complete_value`, nonempty + `required` row paths and `numeric` row paths. Missing values, nonfinite numbers, + unknown completeness and malformed rows fail. Actual broker-reported zero is + valid; the guard never creates a value. +- `empty_proof`: for a legitimately empty positions/orders/fills list, supply + `query_ok`, `row_count` and `pagination_complete` payload paths. Their values + must be exactly `true`, integer `0` and `true`, derived from the broker query. + An empty list plus a `complete` label alone fails. Empty cash is not a complete + cash snapshot. + +Existing platform archive shapes are selected through these explicit paths; +missing producer fields remain incomplete. Do not map an open-order-only list +to full orders, a strategy-only holding list to the whole account, or a wallet +valuation summary to asset detail. The latest observation is checked, so an old +complete record cannot hide a newer incomplete one. Access/identity failures +stop that target while other targets continue. + +## Binance daily detail + +The Worker now persists the entire validated `binance_account_facts.v1` report in +`binance_detail_daily`, keyed by account, source binding and UTC observation date. +The latest report for a day wins; older input and conflicting equal timestamps +are rejected. The existing DO transaction validates an immediate SQL readback +before the ingestion response can acknowledge success. This table has no TTL or +automatic deletion. The two-day latest KV cache and 366-day wallet valuation +projection keep their existing contracts. + +`GET /api/internal/binance-account-facts/archive?date=YYYY-MM-DD` uses the existing +dedicated Binance facts bearer authorization and trusted binding. It returns +the private raw daily report, `404` for absence, and `503` for unavailable or +invalid data. It is not a public/session API. A guard target uses +`archive: {kind: binance_detail, url: , token_env: }`. Keep this bearer credential within the approved +client. Sharing it with a new job requires a separate secret/access review. +The reader currently selects the active binding; retaining a retired binding +does not automatically authorize a new client to read it. + +This preserves only what the collector actually supplied: Spot and Flexible +Earn detail. Funding, Margin, Futures, Locked Earn, full orders and fills are +not invented. Existing expired KV data cannot be reconstructed from wallet +valuation history. + +## Backfill and alerts + +Backfill defaults off. With `backfill_enabled: true`, only `archive_missing` +may invoke a configured `history` hook. It must have +`kind: broker_readable_history`, an HTTPS `/history` or `/archive` URL, +`token_env` and `payload_path`. The method is always GET with a `date` query; +redirects are refused. Configure only a broker-readable historical statement +endpoint approved for this account. There is no shell hook, order call, current +snapshot fallback, zero fill or interpolation. + +The returned payload must pass the same identity, date and four-field checks. +A new `broker-history-backfill.json` is written with generation-match zero, +then independently listed/read and compared. Existing objects are never +overwritten. Nested-envelope backfill needs a producer adapter and remains +blocked; this prevents writing a payload the archive reader cannot consume. +An absent or incomplete historical source leaves the gap unresolved. A limited +recent-order response cannot reconstruct old cash or positions. + +Private reports go under `state_prefix`, using create-only writes and readback. +Telegram uses mounted `TELEGRAM_TOKEN` and `GLOBAL_TELEGRAM_CHAT_ID` values, never +Secret Manager discovery or log output. Alerts contain platform, ordinal target +number, dates and fixed reason codes; no account identifiers, resource names or +payload values. A per-day content claim precedes sending. A send with unknown +outcome is not retried automatically. An ACK means Telegram sender acceptance, +not recipient receipt. A configuration or report-storage failure attempts one +fixed failure alert without persistent deduplication; it never retries an +already attempted notification. Read/configuration failures exit nonzero; +Cloud Run job failure monitoring must remain enabled even if storage or Telegram +is down. + +## Cloud activation requires owner approval + +Build context is the repository root; the container definition is +`python/Dockerfile.snapshot-completeness`. No new Python package is needed. +This change creates no cloud resources, secrets, IAM bindings or schedules. +Do not add a Mac timer or execute a broker backfill during code validation. + +Before activation, submit the exact private target/calendar/field mapping, +image digest, configuration mount, archive/state prefixes, service accounts, +secret references and scheduler proposal to the owner. The proposed scheduler +is hourly at minute 17 in UTC, invoking one Cloud Run Job task with no automatic +job retries. Collection grace is per target. A dedicated runtime identity needs +object list/get on source prefixes and create/get on its state prefix; approved +backfill additionally needs create on the selected archive prefix. Scheduler +invocation and mounted-secret access are separate narrowly scoped grants. + +After approval, deploy the reviewed Worker and guard image, install the +protected configuration, and activate the cloud schedule. Verify a natural job +execution, private report readback, a known missing-day alert, Binance detail +readback after cache expiry, and a real-history-only recovery when available. +Report deployment, scheduler activation, sender ACK and recipient receipt +separately. Rollback pauses only the new scheduler and restores the prior +Worker/image; retain archived financial records. Any gcloud operator command +must include an explicit `--account` argument. + +Storage semantics follow [Cloud Storage generation preconditions](https://docs.cloud.google.com/storage/docs/request-preconditions) +and [Durable Object synchronous transactions](https://developers.cloudflare.com/durable-objects/api/sqlite-storage-api/#transactions). diff --git a/examples/snapshot-completeness.example.json b/examples/snapshot-completeness.example.json new file mode 100644 index 00000000..c2a64f3b --- /dev/null +++ b/examples/snapshot-completeness.example.json @@ -0,0 +1,134 @@ +{ + "window_days": 30, + "state_prefix": "gs://synthetic-private/audit", + "backfill_enabled": false, + "targets": [ + { + "platform": "longbridge", + "status": "active", + "identity": { + "schema": "synthetic.broker.v1", + "platform": "longbridge", + "account": "synthetic-account", + "binding": "synthetic-binding" + }, + "as_of_path": "observed_at", + "archive": { + "prefix": "gs://synthetic-private/archive/{date}/" + }, + "calendar": { + "kind": "sessions", + "id": "synthetic-exchange", + "source": "synthetic-example-not-for-production", + "timezone": "Asia/Hong_Kong", + "start": "2026-10-20", + "end": "2026-10-20", + "grace_seconds": 3600, + "sessions": [ + { + "date": "2026-10-20", + "close_at": "2026-10-20T16:00:00+08:00" + } + ] + }, + "fields": { + "cash": { + "path": "cash", + "complete_path": "coverage.cash", + "complete_value": "complete", + "required": [ + "asset", + "quantity" + ], + "numeric": [ + "quantity" + ], + "empty_proof": { + "query_ok": "proof.cash.query_ok", + "row_count": "proof.cash.row_count", + "pagination_complete": "proof.cash.pagination_complete" + } + }, + "positions": { + "path": "positions", + "complete_path": "coverage.positions", + "complete_value": "complete", + "required": [ + "asset", + "quantity" + ], + "numeric": [ + "quantity" + ], + "empty_proof": { + "query_ok": "proof.positions.query_ok", + "row_count": "proof.positions.row_count", + "pagination_complete": "proof.positions.pagination_complete" + } + }, + "orders": { + "path": "orders", + "complete_path": "coverage.orders", + "complete_value": "complete", + "required": [ + "asset", + "quantity" + ], + "numeric": [ + "quantity" + ], + "empty_proof": { + "query_ok": "proof.orders.query_ok", + "row_count": "proof.orders.row_count", + "pagination_complete": "proof.orders.pagination_complete" + } + }, + "fills": { + "path": "fills", + "complete_path": "coverage.fills", + "complete_value": "complete", + "required": [ + "asset", + "quantity" + ], + "numeric": [ + "quantity" + ], + "empty_proof": { + "query_ok": "proof.fills.query_ok", + "row_count": "proof.fills.row_count", + "pagination_complete": "proof.fills.pagination_complete" + } + } + } + }, + { + "platform": "firstrade", + "status": "unverified" + }, + { + "platform": "schwab", + "status": "unverified" + }, + { + "platform": "ibkr", + "status": "unverified" + }, + { + "platform": "binance", + "status": "unverified" + }, + { + "platform": "alpaca", + "status": "unverified" + }, + { + "platform": "qmt", + "status": "unverified" + }, + { + "platform": "ftmo", + "status": "unverified" + } + ] +} diff --git a/python/Dockerfile.snapshot-completeness b/python/Dockerfile.snapshot-completeness new file mode 100644 index 00000000..0f0c9929 --- /dev/null +++ b/python/Dockerfile.snapshot-completeness @@ -0,0 +1,6 @@ +FROM python:3.12-slim@sha256:e70237b264442328526ff29d6b3c43bd8abcf6cbab78a949322fb69b7d91d6b6 +ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 +WORKDIR /app +COPY python/scripts/snapshot_completeness.py /app/snapshot_completeness.py +USER 65532:65532 +CMD ["python", "/app/snapshot_completeness.py"] diff --git a/python/scripts/snapshot_completeness.py b/python/scripts/snapshot_completeness.py new file mode 100644 index 00000000..a8d7a647 --- /dev/null +++ b/python/scripts/snapshot_completeness.py @@ -0,0 +1,364 @@ +"""Cloud-only archive audit. Configuration and source payloads stay private.""" +from __future__ import annotations + +import hashlib +import json +import os +import sys +from datetime import date, datetime, time, timedelta, timezone +from decimal import Decimal, InvalidOperation +from pathlib import Path +from urllib.error import HTTPError +from urllib.parse import quote, urlencode, urlsplit +from urllib.request import HTTPRedirectHandler, Request, build_opener +from zoneinfo import ZoneInfo + +UTC = timezone.utc +PLATFORMS = {"longbridge", "firstrade", "schwab", "ibkr", "binance", "alpaca", "qmt", "ftmo"} +FIELDS = {"cash", "positions", "orders", "fills"} + + +class AuditError(Exception): + """Only fixed reason codes may leave this module.""" + + +def instant(value): + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + if result.tzinfo is None: + raise AuditError("timestamp_invalid") + return result + + +def at(payload, path): + for key in path.split("."): + if not isinstance(payload, dict) or key not in payload: + return None + payload = payload[key] + return payload + + +def expected_dates(calendar, start, end, now): + """Independent of observed samples, including no samples and trailing loss.""" + first, last = date.fromisoformat(start), date.fromisoformat(end) + if last < first or (last - first).days > 365: + raise AuditError("window_invalid") + grace = calendar.get("grace_seconds") + if type(grace) is not int or not 0 <= grace <= 86400: + raise AuditError("calendar_invalid") + if calendar.get("kind") == "continuous": + if calendar.get("timezone") != "UTC": + raise AuditError("calendar_invalid") + return {str(first + timedelta(days=i)): None for i in range((last-first).days+1) + if datetime.combine(first+timedelta(days=i+1), time(), UTC)+timedelta(seconds=grace) <= now} + if calendar.get("kind") != "sessions" or not calendar.get("source") or not calendar.get("id"): + raise AuditError("calendar_invalid") + zone = ZoneInfo(calendar["timezone"]) + if date.fromisoformat(calendar["start"]) > first or date.fromisoformat(calendar["end"]) < last: + raise AuditError("calendar_coverage_missing") + sessions, result = set(), {} + for row in calendar["sessions"]: + day = date.fromisoformat(row["date"]) + close = instant(row["close_at"]) + if day in sessions or close.astimezone(zone).date() != day: + raise AuditError("calendar_invalid") + sessions.add(day) + if first <= day <= last and close+timedelta(seconds=grace) <= now: + result[str(day)] = close + return result + + +def field_issues(payload, rules): + issues = [] + if set(rules) != FIELDS: + raise AuditError("field_contract_invalid") + for name, rule in rules.items(): + required, numeric = rule.get("required", []), rule.get("numeric", []) + if not required or not numeric: + raise AuditError("field_contract_invalid") + rows = at(payload, rule["path"]) + complete = at(payload, rule["complete_path"]) + if (type(complete) is not type(rule["complete_value"]) + or complete != rule["complete_value"] or not isinstance(rows, list)): + issues.append(name+"_incomplete") + continue + if not rows: + # An empty list and a 'complete' label are not evidence of zero rows. + proof = rule.get("empty_proof", {}) + expected = {"query_ok": True, "row_count": 0, "pagination_complete": True} + if name == "cash" or set(proof) != set(expected) or not all( + type(at(payload, proof[key])) is type(value) and at(payload, proof[key]) == value + for key, value in expected.items()): + issues.append(name+"_empty_unverified") + continue + for row in rows: + if any(at(row, path) is None or at(row, path) == "" for path in required): + issues.append(name+"_row_invalid") + break + try: + if any(isinstance(at(row, path), bool) or at(row, path) is None + or not Decimal(str(at(row, path))).is_finite() for path in numeric): + raise InvalidOperation + except (InvalidOperation, ValueError): + issues.append(name+"_row_invalid") + break + return issues + + +def validate_snapshot(payload, target, day, close): + """Never infer identity/date/completeness from a filename or HTTP 200.""" + identity = target["identity"] + if not identity or not all(type(at(payload, key)) is type(value) and at(payload, key) == value + for key, value in identity.items()): + return ["identity_mismatch"] + try: + observed = instant(at(payload, target["as_of_path"])) + zone = ZoneInfo(target["calendar"]["timezone"]) + if str(observed.astimezone(zone).date()) != day or (close and observed < close): + return ["observation_date_mismatch"] + except (ValueError, TypeError, AttributeError): + return ["observation_time_invalid"] + return field_issues(payload, target["fields"]) + + +def day_prefix(template, day): + d = date.fromisoformat(day) + return template.format(date=day, yyyy=f"{d.year:04}", mm=f"{d.month:02}", dd=f"{d.day:02}") + + +def audit_target(target, start, end, now, io, backfill=False): + platform = target["platform"] + if platform not in PLATFORMS: + raise AuditError("platform_invalid") + if target.get("status") != "active": + return {"platform": platform, "status": "not_configured", "days": []} + expected = expected_dates(target["calendar"], start, end, now) + days = [] + for day, close in expected.items(): + item = {"date": day, "issues": [], "backfill": "not_needed"} + try: + records = io.read_day(target, day) + if not records: + item["issues"] = ["archive_missing"] + else: + # Do not hide a bad latest observation behind a good old one. + try: + latest = max(records, key=lambda p: instant(at(p, target["as_of_path"]))) + except (ValueError, TypeError, AttributeError): + raise AuditError("archive_invalid") from None + item["issues"] = validate_snapshot(latest, target, day, close) + if item["issues"]: + item["backfill"] = "unavailable" + # Identity/read failures are never repair triggers. + if backfill and target.get("history") and item["issues"] == ["archive_missing"]: + candidate = io.history(target, day) + if candidate is not None: + problems = validate_snapshot(candidate, target, day, close) + if problems: + item["backfill"] = "source_incomplete" + else: + io.persist(target, day, candidate) + readback = io.read_day(target, day) + if candidate not in readback: + raise AuditError("archive_readback_failed") + item["issues"], item["backfill"] = [], "verified" + except AuditError as exc: + item["issues"], item["backfill"] = [str(exc)], "stopped" + days.append(item) + if set(item["issues"]) & {"access_denied", "identity_mismatch", "source_restricted", "source_read_failed", + "archive_listing_truncated", "archive_readback_failed", "archive_write_conflict"}: + # Stop the affected source; do not retry permission or identity failures per day. + return {"platform": platform, "status": "unknown", "expected_count": len(expected), + "reason": "source_stopped", "days": days} + return {"platform": platform, "status": "incomplete" if any(x["issues"] for x in days) else "complete", + "expected_count": len(expected), "days": days} + + +class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def request_json(url, *, headers=None, data=None, method="GET", missing=False, conflict=False): + try: + request = Request(url, headers=headers or {}, data=data, method=method) + with build_opener(NoRedirect()).open(request, timeout=30) as response: + raw = response.read(4*1024*1024+1) + if len(raw) > 4*1024*1024: + raise AuditError("response_too_large") + return json.loads(raw) + except HTTPError as exc: + if exc.code == 404 and missing: + return None + if exc.code == 412 and conflict: + return False + if exc.code == 451: + raise AuditError("source_restricted") from None + raise AuditError("access_denied" if exc.code in (401, 403) else "source_read_failed") from None + except AuditError: + raise + except Exception: + raise AuditError("source_read_failed") from None + + +class CloudIO: + def __init__(self, env): + self.env = env + self.token = request_json( + "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token", + headers={"Metadata-Flavor": "Google"})["access_token"] + + def gcs(self, uri): + parsed = urlsplit(uri) + if parsed.scheme != "gs" or not parsed.netloc or parsed.query or parsed.fragment: + raise AuditError("archive_configuration_invalid") + return parsed.netloc, parsed.path.lstrip("/") + + def get(self, uri): + bucket, key = self.gcs(uri) + return request_json(f"https://storage.googleapis.com/storage/v1/b/{quote(bucket, safe='')}/o/{quote(key, safe='')}?alt=media", + headers={"Authorization": "Bearer "+self.token}, missing=True) + + def create(self, uri, payload): + bucket, key = self.gcs(uri) + query = urlencode({"uploadType": "media", "name": key, "ifGenerationMatch": "0"}) + result = request_json(f"https://storage.googleapis.com/upload/storage/v1/b/{quote(bucket, safe='')}/o?{query}", + headers={"Authorization": "Bearer "+self.token, "Content-Type": "application/json"}, + data=json.dumps(payload, sort_keys=True).encode(), method="POST", conflict=True) + return result is not False + + def read_day(self, target, day): + if target["archive"].get("kind") == "binance_detail": + result = self.read_endpoint(target["archive"], day) + return [] if result is None else [result["report"]] + bucket, prefix = self.gcs(day_prefix(target["archive"]["prefix"], day)) + if not prefix.endswith("/"): + raise AuditError("archive_configuration_invalid") + records, page = [], None + for _ in range(10): + query = {"prefix": prefix, "maxResults": 100} + if page: + query["pageToken"] = page + listing = request_json(f"https://storage.googleapis.com/storage/v1/b/{quote(bucket, safe='')}/o?{urlencode(query)}", + headers={"Authorization": "Bearer "+self.token}) + for obj in listing.get("items", []): + if not obj["name"].endswith(".json"): + continue + # Pin the listed generation; a concurrent overwrite is not a valid readback. + query = urlencode({"alt": "media", "generation": obj["generation"]}) + record = request_json(f"https://storage.googleapis.com/storage/v1/b/{quote(bucket, safe='')}/o/{quote(obj['name'], safe='')}?{query}", + headers={"Authorization": "Bearer "+self.token}) + records.append(at(record, target["payload_path"]) if target.get("payload_path") else record) + page = listing.get("nextPageToken") + if not page: + return records + raise AuditError("archive_listing_truncated") + + def read_endpoint(self, config, day): + parsed = urlsplit(config["url"]) + if (parsed.scheme != "https" or parsed.username or parsed.password or parsed.query or parsed.fragment + or not parsed.path.endswith(("/history", "/archive"))): + raise AuditError("history_configuration_invalid") + token = self.env.get(config["token_env"]) + if not token: + raise AuditError("history_credentials_unavailable") + return request_json(config["url"]+"?"+urlencode({"date": day}), + headers={"Authorization": "Bearer "+token}, missing=True) + + def history(self, target, day): + if target["history"].get("kind") != "broker_readable_history": + raise AuditError("history_configuration_invalid") + result = self.read_endpoint(target["history"], day) + return None if result is None else at(result, target["history"]["payload_path"]) + + def persist(self, target, day, payload): + # Canonical archive is append-only; failed/conflicting writes are never overwritten. + if target.get("payload_path"): + raise AuditError("backfill_envelope_adapter_required") + uri = day_prefix(target["archive"]["prefix"], day)+"broker-history-backfill.json" + if not self.create(uri, payload) and self.get(uri) != payload: + raise AuditError("archive_write_conflict") + + def alert(self, prefix, report, now): + content = json.dumps(report, sort_keys=True, separators=(",", ":")) + key = hashlib.sha256(content.encode()).hexdigest() + root = prefix.rstrip("/")+f"/alerts/{now.date()}/{key}" + if self.get(root+".ack.json"): + return "sender_acknowledged" + if not self.env.get("TELEGRAM_TOKEN") or not self.env.get("GLOBAL_TELEGRAM_CHAT_ID"): + raise AuditError("notification_configuration_missing") + # Claim before send: an uncertain send is never blindly repeated. + if not self.create(root+".claim.json", {"at": now.isoformat()}): + return "delivery_unknown" + text = "快照完整性检查异常\n"+"\n".join( + f"{r['platform']} target#{i+1}: {r['status']}; "+", ".join( + f"{d['date']}:{'/'.join(d['issues'])}" for d in r.get("days", []) if d["issues"]) + for i, r in enumerate(report["targets"]) if r["status"] != "complete") + send_notification(self.env, text) + self.create(root+".ack.json", {"at": now.isoformat(), "sender_ack": True, "receiver_receipt": None}) + return "sender_acknowledged" + + +def send_notification(env, text): + token, chat = env.get("TELEGRAM_TOKEN"), env.get("GLOBAL_TELEGRAM_CHAT_ID") + if not token or not chat: + raise AuditError("notification_configuration_missing") + response = request_json(f"https://api.telegram.org/bot{token}/sendMessage", method="POST", + headers={"Content-Type": "application/json"}, + data=json.dumps({"chat_id": chat, "text": text[:3900]}).encode()) + if response.get("ok") is not True: + raise AuditError("notification_failed") + + +def run(config, io, now): + targets = [] + for target in config["targets"]: + try: + targets.append(audit_target(target, config["start"], config["end"], now, io, + backfill=config.get("backfill_enabled") is True)) + except Exception: + platform = target.get("platform") if target.get("platform") in PLATFORMS else "unknown" + targets.append({"platform": platform, "status": "unknown", "reason": "configuration_or_calendar_invalid", "days": []}) + for missing in sorted(PLATFORMS - {t["platform"] for t in targets}): + targets.append({"platform": missing, "status": "not_configured", "days": []}) + report = {"schema_version": "snapshot_completeness.v1", "targets": targets, + "status": "complete" if targets and all(t["status"] == "complete" for t in targets) else "incomplete"} + return report + + +def main(): + cloud = bool(os.environ.get("CLOUD_RUN_JOB")) + notification_attempted = False + try: + if not cloud: + raise AuditError("cloud_run_job_required") + config = json.loads(Path(os.environ["SNAPSHOT_COMPLETENESS_CONFIG"]).read_text()) + now = datetime.now(UTC) + # Rolling window is anchored to the clock, never to the latest stored sample. + window = config.get("window_days", 30) + if type(window) is not int or not 1 <= window <= 366: + raise AuditError("window_invalid") + end = now.date()-timedelta(days=1) + config["end"] = str(end) + config["start"] = str(end-timedelta(days=window-1)) + io = CloudIO(os.environ) + report = run(config, io, now) + uri = config["state_prefix"].rstrip("/")+f"/reports/{now.strftime('%Y%m%dT%H%M%S%fZ')}.json" + if not io.create(uri, report) or io.get(uri) != report: + raise AuditError("report_readback_failed") + notification_attempted = report["status"] != "complete" + delivery = "not_needed" if not notification_attempted else io.alert(config["state_prefix"], report, now) + print(json.dumps({"status": report["status"], "notification": delivery, "targets": len(report["targets"])})) + return 0 if report["status"] == "complete" else 1 + except Exception: + # Exception strings, account identifiers, URIs and provider bodies are private. + if cloud and not notification_attempted: + try: + send_notification(os.environ, "快照完整性检查失败：配置或归档不可读；持久去重状态未知。请检查云端任务。") + except Exception: + pass + print(json.dumps({"status": "unknown", "reason": "snapshot_guard_failed"})) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python/tests/test_snapshot_completeness.py b/python/tests/test_snapshot_completeness.py new file mode 100644 index 00000000..19875a3b --- /dev/null +++ b/python/tests/test_snapshot_completeness.py @@ -0,0 +1,188 @@ +"""Synthetic regression tests; no broker, cloud or notification calls.""" +from datetime import datetime, timezone +from pathlib import Path +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) +import snapshot_completeness as guard + +NOW = datetime(2026, 11, 1, 12, tzinfo=timezone.utc) + + +def target(platform="longbridge"): + return {"platform": platform, "status": "active", "identity": {"account": "synthetic", "schema": "broker.v1"}, + "as_of_path": "observed_at", "archive": {"prefix": "gs://synthetic/archive/{date}/"}, + "calendar": {"kind": "continuous", "timezone": "UTC", "grace_seconds": 0}, + "fields": {name: {"path": name, "complete_path": "coverage."+name, "complete_value": "complete", + "required": ["asset", "quantity"], "numeric": ["quantity"], + "empty_proof": {"query_ok": "proof."+name+".query_ok", + "row_count": "proof."+name+".row_count", + "pagination_complete": "proof."+name+".pagination_complete"}} + for name in guard.FIELDS}} + + +def record(day): + return {"account": "synthetic", "schema": "broker.v1", "observed_at": day+"T22:00:00Z", + "coverage": {name: "complete" for name in guard.FIELDS}, + **{name: [{"asset": "SYNTH", "quantity": "1.5"}] for name in guard.FIELDS}} + + +class IO: + def __init__(self, records=None, historical=None): + self.records, self.historical = records or {}, historical + self.writes, self.reads, self.history_calls = [], [], [] + + def read_day(self, target, day): + self.reads.append(day) + return self.records.get(day, []) + + def history(self, target, day): + self.history_calls.append(day) + return self.historical + + def persist(self, target, day, payload): + self.writes.append(payload) + self.records[day] = [payload] + + +class CoverageTests(unittest.TestCase): + def test_trailing_leading_and_no_samples(self): + io = IO({"2026-10-03": [record("2026-10-03")]}) + result = guard.audit_target(target(), "2026-10-02", "2026-10-05", NOW, io) + self.assertEqual([d["date"] for d in result["days"] if d["issues"]], + ["2026-10-02", "2026-10-04", "2026-10-05"]) + self.assertEqual(guard.audit_target(target(), "2026-10-02", "2026-10-05", NOW, IO())["expected_count"], 4) + + def test_longbridge_audit_gaps(self): + for gaps in (["2026-10-20"], [f"2026-10-{d:02}" for d in range(2, 9)]+["2026-10-20"]): + io = IO({f"2026-10-{d:02}": [record(f"2026-10-{d:02}")] for d in range(2, 32) + if f"2026-10-{d:02}" not in gaps}) + result = guard.audit_target(target(), "2026-10-02", "2026-10-31", NOW, io) + self.assertEqual([d["date"] for d in result["days"] if d["issues"]], gaps) + + def test_sessions_holiday_weekend_and_deadline(self): + calendar = {"kind": "sessions", "id": "synthetic-exchange", "source": "synthetic-exchange-calendar", + "timezone": "Asia/Hong_Kong", "start": "2026-10-01", "end": "2026-10-31", + "grace_seconds": 3600, "sessions": [ + {"date": "2026-10-02", "close_at": "2026-10-02T16:00:00+08:00"}, + {"date": "2026-10-05", "close_at": "2026-10-05T16:00:00+08:00"}]} + expected = guard.expected_dates(calendar, "2026-10-01", "2026-10-05", guard.instant("2026-10-05T08:59:59Z")) + self.assertEqual(list(expected), ["2026-10-02"]) + self.assertEqual(list(guard.expected_dates(calendar, "2026-10-01", "2026-10-05", NOW)), + ["2026-10-02", "2026-10-05"]) + with self.assertRaises(guard.AuditError): + guard.expected_dates(calendar, "2026-09-30", "2026-10-05", NOW) + + def test_dst_and_early_close_use_published_instants(self): + calendar = {"kind": "sessions", "id": "synthetic-us", "source": "synthetic", "timezone": "America/New_York", + "start": "2026-10-30", "end": "2026-11-27", "grace_seconds": 0, + "sessions": [{"date": "2026-10-30", "close_at": "2026-10-30T20:00:00Z"}, + {"date": "2026-11-02", "close_at": "2026-11-02T21:00:00Z"}, + {"date": "2026-11-27", "close_at": "2026-11-27T18:00:00Z"}]} + self.assertEqual(len(guard.expected_dates(calendar, "2026-10-30", "2026-11-27", + guard.instant("2026-11-27T18:00:00Z"))), 3) + + def test_crypto_does_not_use_stock_weekends_or_today(self): + result = guard.expected_dates(target()["calendar"], "2026-10-30", "2026-11-01", NOW) + self.assertEqual(list(result), ["2026-10-30", "2026-10-31"]) + + def test_empty_positions_needs_real_zero_proof(self): + payload = record("2026-10-20") + payload["positions"] = [] + self.assertIn("positions_empty_unverified", guard.field_issues(payload, target()["fields"])) + payload["proof"] = {"positions": {"query_ok": True, "row_count": 0, "pagination_complete": True}} + self.assertEqual(guard.field_issues(payload, target()["fields"]), []) + payload["proof"]["positions"]["row_count"] = False + self.assertIn("positions_empty_unverified", guard.field_issues(payload, target()["fields"])) + + def test_missing_fields_null_nan_and_legitimate_zero(self): + for name in guard.FIELDS: + for value in (None, "NaN", "Infinity", True): + payload = record("2026-10-20") + payload[name][0]["quantity"] = value + self.assertIn(name+"_row_invalid", guard.field_issues(payload, target()["fields"])) + payload = record("2026-10-20") + payload["cash"][0]["quantity"] = "0" + self.assertEqual(guard.field_issues(payload, target()["fields"]), []) + del payload["orders"] + self.assertIn("orders_incomplete", guard.field_issues(payload, target()["fields"])) + + def test_identity_failure_stops_target(self): + payload = record("2026-10-20") + payload["account"] = "other-synthetic" + io = IO({"2026-10-20": [payload]}) + result = guard.audit_target(target(), "2026-10-20", "2026-10-22", NOW, io, True) + self.assertEqual(result["status"], "unknown") + self.assertEqual(io.reads, ["2026-10-20"]) + self.assertEqual(io.history_calls, []) + + def test_permission_failure_is_not_a_gap_and_stops(self): + io = IO() + io.read_day = lambda *_: (_ for _ in ()).throw(guard.AuditError("access_denied")) + result = guard.audit_target(target(), "2026-10-20", "2026-10-22", NOW, io, True) + self.assertEqual(result["days"][0]["issues"], ["access_denied"]) + self.assertEqual(len(result["days"]), 1) + + def test_old_good_sample_cannot_hide_new_incomplete_sample(self): + old, latest = record("2026-10-20"), record("2026-10-20") + latest["observed_at"] = "2026-10-20T23:00:00Z" + latest["positions"] = [] + result = guard.audit_target(target(), "2026-10-20", "2026-10-20", NOW, + IO({"2026-10-20": [old, latest]})) + self.assertEqual(result["status"], "incomplete") + + def test_backfill_requires_real_matching_date_and_readback(self): + configured = target() + configured["history"] = {"kind": "broker_readable_history"} + for source in (None, record("2026-10-21")): + io = IO(historical=source) + result = guard.audit_target(configured, "2026-10-20", "2026-10-20", NOW, io, True) + self.assertEqual(result["status"], "incomplete") + self.assertEqual(io.writes, []) + io = IO(historical=record("2026-10-20")) + result = guard.audit_target(configured, "2026-10-20", "2026-10-20", NOW, io, True) + self.assertEqual(result["days"][0]["backfill"], "verified") + self.assertEqual(len(io.reads), 2) + io = IO(historical=record("2026-10-20")) + io.persist = lambda *_: None + result = guard.audit_target(configured, "2026-10-20", "2026-10-20", NOW, io, True) + self.assertEqual(result["days"][0]["issues"], ["archive_readback_failed"]) + + def test_every_platform_and_unconfigured_account_are_explicit(self): + configured = [target(platform) for platform in sorted(guard.PLATFORMS)] + result = guard.run({"targets": configured, "start": "2026-10-20", "end": "2026-10-20"}, IO(), NOW) + self.assertEqual(len(result["targets"]), len(guard.PLATFORMS)) + self.assertTrue(all(t["status"] == "incomplete" for t in result["targets"])) + configured[0]["status"] = "unverified" + result = guard.run({"targets": configured, "start": "2026-10-20", "end": "2026-10-20"}, IO(), NOW) + self.assertEqual(result["targets"][0]["status"], "not_configured") + + def test_real_runner_refuses_mac(self): + with patch.dict(guard.os.environ, {}, clear=True), patch("builtins.print") as output: + self.assertEqual(guard.main(), 1) + self.assertNotIn("token", str(output.call_args)) + + def test_generation_precondition_and_redacted_transport_errors(self): + io = object.__new__(guard.CloudIO) + io.token = "synthetic-secret" + with patch.object(guard, "request_json", return_value={}) as request: + self.assertTrue(io.create("gs://synthetic/new.json", {})) + self.assertIn("ifGenerationMatch=0", request.call_args.args[0]) + with patch.object(guard, "build_opener", side_effect=RuntimeError("synthetic-secret")): + with self.assertRaisesRegex(guard.AuditError, "^source_read_failed$"): + guard.request_json("https://synthetic.invalid") + + def test_uncertain_telegram_send_is_not_retried(self): + io = object.__new__(guard.CloudIO) + io.env = {"TELEGRAM_TOKEN": "synthetic", "GLOBAL_TELEGRAM_CHAT_ID": "synthetic"} + io.get = lambda _: None + io.create = lambda *_: False + with patch.object(guard, "request_json") as request: + self.assertEqual(io.alert("gs://synthetic/state", {"targets": []}, NOW), "delivery_unknown") + request.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/binance_account_facts_validation.mjs b/tests/binance_account_facts_validation.mjs index 9638ef54..c931956a 100644 --- a/tests/binance_account_facts_validation.mjs +++ b/tests/binance_account_facts_validation.mjs @@ -353,6 +353,33 @@ const accepted = await post(report); assert.equal(accepted.status, 200, JSON.stringify(await accepted.clone().json())); assert.equal((await accepted.json()).status, 'published'); assert.equal(puts.length, 1); +const detailGet = (date, token = env.BINANCE_ACCOUNT_FACTS_SYNC_TOKEN) => worker.fetch(new Request( + `https://console.example/api/internal/binance-account-facts/archive?date=${date}`, + { headers: { Authorization: `Bearer ${token}` } }), env); +const reportDay = report.observed_finished_at.slice(0, 10); +assert.equal((await detailGet(reportDay, 'wrong-purpose')).status, 401); +assert.deepEqual((await (await detailGet(reportDay)).json()).report, report, + 'daily archive preserves native asset detail, source identity and observation times'); +const oldReport = structuredClone(report); +for (const field of ['observed_started_at', 'spot_observed_at', 'earn_observed_at', 'observed_finished_at']) { + oldReport[field] = new Date(Date.parse(report[field]) - 5 * 86400000).toISOString(); +} +const detailCommand = async (body) => runtimeStub.fetch('https://runtime-instances/', { + method: 'POST', body: JSON.stringify(body), +}); +const oldDay = oldReport.observed_finished_at.slice(0, 10); +assert.equal((await detailCommand({ action: 'binance_detail_put', binding, date: oldDay, report: oldReport })).status, 200); +values.delete(`${BINANCE_FACTS_KEY}:${binding.source_binding.id}`); +assert.deepEqual((await (await detailGet(oldDay)).json()).report, oldReport, + 'daily native detail survives latest-cache expiry and is readable beyond two days'); +values.set(`${BINANCE_FACTS_KEY}:${binding.source_binding.id}`, JSON.stringify(report)); +const conflictDetail = structuredClone(oldReport); +conflictDetail.assets[0].quantity = '15.5'; +conflictDetail.assets[0].spot_free = '4.1'; +assert.ok((await detailCommand({ action: 'binance_detail_put', binding, date: oldDay, report: conflictDetail })).status >= 400); +assert.ok((await detailCommand({ action: 'binance_detail_put', binding, date: reportDay, report: oldReport })).status >= 400); +assert.deepEqual((await (await detailGet(oldDay)).json()).report, oldReport, 'conflict never overwrites archived facts'); +assert.equal((await detailGet('2020-01-01')).status, 404); assert.equal(puts[0].key, `${BINANCE_FACTS_KEY}:${binding.source_binding.id}`); assert.equal((await (await post(report)).json()).status, 'unchanged'); assert.equal(puts.length, 1); @@ -510,7 +537,7 @@ const conflicting = structuredClone(report); conflicting.assets[0].spot_free = ' assert.equal((await post(conflicting)).status, 409); assert.equal(puts.length, 2); const invalidReceiverCases = invalid.map((item, index) => index === 12 - ? { ...item, observed_finished_at: new Date(Date.now() + 61_000).toISOString() } + ? { ...item, observed_finished_at: new Date(Date.now() + 120_000).toISOString() } : item); for (const [index, bad] of invalidReceiverCases.entries()) { assert.notEqual((await post(bad)).status, 200, `invalid report case ${index} must not be accepted`); diff --git a/web/strategy-switch-console/binance_detail_archive.js b/web/strategy-switch-console/binance_detail_archive.js new file mode 100644 index 00000000..e7ccbf19 --- /dev/null +++ b/web/strategy-switch-console/binance_detail_archive.js @@ -0,0 +1,44 @@ +import { normalizeBinanceAccountFacts, normalizeBinanceFactsBinding } from './binance_account_facts.js'; + +// Separate from the 366-day valuation projection and the two-day latest cache. +// Daily native detail has no automatic expiration. No new trading authority. +export function binanceDetailCommand(sql, command) { + const binding = normalizeBinanceFactsBinding(command.binding); + const day = command.date; + if (typeof day !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(day) + || !Number.isFinite(Date.parse(`${day}T00:00:00Z`)) + || new Date(`${day}T00:00:00Z`).toISOString().slice(0, 10) !== day) { + throw new Error('binance_detail_date_invalid'); + } + sql.exec(`CREATE TABLE IF NOT EXISTS binance_detail_daily ( + account_key TEXT NOT NULL, binding_id TEXT NOT NULL, observation_date TEXT NOT NULL, + payload_json TEXT NOT NULL, PRIMARY KEY(account_key, binding_id, observation_date))`); + const read = () => { + const row = sql.exec(`SELECT payload_json FROM binance_detail_daily + WHERE account_key = ? AND binding_id = ? AND observation_date = ?`, + binding.account_key, binding.source_binding.id, day).toArray()[0]; + if (!row) return null; + const report = normalizeBinanceAccountFacts(JSON.parse(row.payload_json), binding); + if (report.observed_finished_at.slice(0, 10) !== day) throw new Error('binance_detail_date_mismatch'); + return report; + }; + const existing = read(); + if (command.action === 'binance_detail_read') return { ok: true, report: existing }; + const report = normalizeBinanceAccountFacts(command.report, binding); + if (report.observed_finished_at.slice(0, 10) !== day) throw new Error('binance_detail_date_mismatch'); + const canonical = value => JSON.stringify(value, function (key, item) { + return item && typeof item === 'object' && !Array.isArray(item) + ? Object.fromEntries(Object.keys(item).sort().map(name => [name, item[name]])) : item; + }); + if (existing) { + const prior = Date.parse(existing.observed_finished_at), incoming = Date.parse(report.observed_finished_at); + if (incoming < prior || (incoming === prior && canonical(existing) !== canonical(report))) { + throw new Error('binance_detail_conflict'); + } + } + sql.exec(`INSERT INTO binance_detail_daily VALUES (?, ?, ?, ?) + ON CONFLICT(account_key, binding_id, observation_date) DO UPDATE SET payload_json = excluded.payload_json`, + binding.account_key, binding.source_binding.id, day, JSON.stringify(report)); + if (canonical(read()) !== canonical(report)) throw new Error('binance_detail_readback_failed'); + return { ok: true, readback_verified: true }; +} diff --git a/web/strategy-switch-console/worker.js b/web/strategy-switch-console/worker.js index c41c6f01..9d52031c 100644 --- a/web/strategy-switch-console/worker.js +++ b/web/strategy-switch-console/worker.js @@ -1,3 +1,4 @@ +import { binanceDetailCommand } from "./binance_detail_archive.js"; import { initialEnrollmentBindingsMatch, initialEnrollmentDigest } from "./application/console_api/initial_enrollment.js"; import { readNotificationLog } from "./application/console_api/notification_receipts.js"; import { readRuntimeControlReceipt } from "./application/console_api/runtime_control_receipts.js"; @@ -483,6 +484,7 @@ const ACCOUNT_FACTS_DO_ACTIONS = new Set([ "account_period_return_put", "account_period_return_read", "account_daily_native_return_research_read", "account_facts_put", "account_facts_read", "account_facts_list", "account_facts_history_read", "account_facts_history_put", "binance_wallet_history_put", "binance_wallet_history_read", + "binance_detail_put", "binance_detail_read", ]); const MARKET_BENCHMARK_DO_ACTIONS = new Set([ "market_benchmark_put", "market_benchmark_read", "market_benchmark_revise", @@ -940,6 +942,9 @@ export default { if (url.pathname === "/api/binance-private-scope" && request.method === "GET") { return await binancePrivateScopeResponse(request, env); } + if (url.pathname === "/api/internal/binance-account-facts/archive" && request.method === "GET") { + return await binanceDetailArchiveResponse(request, env, url); + } if (url.pathname === "/api/internal/binance-account-facts" && request.method === "GET") { return await binanceAccountFactsReadinessResponse(request, env); } @@ -2805,6 +2810,7 @@ export class RuntimeInstances { accountFactsCommand(command) { if (!command || typeof command.action !== "string") throw new HttpError("unsupported_account_facts_action", 400); if (command.action === "account_daily_native_return_research_read") return this.researchNativeDailyReturns(command); + if (["binance_detail_put", "binance_detail_read"].includes(command.action)) return binanceDetailCommand(this.sql, command); if (command.action === "binance_wallet_history_put" || command.action === "binance_wallet_history_read") { return this.binanceWalletHistoryCommand(command); } @@ -8506,6 +8512,10 @@ async function syncBinanceAccountFactsResponse(request, env) { unchanged = incomingTime === previousTime; } try { + const detail = await runtimeInstanceCommand(env, { + action: "binance_detail_put", binding, report, date: report.observed_finished_at.slice(0, 10), + }); + if (detail.readback_verified !== true) throw new Error("binance_detail_readback_failed"); const history = binanceWalletHistoryEntry(report, binding); await runtimeInstanceCommand(env, { action: "binance_wallet_history_put", @@ -8529,6 +8539,21 @@ async function syncBinanceAccountFactsResponse(request, env) { } } +async function binanceDetailArchiveResponse(request, env, url) { + const authError = binanceFactsSyncAuthorizationResponse(request, env); + if (authError) return authError; + try { + const binding = await trustedBinanceFactsBinding(env); + if (!binding) throw new Error("binding_missing"); + const result = await runtimeInstanceCommand(env, { + action: "binance_detail_read", binding, date: url.searchParams.get("date"), + }); + return json(result, result.report ? 200 : 404); + } catch { + return json({ ok: false, error: "binance_detail_archive_unavailable" }, 503); + } +} + async function binanceWalletHistoryResponse(request, env, url) { if (!accountFactsReadModelEnabled(env)) return accountFactsDisabledResponse(); const session = await readSession(request, env); From cacceb2a0a3bae86f53ed062732e64698f65a248 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:16:44 +0800 Subject: [PATCH 2/3] test: verify completed-day windows and cloud failure alerts Co-Authored-By: Codex --- python/tests/test_snapshot_completeness.py | 34 ++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/python/tests/test_snapshot_completeness.py b/python/tests/test_snapshot_completeness.py index 19875a3b..230d4fac 100644 --- a/python/tests/test_snapshot_completeness.py +++ b/python/tests/test_snapshot_completeness.py @@ -1,6 +1,8 @@ """Synthetic regression tests; no broker, cloud or notification calls.""" from datetime import datetime, timezone from pathlib import Path +import json +import tempfile import sys import unittest from unittest.mock import patch @@ -174,6 +176,38 @@ def test_generation_precondition_and_redacted_transport_errors(self): with self.assertRaisesRegex(guard.AuditError, "^source_read_failed$"): guard.request_json("https://synthetic.invalid") + def test_cloud_window_keeps_thirty_completed_dates(self): + class Clock(datetime): + @classmethod + def now(cls, tz=None): + return NOW + config = {"window_days": 30, "state_prefix": "gs://synthetic/state", "targets": []} + class Store: + def __init__(self, env): + self.saved = None + def create(self, uri, value): + self.saved = value + return True + def get(self, uri): + return self.saved + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.json" + path.write_text(json.dumps(config)) + with patch.dict(guard.os.environ, {"CLOUD_RUN_JOB": "synthetic", "SNAPSHOT_COMPLETENESS_CONFIG": str(path)}, clear=True), \ + patch.object(guard, "CloudIO", Store), patch.object(guard, "datetime", Clock), \ + patch.object(guard, "run", return_value={"status": "complete", "targets": []}) as audit, \ + patch("builtins.print"): + self.assertEqual(guard.main(), 0) + self.assertEqual(audit.call_args.args[0]["start"], "2026-10-02") + self.assertEqual(audit.call_args.args[0]["end"], "2026-10-31") + + def test_cloud_failure_alert_contains_no_exception_or_account(self): + with patch.dict(guard.os.environ, {"CLOUD_RUN_JOB": "synthetic"}, clear=True), \ + patch.object(guard, "send_notification") as send, patch("builtins.print") as output: + self.assertEqual(guard.main(), 1) + self.assertEqual(send.call_count, 1) + self.assertEqual(output.call_args.args[0], '{"status": "unknown", "reason": "snapshot_guard_failed"}') + def test_uncertain_telegram_send_is_not_retried(self): io = object.__new__(guard.CloudIO) io.env = {"TELEGRAM_TOKEN": "synthetic", "GLOBAL_TELEGRAM_CHAT_ID": "synthetic"} From 108177aa33fe4f9e46629c0bcbc36f024b43e13a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:23:04 +0800 Subject: [PATCH 3/3] fix: reject ambiguous completeness and recheck backfill results Co-Authored-By: Codex --- docs/snapshot-completeness.md | 3 ++- python/scripts/snapshot_completeness.py | 11 ++++++++--- python/tests/test_snapshot_completeness.py | 20 ++++++++++++++++++++ 3 files changed, 30 insertions(+), 4 deletions(-) diff --git a/docs/snapshot-completeness.md b/docs/snapshot-completeness.md index c7aa7208..db17eb10 100644 --- a/docs/snapshot-completeness.md +++ b/docs/snapshot-completeness.md @@ -46,7 +46,8 @@ Each target declares: The guard lists bounded pages and reads each JSON object at its listed generation. `payload_path` optionally selects a nested account snapshot. - `fields`: rules for all four groups, `cash`, `positions`, `orders`, `fills`. - Each rule supplies `path`, `complete_path`, `complete_value`, nonempty + Each rule supplies `path`, `complete_path`, `complete_value` (exactly `true`, + `complete` or `complete_for_scope`), nonempty `required` row paths and `numeric` row paths. Missing values, nonfinite numbers, unknown completeness and malformed rows fail. Actual broker-reported zero is valid; the guard never creates a value. diff --git a/python/scripts/snapshot_completeness.py b/python/scripts/snapshot_completeness.py index a8d7a647..f29570c0 100644 --- a/python/scripts/snapshot_completeness.py +++ b/python/scripts/snapshot_completeness.py @@ -73,7 +73,9 @@ def field_issues(payload, rules): raise AuditError("field_contract_invalid") for name, rule in rules.items(): required, numeric = rule.get("required", []), rule.get("numeric", []) - if not required or not numeric: + declared = rule.get("complete_value") + if (not required or not numeric or not (declared is True + or isinstance(declared, str) and declared in {"complete", "complete_for_scope"})): raise AuditError("field_contract_invalid") rows = at(payload, rule["path"]) complete = at(payload, rule["complete_path"]) @@ -158,9 +160,12 @@ def audit_target(target, start, end, now, io, backfill=False): else: io.persist(target, day, candidate) readback = io.read_day(target, day) - if candidate not in readback: + encoded = json.dumps(candidate, sort_keys=True) + if not any(json.dumps(row, sort_keys=True) == encoded for row in readback): raise AuditError("archive_readback_failed") - item["issues"], item["backfill"] = [], "verified" + latest = max(readback, key=lambda p: instant(at(p, target["as_of_path"]))) + item["issues"] = validate_snapshot(latest, target, day, close) + item["backfill"] = "verified" if not item["issues"] else "readback_incomplete" except AuditError as exc: item["issues"], item["backfill"] = [str(exc)], "stopped" days.append(item) diff --git a/python/tests/test_snapshot_completeness.py b/python/tests/test_snapshot_completeness.py index 230d4fac..ba7ab30e 100644 --- a/python/tests/test_snapshot_completeness.py +++ b/python/tests/test_snapshot_completeness.py @@ -111,6 +111,13 @@ def test_missing_fields_null_nan_and_legitimate_zero(self): del payload["orders"] self.assertIn("orders_incomplete", guard.field_issues(payload, target()["fields"])) + def test_unknown_completeness_contract_cannot_pass(self): + rules = target()["fields"] + for value in (None, False, 1, "unknown", "partial"): + rules["positions"]["complete_value"] = value + with self.assertRaises(guard.AuditError): + guard.field_issues(record("2026-10-20"), rules) + def test_identity_failure_stops_target(self): payload = record("2026-10-20") payload["account"] = "other-synthetic" @@ -152,6 +159,19 @@ def test_backfill_requires_real_matching_date_and_readback(self): result = guard.audit_target(configured, "2026-10-20", "2026-10-20", NOW, io, True) self.assertEqual(result["days"][0]["issues"], ["archive_readback_failed"]) + def test_backfill_readback_cannot_hide_concurrent_incomplete_record(self): + configured = target() + configured["history"] = {"kind": "broker_readable_history"} + io = IO(historical=record("2026-10-20")) + def persist(target, day, payload): + newer = record(day) + newer["observed_at"], newer["positions"] = day+"T23:00:00Z", [] + io.records[day] = [payload, newer] + io.persist = persist + result = guard.audit_target(configured, "2026-10-20", "2026-10-20", NOW, io, True) + self.assertEqual(result["status"], "incomplete") + self.assertEqual(result["days"][0]["backfill"], "readback_incomplete") + def test_every_platform_and_unconfigured_account_are_explicit(self): configured = [target(platform) for platform in sorted(guard.PLATFORMS)] result = guard.run({"targets": configured, "start": "2026-10-20", "end": "2026-10-20"}, IO(), NOW)