diff --git a/.github/workflows/publish-ibkr-digest-cash.yml b/.github/workflows/publish-ibkr-digest-cash.yml new file mode 100644 index 00000000..27598612 --- /dev/null +++ b/.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 }} diff --git a/docs/ibkr-flex-ledger.zh-CN.md b/docs/ibkr-flex-ledger.zh-CN.md index 55cdd896..e4a7074d 100644 --- a/docs/ibkr-flex-ledger.zh-CN.md +++ b/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` diff --git a/scripts/publish_ibkr_digest_cash.py b/scripts/publish_ibkr_digest_cash.py new file mode 100644 index 00000000..d01bd7b1 --- /dev/null +++ b/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()) diff --git a/tests/test_publish_ibkr_digest_cash.py b/tests/test_publish_ibkr_digest_cash.py new file mode 100644 index 00000000..fa93039f --- /dev/null +++ b/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'' + f'' + '' + ) + raw = ('' + ''.join(statements) + + '').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"}