feat: Publish account-bound Flex settled cash for the digest

.github/workflows/publish-ibkr-digest-cash.yml

@@ -0,0 +1,62 @@
+name: Publish IBKR Digest Settled Cash
+
+on:
+  workflow_dispatch:
+    inputs:
+      read_once:
+        description: Read one configured Flex report and publish private period-end cash
+        type: boolean
+        default: false
+  schedule:
+    # Saturday UTC also collects Friday's completed statement if it was not ready on Friday.
+    - cron: "5 22 * * 1-6"
+
+permissions: {}
+concurrency:
+  group: ibkr-flex-period-return
+  cancel-in-progress: false
+
+jobs:
+  publish:
+    if: >-
+      github.repository == 'QuantStrategyLab/InteractiveBrokersPlatform' &&
+      github.ref == 'refs/heads/main' && github.run_attempt == 1 &&
+      (github.event_name == 'schedule' || inputs.read_once) &&
+      vars.IBKR_FLEX_DIGEST_JOB_REFS != '' && vars.IBKR_FLEX_DIGEST_WIF_PROVIDER != '' &&
+      vars.DIGEST_CANDIDATES_GCS_URI != ''
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      id-token: write
+    steps:
+      - name: Require infrastructure variables
+        env:
+          IBKR_DEPLOY_SERVICE_ACCOUNT: ${{ vars.IBKR_DEPLOY_SERVICE_ACCOUNT }}
+        shell: bash
+        run: |
+          set -euo pipefail
+          if [[ -z "${IBKR_DEPLOY_SERVICE_ACCOUNT:-}" ]]; then
+            echo "::error::IBKR_DEPLOY_SERVICE_ACCOUNT is required" >&2
+            exit 2
+          fi
+      - uses: actions/checkout@v6
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78
+      - run: uv sync --frozen --no-dev
+      - name: Authenticate to the existing digest candidate store
+        uses: google-github-actions/auth@7c6bc770dae815cd3e89ee6cdf493a5fab2cc093 # v3
+        with:
+          workload_identity_provider: ${{ vars.IBKR_FLEX_DIGEST_WIF_PROVIDER }}
+          service_account: ${{ vars.IBKR_DEPLOY_SERVICE_ACCOUNT }}
+      - name: Generate one Flex report and write bound cash objects
+        run: uv run --no-sync python -m scripts.publish_ibkr_digest_cash
+        env:
+          IBKR_FLEX_TOKEN: ${{ secrets.IBKR_FLEX_TOKEN }}
+          IBKR_FLEX_QUERY_ID: ${{ secrets.IBKR_FLEX_QUERY_ID }}
+          IBKR_FLEX_EXPECTED_ACCOUNT_IDS_JSON: ${{ secrets.IBKR_FLEX_EXPECTED_ACCOUNT_IDS_JSON }}
+          IBKR_ACCOUNT_FACTS_ACCOUNT_SELECTOR_JSON: ${{ secrets.IBKR_ACCOUNT_FACTS_ACCOUNT_SELECTOR_JSON }}
+          IBKR_ACCOUNT_FACTS_ADDITIONAL_TARGETS_JSON: ${{ secrets.IBKR_ACCOUNT_FACTS_ADDITIONAL_TARGETS_JSON }}
+          IBKR_FLEX_DIGEST_JOB_REFS: ${{ vars.IBKR_FLEX_DIGEST_JOB_REFS }}
+          DIGEST_CANDIDATES_GCS_URI: ${{ vars.DIGEST_CANDIDATES_GCS_URI }}

docs/ibkr-flex-ledger.zh-CN.md

@@ -162,6 +162,12 @@ TWR，不是账户收费计划、当前已交收资金、证券/碎股权限、
 
 ## 原生期间收益的一次发布
 
+配置 `IBKR_FLEX_DIGEST_JOB_REFS` 和现有摘要存储的云身份后，定时任务从一份独立的
+Flex 报表取四个日报账户的 USD 期末已结算现金。每户通过现有 Cloud Run 日报任务的
+账号和身份核对，金额与报表日期保存到各自的私有对象。完整收益报表仍按原规则校验。
+手动运行 `Publish IBKR Digest Settled Cash` 并选择 `read_once` 时只写现金摘要。
+现金和收益任务共用查询并发组，避免同时生成报表。
+
 `application.ibkr_period_return.build_ibkr_period_return` 从已复核账户的账本提取
 `ibkr_account_period_return.v1`。它只接受唯一受保护预期账户、有效且一致的报表/原生收益
 期间、`return_assessment.computable=True`、`native_ibkr_twr` 方法与 `ChangeInNAV.twr`

scripts/publish_ibkr_digest_cash.py

@@ -0,0 +1,161 @@
+"""Publish only account-bound IBKR Flex period-end cash for the private digest."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import time
+from datetime import date, datetime
+from urllib.parse import urlsplit
+from zoneinfo import ZoneInfo
+
+from application.ibkr_flex_ledger import build_flex_period_cash_facts
+from application.account_facts_target_config import account_selector, parse_additional_target_configs
+from application.ibkr_flex_source import (
+    FlexReportPending, _xml_root, collect_activity_flex_xml, fetch_activity_flex_xml, verify_report_accounts,
+)
+
+_JOB = re.compile(r"projects/[a-z0-9-]+/locations/[a-z0-9-]+/jobs/[a-z0-9-]+")
+_SLOT = re.compile(r"[a-z0-9][a-z0-9_-]{0,40}")
+_AMOUNT = re.compile(r"-?[0-9]+(?:\.[0-9]+)?")
+_SCHEMA = "ibkr_digest_settled_cash.v1"
+
+
+def _bindings(job_refs, destination, session):
+    refs = json.loads(job_refs)
+    if (not isinstance(refs, list) or len(refs) != 4
+            or any(not isinstance(ref, str) or not _JOB.fullmatch(ref) for ref in refs)
+            or len(set(refs)) != 4):
+        raise ValueError("digest_cash_jobs_invalid")
+    bindings = {}
+    for ref in refs:
+        response = session.get("https://run.googleapis.com/v2/" + ref, timeout=10, allow_redirects=False)
+        response.raise_for_status()
+        job = response.json()
+        container, = job["template"]["template"]["containers"]
+        env = {row["name"]: row.get("value") for row in container["env"]}
+        runtime = json.loads(env["RUNTIME_TARGET_JSON"])
+        account, = runtime["account_selector"]
+        slot = env["DIGEST_STORE_TARGET"]
+        uid, target = env["DIGEST_OPAQUE_ACCOUNT_UID"], env["DIGEST_TARGET_ID"]
+        if (job["name"] != ref or env["DIGEST_CANDIDATES_GCS_URI"].rstrip("/") != destination
+                or runtime["platform_id"] != "ibkr" or runtime["dry_run_only"] is not False
+                or not re.fullmatch(r"U[0-9]+", account) or account in bindings or not _SLOT.fullmatch(slot)
+                or not all(isinstance(value, str) and 0 < len(value) <= 128 for value in (uid, target))):
+            raise ValueError("digest_cash_binding_invalid")
+        bindings[account] = (slot, uid, target)
+    if any(len(set(values)) != 4 for values in zip(*bindings.values(), strict=True)):
+        raise ValueError("digest_cash_binding_duplicate")
+    return bindings
+
+
+def project_digest_cash(report, *, expected_primary, allowed_accounts, bindings, latest_day: date):
+    """Project verified Cash Report values without reading other statement sections."""
+    if len(bindings) != 4 or set(bindings) != set(allowed_accounts):
+        raise ValueError("digest_cash_account_scope_mismatch")
+    verify_report_accounts(report, expected_account_ids=expected_primary, allowed_report_account_ids=allowed_accounts)
+    root = _xml_root(report.content, endpoint="GetStatement")
+    if (root.get("type") or "").strip() not in {"", "AF"}:
+        raise ValueError("digest_cash_activity_report_required")
+    from application.ibkr_flex_executions import _day
+
+    payloads = {}
+    for statement in root.findall("./FlexStatements/FlexStatement"):
+        account = statement.get("accountId", "").strip()
+        slot, uid, target = bindings[account]
+        period = {"from": _day(statement.get("fromDate")), "to": _day(statement.get("toDate"))}
+        start, day = period["from"], period["to"]
+        if not start or not day or start > day or date.fromisoformat(day) > latest_day:
+            continue
+        if any(row.get("accountId", "").strip() not in {"", account}
+               for section in statement.findall("CashReport") for row in section.iter()):
+            continue
+        facts = build_flex_period_cash_facts(statement, source_ref=report.content_sha256, period=period)
+        if facts["status"] != "available_period_end":
+            continue
+        amount = facts["values"]["endingSettledCash"]
+        if len(amount) > 64 or not _AMOUNT.fullmatch(amount):
+            continue
+        payloads[slot] = {"schema_version": _SCHEMA, "target_id": target, "opaque_account_uid": uid,
+                          "statement_date": day, "currency": "USD", "amount": amount}
+    return payloads
+
+
+def publish_digest_cash(report, *, expected_primary, allowed_accounts, job_refs, destination):
+    import google.auth
+    from google.api_core.exceptions import NotFound
+    from google.auth.transport.requests import AuthorizedSession
+    from google.cloud import storage
+
+    destination = destination.rstrip("/")
+    uri = urlsplit(destination)
+    if uri.scheme != "gs" or not uri.netloc or not uri.path.strip("/") or uri.query or uri.fragment:
+        raise ValueError("digest_cash_store_invalid")
+    credentials, project = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
+    with AuthorizedSession(credentials) as session:
+        bindings = _bindings(job_refs, destination, session)
+    payloads = project_digest_cash(report, expected_primary=expected_primary, allowed_accounts=allowed_accounts, bindings=bindings,
+                                   latest_day=datetime.now(ZoneInfo("America/New_York")).date())
+    bucket = storage.Client(credentials=credentials, project=project).bucket(uri.netloc)
+    count = 0
+    for slot, payload in payloads.items():
+        blob = bucket.blob(f"{uri.path.strip('/')}/ibkr/flex-cash/{slot}.json")
+        try:
+            blob.reload(timeout=10, retry=None)
+        except NotFound:
+            generation = 0
+        else:
+            generation = int(blob.generation)
+            current = json.loads(blob.download_as_bytes(if_generation_match=generation, timeout=10, retry=None))
+            if (current["target_id"], current["opaque_account_uid"]) != (payload["target_id"], payload["opaque_account_uid"]):
+                raise ValueError("digest_cash_stored_binding_mismatch")
+            if current["statement_date"] > payload["statement_date"]:
+                continue
+            if current == payload:
+                count += 1
+                continue
+        blob.upload_from_string(json.dumps(payload, sort_keys=True), content_type="application/json",
+                                if_generation_match=generation, timeout=10, retry=None)
+        count += 1
+    return count
+
+
+def main():
+    try:
+        env = os.environ
+        primary = account_selector(json.loads(env["IBKR_FLEX_EXPECTED_ACCOUNT_IDS_JSON"]))
+        if len(primary) != 1 or primary != account_selector(json.loads(env["IBKR_ACCOUNT_FACTS_ACCOUNT_SELECTOR_JSON"])):
+            raise ValueError("digest_cash_primary_invalid")
+        configs = parse_additional_target_configs(env["IBKR_ACCOUNT_FACTS_ADDITIONAL_TARGETS_JSON"])
+        allowed = (*primary, *(value["account_selector"][0] for value in configs.values()))
+        if len(set(allowed)) != 4:
+            raise ValueError("digest_cash_account_scope_mismatch")
+        try:
+            report = fetch_activity_flex_xml(token=env["IBKR_FLEX_TOKEN"], query_id=env["IBKR_FLEX_QUERY_ID"])
+        except FlexReportPending as pending:
+            handle = pending.request
+            for _ in range(5):
+                time.sleep(3)
+                try:
+                    report = collect_activity_flex_xml(handle)
+                    break
+                except FlexReportPending as error:
+                    if error.request != handle:
+                        raise
+            else:
+                raise FlexReportPending(handle) from None
+        count = publish_digest_cash(report, expected_primary=primary, allowed_accounts=allowed,
+                                    job_refs=env["IBKR_FLEX_DIGEST_JOB_REFS"],
+                                    destination=env["DIGEST_CANDIDATES_GCS_URI"])
+        receipt = {"settled_cash_status": "published" if count == 4 else "partial", "settled_cash_accounts": count}
+    except FlexReportPending:
+        receipt = {"settled_cash_status": "flex_pending"}
+    except Exception:
+        receipt = {"settled_cash_status": "read_or_publish_failed"}
+    print(json.dumps(receipt, sort_keys=True))
+    return 0 if receipt["settled_cash_status"] == "published" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_publish_ibkr_digest_cash.py

@@ -0,0 +1,54 @@
+from datetime import date
+
+import pytest
+
+from application.ibkr_flex_source import FlexReport, FlexSourceError
+from scripts.publish_ibkr_digest_cash import project_digest_cash
+
+
+BINDINGS = {f"U000000{i}": (f"slot-{i}", f"opaque-{i}", f"target-{i}") for i in range(4)}
+
+
+def _report(*, extra=False, wrong_cash=False):
+    accounts = [*BINDINGS, "U0000099"] if extra else list(BINDINGS)
+    statements = []
+    for i, account in enumerate(accounts):
+        cash_account = accounts[0] if wrong_cash and i == 2 else account
+        statements.append(
+            f'<FlexStatement accountId="{account}" fromDate="20261015" toDate="20261015">'
+            f'<CashReport><CashReportCurrency accountId="{cash_account}" currency="USD" model="" '
+            f'fromDate="20261015" toDate="20261015" endingCash="500" endingSettledCash="-{i}.50"/></CashReport>'
+            '<Trades><Trade accountId="U0000099"/></Trades></FlexStatement>'
+        )
+    raw = ('<FlexQueryResponse type="AF"><FlexStatements>' + ''.join(statements)
+           + '</FlexStatements></FlexQueryResponse>').encode()
+    return FlexReport(content_sha256="a" * 64, content=raw)
+
+
+def _project(report, latest=date(2026, 10, 15)):
+    return project_digest_cash(report, expected_primary=(next(iter(BINDINGS)),), allowed_accounts=tuple(BINDINGS),
+                               bindings=BINDINGS, latest_day=latest)
+
+
+def test_projects_each_cash_segment_without_broker_account_ids():
+    result = _project(_report())
+    assert set(result) == {f"slot-{i}" for i in range(4)}
+    assert result["slot-3"] == {"schema_version": "ibkr_digest_settled_cash.v1", "target_id": "target-3",
+                                "opaque_account_uid": "opaque-3", "statement_date": "2026-10-15",
+                                "currency": "USD", "amount": "-3.50"}
+    assert "U0000" not in repr(result)
+
+
+def test_bad_cash_binding_only_excludes_affected_account_and_future_dates_are_ignored():
+    assert set(_project(_report(wrong_cash=True))) == {"slot-0", "slot-1", "slot-3"}
+    assert _project(_report(), date(2026, 10, 14)) == {}
+    with pytest.raises(FlexSourceError):
+        _project(_report(extra=True))
+
+
+def test_multiday_report_uses_the_period_end_date():
+    original = _report()
+    raw = original.content.replace(b'fromDate="20261015"', b'fromDate="20261001"')
+    result = _project(FlexReport(content_sha256="a" * 64, content=raw))
+    assert len(result) == 4
+    assert {record["statement_date"] for record in result.values()} == {"2026-10-15"}