From 97c805acabef160acc6b020cc4aa10c89856bc85 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 12 Nov 2026 06:57:35 +0800 Subject: [PATCH 1/4] feat: Publish account-bound Flex settled cash for the digest Co-Authored-By: Codex --- .../publish-ibkr-flex-period-return.yml | 39 +++++- docs/ibkr-flex-ledger.zh-CN.md | 5 + scripts/publish_ibkr_digest_cash.py | 116 ++++++++++++++++++ scripts/publish_ibkr_flex_period_return.py | 67 +++++++++- tests/test_publish_ibkr_digest_cash.py | 46 +++++++ 5 files changed, 266 insertions(+), 7 deletions(-) create mode 100644 scripts/publish_ibkr_digest_cash.py create mode 100644 tests/test_publish_ibkr_digest_cash.py diff --git a/.github/workflows/publish-ibkr-flex-period-return.yml b/.github/workflows/publish-ibkr-flex-period-return.yml index 4048d20b..db2130aa 100644 --- a/.github/workflows/publish-ibkr-flex-period-return.yml +++ b/.github/workflows/publish-ibkr-flex-period-return.yml @@ -12,6 +12,11 @@ on: type: boolean required: false default: false + digest_cash_only: + description: "Publish only the four private daily settled cash objects." + type: boolean + required: false + default: false report_date: description: "Optional one historical native daily return (YYYY-MM-DD), within the last 365 days. No trading or query template change." type: string @@ -62,9 +67,18 @@ jobs: publish: # A rerun would generate a new reference. Pending/unknown outcomes require # investigation, not a rerun or a fresh dispatch to replace the request. - if: ${{ github.repository == 'QuantStrategyLab/InteractiveBrokersPlatform' && github.ref == 'refs/heads/main' && github.run_attempt == 1 && ((github.event_name == 'workflow_dispatch' && inputs.publish_once && vars.IBKR_PERIOD_RETURN_PUBLISH_ENABLED == 'true') || (github.event_name == 'schedule' && vars.IBKR_PERIOD_RETURN_AUTO_SYNC_ENABLED == 'true')) }} + if: >- + github.repository == 'QuantStrategyLab/InteractiveBrokersPlatform' && + github.ref == 'refs/heads/main' && github.run_attempt == 1 && + ((github.event_name == 'workflow_dispatch' && + ((inputs.digest_cash_only && vars.IBKR_FLEX_DIGEST_JOB_REFS != '') || + (inputs.publish_once && vars.IBKR_PERIOD_RETURN_PUBLISH_ENABLED == 'true'))) || + (github.event_name == 'schedule' && vars.IBKR_PERIOD_RETURN_AUTO_SYNC_ENABLED == 'true')) runs-on: ubuntu-latest timeout-minutes: 10 + permissions: + contents: read + id-token: write steps: - name: Require infrastructure variables env: @@ -90,6 +104,16 @@ jobs: - name: Install locked runtime dependencies run: uv sync --frozen --no-dev + - name: Authenticate to the existing digest candidate store + if: >- + vars.IBKR_FLEX_DIGEST_JOB_REFS != '' && + inputs.report_date == '' && !inputs.primary_return_only && + !inputs.additional_returns_only && !inputs.external_flow_only + 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, collect and publish once run: | publisher_args=() @@ -97,7 +121,13 @@ jobs: echo "additional_return_slot_mode_invalid" >&2 exit 2 fi - if [ "$PRIMARY_RETURN_ONLY" = "true" ]; then + if [ "$DIGEST_CASH_ONLY" = "true" ]; then + if [ "$PRIMARY_RETURN_ONLY" = "true" ] || [ "$EXTERNAL_FLOW_ONLY" = "true" ] || [ "$ADDITIONAL_RETURNS_ONLY" = "true" ] || [ -n "$IBKR_FLOW_PERIOD_FROM$IBKR_FLOW_PERIOD_TO$IBKR_NATIVE_RETURN_DATE" ]; then + echo "digest_cash_only_mode_invalid" >&2 + exit 2 + fi + publisher_args=(--digest-cash-only) + elif [ "$PRIMARY_RETURN_ONLY" = "true" ]; then if [ "$EXTERNAL_FLOW_ONLY" = "true" ] || [ "$ADDITIONAL_RETURNS_ONLY" = "true" ] || [ -n "$IBKR_FLOW_PERIOD_FROM$IBKR_FLOW_PERIOD_TO" ] || [ -z "$IBKR_NATIVE_RETURN_DATE" ]; then echo "primary_return_mode_invalid" >&2 exit 2 @@ -121,6 +151,9 @@ jobs: uv run --no-sync python -m scripts.publish_ibkr_flex_period_return "${publisher_args[@]}" env: IBKR_PERIOD_RETURN_PUBLISH_ENABLED: "true" + DIGEST_CASH_ONLY: ${{ inputs.digest_cash_only }} + IBKR_FLEX_DIGEST_JOB_REFS: ${{ vars.IBKR_FLEX_DIGEST_JOB_REFS }} + DIGEST_CANDIDATES_GCS_URI: ${{ vars.DIGEST_CANDIDATES_GCS_URI }} EXTERNAL_FLOW_ONLY: ${{ inputs.external_flow_only }} ADDITIONAL_RETURNS_ONLY: ${{ inputs.additional_returns_only }} ADDITIONAL_RETURN_SLOT: ${{ inputs.additional_return_slot }} @@ -143,4 +176,4 @@ jobs: IBKR_PERIOD_RETURN_ADDITIONAL1_ACCOUNT_KEY: ${{ secrets.IBKR_PERIOD_RETURN_ADDITIONAL1_ACCOUNT_KEY }} IBKR_FLEX_TOKEN: ${{ secrets.IBKR_FLEX_TOKEN }} IBKR_FLEX_QUERY_ID: ${{ secrets.IBKR_FLEX_QUERY_ID }} - IBKR_ACCOUNT_FACTS_SYNC_TOKEN: ${{ secrets.IBKR_ACCOUNT_FACTS_SYNC_TOKEN }} + IBKR_ACCOUNT_FACTS_SYNC_TOKEN: ${{ !inputs.digest_cash_only && secrets.IBKR_ACCOUNT_FACTS_SYNC_TOKEN || '' }} diff --git a/docs/ibkr-flex-ledger.zh-CN.md b/docs/ibkr-flex-ledger.zh-CN.md index 55cdd896..ec199d67 100644 --- a/docs/ibkr-flex-ledger.zh-CN.md +++ b/docs/ibkr-flex-ledger.zh-CN.md @@ -162,6 +162,11 @@ TWR，不是账户收费计划、当前已交收资金、证券/碎股权限、 ## 原生期间收益的一次发布 +配置 `IBKR_FLEX_DIGEST_JOB_REFS` 和现有摘要存储的云身份后，定时发布会从同一次 +Flex 报表取四个日报账户的 USD 期末已结算现金。每户通过现有 Cloud Run 日报任务的 +账号和身份核对，金额与报表日期保存到各自的私有对象。完整收益报表仍按原规则校验。 +手动选 `digest_cash_only` 时只写现金摘要，现有收益、成交和资金流发布保持独立。 + `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..5c8c6cbb --- /dev/null +++ b/scripts/publish_ibkr_digest_cash.py @@ -0,0 +1,116 @@ +"""Publish only account-bound IBKR Flex period-end cash for the private digest.""" + +from __future__ import annotations + +import json +import re +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.ibkr_flex_source import _xml_root, 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"))} + day = period["to"] + if day is None or period["from"] != 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 diff --git a/scripts/publish_ibkr_flex_period_return.py b/scripts/publish_ibkr_flex_period_return.py index ea271c10..9927c963 100644 --- a/scripts/publish_ibkr_flex_period_return.py +++ b/scripts/publish_ibkr_flex_period_return.py @@ -496,7 +496,7 @@ def _collect_configured_ledger(config, session): session=session, allowed_report_account_ids=allowed) -def _collect_return_ledgers(config, session): +def _collect_configured_report(config, session): """One generation for all selected accounts, with the existing 1019 limit.""" day = config.get("report_date") request = request_activity_flex_report(token=config["flex_token"], query_id=config["query_id"], @@ -504,19 +504,69 @@ def _collect_return_ledgers(config, session): for attempt in range(6): try: report = collect_activity_flex_xml(request, session=session) - break + return report except FlexReportPending as pending: if pending.request != request: raise FlexSourceError("IBKR Flex pending handle changed") from None if attempt == 5: raise time.sleep(3) + + +def _build_return_ledgers(config, report): # Validate every selected ledger before any publication. Private XML stays in memory. return {slot: build_flex_ledger(report, expected_account_ids=target["expected_account_ids"], allowed_report_account_ids=config["allowed_report_account_ids"]) for slot, target in {"primary": config, **config["additional_returns"]}.items()} +def _collect_return_ledgers(config, session): + return _build_return_ledgers(config, _collect_configured_report(config, session)) + + +def _digest_cash(report, config): + from scripts.publish_ibkr_digest_cash import publish_digest_cash + + try: + count = publish_digest_cash( + report, expected_primary=config["expected_account_ids"], + allowed_accounts=config["allowed_report_account_ids"], + job_refs=os.environ["IBKR_FLEX_DIGEST_JOB_REFS"], + destination=os.environ["DIGEST_CANDIDATES_GCS_URI"], + ) + except Exception: + return {"settled_cash_status": "publish_failed"} + return {"settled_cash_status": "published" if count == 4 else "partial", + "settled_cash_accounts": count} + + +def _digest_cash_only(): + try: + env = os.environ + if (env.get("IBKR_PERIOD_RETURN_PUBLISH_ENABLED") != "true" or not env.get("IBKR_FLEX_DIGEST_JOB_REFS") + or any(env.get(name) for name in ("IBKR_NATIVE_RETURN_DATE", "IBKR_FLOW_PERIOD_FROM", "IBKR_FLOW_PERIOD_TO"))): + raise ValueError("digest_cash_configuration_missing") + primary = _selector(json.loads(env["IBKR_FLEX_EXPECTED_ACCOUNT_IDS_JSON"])) + if len(primary) != 1 or primary != _selector(json.loads(env["IBKR_ACCOUNT_FACTS_ACCOUNT_SELECTOR_JSON"])): + raise ValueError("digest_cash_primary_invalid") + config = { + "expected_account_ids": primary, + "allowed_report_account_ids": _configured_report_accounts(env, primary), + "flex_token": env["IBKR_FLEX_TOKEN"], + "query_id": env["IBKR_FLEX_QUERY_ID"], + } + with requests.Session() as session: + session.trust_env = False + report = _collect_configured_report(config, session) + result = _digest_cash(report, config) + except FlexReportPending: + result = {"settled_cash_status": "flex_pending"} + except Exception: + result = {"settled_cash_status": "read_failed"} + print(json.dumps(result, sort_keys=True)) + return 0 if result["settled_cash_status"] == "published" else 1 + + def _publish_additional_return(ledger, config, observed_at, *, diagnostics=None): try: payload = build_ibkr_period_return(ledger, observed_at=observed_at, @@ -547,6 +597,8 @@ def main(argv: Sequence[str] = ()) -> int: return 2 if tuple(argv) == ("--cashflow-only",): return _cashflow_only() + if tuple(argv) == ("--digest-cash-only",): + return _digest_cash_only() diagnostics = {} if tuple(argv) == ("--preflight",): result = _configuration_preflight(os.environ) @@ -562,6 +614,8 @@ def main(argv: Sequence[str] = ()) -> int: else: try: config = _config(os.environ) + digest_enabled = (not config.get("report_date") and not argv + and bool(os.environ.get("IBKR_FLEX_DIGEST_JOB_REFS"))) if primary_only and not config.get("report_date"): raise PeriodReturnError("historical_date_required") if additional_only and not config["additional_returns"]: @@ -577,7 +631,12 @@ def main(argv: Sequence[str] = ()) -> int: try: with requests.Session() as session: session.trust_env = False - if config["additional_returns"] and not primary_only: + if digest_enabled: + report = _collect_configured_report(config, session) + diagnostics.update(_digest_cash(report, config)) + ledgers = _build_return_ledgers(config, report) + ledger = ledgers["primary"] + elif config["additional_returns"] and not primary_only: ledgers = _collect_return_ledgers(config, session) ledger = ledgers["primary"] else: @@ -588,7 +647,7 @@ def main(argv: Sequence[str] = ()) -> int: status = "flex_pending" except FlexSourceError as exc: status = "flex_import_failed" - diagnostics = exc.safe_diagnostic + diagnostics.update(exc.safe_diagnostic) except Exception as exc: status = "flex_import_failed" name = type(exc).__name__ diff --git a/tests/test_publish_ibkr_digest_cash.py b/tests/test_publish_ibkr_digest_cash.py new file mode 100644 index 00000000..cabda89b --- /dev/null +++ b/tests/test_publish_ibkr_digest_cash.py @@ -0,0 +1,46 @@ +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)) From e3eae551483a3e8624afa5c0227701d5182980bb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 12 Nov 2026 07:03:52 +0800 Subject: [PATCH 2/4] fix: Isolate Flex cash publication from the return workflow Co-Authored-By: Codex --- .../workflows/publish-ibkr-digest-cash.yml | 51 ++++++++++++++ .../publish-ibkr-flex-period-return.yml | 39 +---------- docs/ibkr-flex-ledger.zh-CN.md | 5 +- scripts/publish_ibkr_digest_cash.py | 47 ++++++++++++- scripts/publish_ibkr_flex_period_return.py | 67 ++----------------- 5 files changed, 107 insertions(+), 102 deletions(-) create mode 100644 .github/workflows/publish-ibkr-digest-cash.yml diff --git a/.github/workflows/publish-ibkr-digest-cash.yml b/.github/workflows/publish-ibkr-digest-cash.yml new file mode 100644 index 00000000..af3e1acc --- /dev/null +++ b/.github/workflows/publish-ibkr-digest-cash.yml @@ -0,0 +1,51 @@ +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: + - cron: "5 22 * * 1-5" + +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: + - 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/.github/workflows/publish-ibkr-flex-period-return.yml b/.github/workflows/publish-ibkr-flex-period-return.yml index db2130aa..4048d20b 100644 --- a/.github/workflows/publish-ibkr-flex-period-return.yml +++ b/.github/workflows/publish-ibkr-flex-period-return.yml @@ -12,11 +12,6 @@ on: type: boolean required: false default: false - digest_cash_only: - description: "Publish only the four private daily settled cash objects." - type: boolean - required: false - default: false report_date: description: "Optional one historical native daily return (YYYY-MM-DD), within the last 365 days. No trading or query template change." type: string @@ -67,18 +62,9 @@ jobs: publish: # A rerun would generate a new reference. Pending/unknown outcomes require # investigation, not a rerun or a fresh dispatch to replace the request. - if: >- - github.repository == 'QuantStrategyLab/InteractiveBrokersPlatform' && - github.ref == 'refs/heads/main' && github.run_attempt == 1 && - ((github.event_name == 'workflow_dispatch' && - ((inputs.digest_cash_only && vars.IBKR_FLEX_DIGEST_JOB_REFS != '') || - (inputs.publish_once && vars.IBKR_PERIOD_RETURN_PUBLISH_ENABLED == 'true'))) || - (github.event_name == 'schedule' && vars.IBKR_PERIOD_RETURN_AUTO_SYNC_ENABLED == 'true')) + if: ${{ github.repository == 'QuantStrategyLab/InteractiveBrokersPlatform' && github.ref == 'refs/heads/main' && github.run_attempt == 1 && ((github.event_name == 'workflow_dispatch' && inputs.publish_once && vars.IBKR_PERIOD_RETURN_PUBLISH_ENABLED == 'true') || (github.event_name == 'schedule' && vars.IBKR_PERIOD_RETURN_AUTO_SYNC_ENABLED == 'true')) }} runs-on: ubuntu-latest timeout-minutes: 10 - permissions: - contents: read - id-token: write steps: - name: Require infrastructure variables env: @@ -104,16 +90,6 @@ jobs: - name: Install locked runtime dependencies run: uv sync --frozen --no-dev - - name: Authenticate to the existing digest candidate store - if: >- - vars.IBKR_FLEX_DIGEST_JOB_REFS != '' && - inputs.report_date == '' && !inputs.primary_return_only && - !inputs.additional_returns_only && !inputs.external_flow_only - 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, collect and publish once run: | publisher_args=() @@ -121,13 +97,7 @@ jobs: echo "additional_return_slot_mode_invalid" >&2 exit 2 fi - if [ "$DIGEST_CASH_ONLY" = "true" ]; then - if [ "$PRIMARY_RETURN_ONLY" = "true" ] || [ "$EXTERNAL_FLOW_ONLY" = "true" ] || [ "$ADDITIONAL_RETURNS_ONLY" = "true" ] || [ -n "$IBKR_FLOW_PERIOD_FROM$IBKR_FLOW_PERIOD_TO$IBKR_NATIVE_RETURN_DATE" ]; then - echo "digest_cash_only_mode_invalid" >&2 - exit 2 - fi - publisher_args=(--digest-cash-only) - elif [ "$PRIMARY_RETURN_ONLY" = "true" ]; then + if [ "$PRIMARY_RETURN_ONLY" = "true" ]; then if [ "$EXTERNAL_FLOW_ONLY" = "true" ] || [ "$ADDITIONAL_RETURNS_ONLY" = "true" ] || [ -n "$IBKR_FLOW_PERIOD_FROM$IBKR_FLOW_PERIOD_TO" ] || [ -z "$IBKR_NATIVE_RETURN_DATE" ]; then echo "primary_return_mode_invalid" >&2 exit 2 @@ -151,9 +121,6 @@ jobs: uv run --no-sync python -m scripts.publish_ibkr_flex_period_return "${publisher_args[@]}" env: IBKR_PERIOD_RETURN_PUBLISH_ENABLED: "true" - DIGEST_CASH_ONLY: ${{ inputs.digest_cash_only }} - IBKR_FLEX_DIGEST_JOB_REFS: ${{ vars.IBKR_FLEX_DIGEST_JOB_REFS }} - DIGEST_CANDIDATES_GCS_URI: ${{ vars.DIGEST_CANDIDATES_GCS_URI }} EXTERNAL_FLOW_ONLY: ${{ inputs.external_flow_only }} ADDITIONAL_RETURNS_ONLY: ${{ inputs.additional_returns_only }} ADDITIONAL_RETURN_SLOT: ${{ inputs.additional_return_slot }} @@ -176,4 +143,4 @@ jobs: IBKR_PERIOD_RETURN_ADDITIONAL1_ACCOUNT_KEY: ${{ secrets.IBKR_PERIOD_RETURN_ADDITIONAL1_ACCOUNT_KEY }} IBKR_FLEX_TOKEN: ${{ secrets.IBKR_FLEX_TOKEN }} IBKR_FLEX_QUERY_ID: ${{ secrets.IBKR_FLEX_QUERY_ID }} - IBKR_ACCOUNT_FACTS_SYNC_TOKEN: ${{ !inputs.digest_cash_only && secrets.IBKR_ACCOUNT_FACTS_SYNC_TOKEN || '' }} + IBKR_ACCOUNT_FACTS_SYNC_TOKEN: ${{ secrets.IBKR_ACCOUNT_FACTS_SYNC_TOKEN }} diff --git a/docs/ibkr-flex-ledger.zh-CN.md b/docs/ibkr-flex-ledger.zh-CN.md index ec199d67..e4a7074d 100644 --- a/docs/ibkr-flex-ledger.zh-CN.md +++ b/docs/ibkr-flex-ledger.zh-CN.md @@ -162,10 +162,11 @@ TWR，不是账户收费计划、当前已交收资金、证券/碎股权限、 ## 原生期间收益的一次发布 -配置 `IBKR_FLEX_DIGEST_JOB_REFS` 和现有摘要存储的云身份后，定时发布会从同一次 +配置 `IBKR_FLEX_DIGEST_JOB_REFS` 和现有摘要存储的云身份后，定时任务从一份独立的 Flex 报表取四个日报账户的 USD 期末已结算现金。每户通过现有 Cloud Run 日报任务的 账号和身份核对，金额与报表日期保存到各自的私有对象。完整收益报表仍按原规则校验。 -手动选 `digest_cash_only` 时只写现金摘要，现有收益、成交和资金流发布保持独立。 +手动运行 `Publish IBKR Digest Settled Cash` 并选择 `read_once` 时只写现金摘要。 +现金和收益任务共用查询并发组，避免同时生成报表。 `application.ibkr_period_return.build_ibkr_period_return` 从已复核账户的账本提取 `ibkr_account_period_return.v1`。它只接受唯一受保护预期账户、有效且一致的报表/原生收益 diff --git a/scripts/publish_ibkr_digest_cash.py b/scripts/publish_ibkr_digest_cash.py index 5c8c6cbb..e04dddaa 100644 --- a/scripts/publish_ibkr_digest_cash.py +++ b/scripts/publish_ibkr_digest_cash.py @@ -3,13 +3,18 @@ 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.ibkr_flex_source import _xml_root, verify_report_accounts +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}") @@ -114,3 +119,43 @@ def publish_digest_cash(report, *, expected_primary, allowed_accounts, job_refs, 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/scripts/publish_ibkr_flex_period_return.py b/scripts/publish_ibkr_flex_period_return.py index 9927c963..ea271c10 100644 --- a/scripts/publish_ibkr_flex_period_return.py +++ b/scripts/publish_ibkr_flex_period_return.py @@ -496,7 +496,7 @@ def _collect_configured_ledger(config, session): session=session, allowed_report_account_ids=allowed) -def _collect_configured_report(config, session): +def _collect_return_ledgers(config, session): """One generation for all selected accounts, with the existing 1019 limit.""" day = config.get("report_date") request = request_activity_flex_report(token=config["flex_token"], query_id=config["query_id"], @@ -504,69 +504,19 @@ def _collect_configured_report(config, session): for attempt in range(6): try: report = collect_activity_flex_xml(request, session=session) - return report + break except FlexReportPending as pending: if pending.request != request: raise FlexSourceError("IBKR Flex pending handle changed") from None if attempt == 5: raise time.sleep(3) - - -def _build_return_ledgers(config, report): # Validate every selected ledger before any publication. Private XML stays in memory. return {slot: build_flex_ledger(report, expected_account_ids=target["expected_account_ids"], allowed_report_account_ids=config["allowed_report_account_ids"]) for slot, target in {"primary": config, **config["additional_returns"]}.items()} -def _collect_return_ledgers(config, session): - return _build_return_ledgers(config, _collect_configured_report(config, session)) - - -def _digest_cash(report, config): - from scripts.publish_ibkr_digest_cash import publish_digest_cash - - try: - count = publish_digest_cash( - report, expected_primary=config["expected_account_ids"], - allowed_accounts=config["allowed_report_account_ids"], - job_refs=os.environ["IBKR_FLEX_DIGEST_JOB_REFS"], - destination=os.environ["DIGEST_CANDIDATES_GCS_URI"], - ) - except Exception: - return {"settled_cash_status": "publish_failed"} - return {"settled_cash_status": "published" if count == 4 else "partial", - "settled_cash_accounts": count} - - -def _digest_cash_only(): - try: - env = os.environ - if (env.get("IBKR_PERIOD_RETURN_PUBLISH_ENABLED") != "true" or not env.get("IBKR_FLEX_DIGEST_JOB_REFS") - or any(env.get(name) for name in ("IBKR_NATIVE_RETURN_DATE", "IBKR_FLOW_PERIOD_FROM", "IBKR_FLOW_PERIOD_TO"))): - raise ValueError("digest_cash_configuration_missing") - primary = _selector(json.loads(env["IBKR_FLEX_EXPECTED_ACCOUNT_IDS_JSON"])) - if len(primary) != 1 or primary != _selector(json.loads(env["IBKR_ACCOUNT_FACTS_ACCOUNT_SELECTOR_JSON"])): - raise ValueError("digest_cash_primary_invalid") - config = { - "expected_account_ids": primary, - "allowed_report_account_ids": _configured_report_accounts(env, primary), - "flex_token": env["IBKR_FLEX_TOKEN"], - "query_id": env["IBKR_FLEX_QUERY_ID"], - } - with requests.Session() as session: - session.trust_env = False - report = _collect_configured_report(config, session) - result = _digest_cash(report, config) - except FlexReportPending: - result = {"settled_cash_status": "flex_pending"} - except Exception: - result = {"settled_cash_status": "read_failed"} - print(json.dumps(result, sort_keys=True)) - return 0 if result["settled_cash_status"] == "published" else 1 - - def _publish_additional_return(ledger, config, observed_at, *, diagnostics=None): try: payload = build_ibkr_period_return(ledger, observed_at=observed_at, @@ -597,8 +547,6 @@ def main(argv: Sequence[str] = ()) -> int: return 2 if tuple(argv) == ("--cashflow-only",): return _cashflow_only() - if tuple(argv) == ("--digest-cash-only",): - return _digest_cash_only() diagnostics = {} if tuple(argv) == ("--preflight",): result = _configuration_preflight(os.environ) @@ -614,8 +562,6 @@ def main(argv: Sequence[str] = ()) -> int: else: try: config = _config(os.environ) - digest_enabled = (not config.get("report_date") and not argv - and bool(os.environ.get("IBKR_FLEX_DIGEST_JOB_REFS"))) if primary_only and not config.get("report_date"): raise PeriodReturnError("historical_date_required") if additional_only and not config["additional_returns"]: @@ -631,12 +577,7 @@ def main(argv: Sequence[str] = ()) -> int: try: with requests.Session() as session: session.trust_env = False - if digest_enabled: - report = _collect_configured_report(config, session) - diagnostics.update(_digest_cash(report, config)) - ledgers = _build_return_ledgers(config, report) - ledger = ledgers["primary"] - elif config["additional_returns"] and not primary_only: + if config["additional_returns"] and not primary_only: ledgers = _collect_return_ledgers(config, session) ledger = ledgers["primary"] else: @@ -647,7 +588,7 @@ def main(argv: Sequence[str] = ()) -> int: status = "flex_pending" except FlexSourceError as exc: status = "flex_import_failed" - diagnostics.update(exc.safe_diagnostic) + diagnostics = exc.safe_diagnostic except Exception as exc: status = "flex_import_failed" name = type(exc).__name__ From d7864d1dce03959f47cecba7a2abdab6cd73bd47 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 12 Nov 2026 07:07:41 +0800 Subject: [PATCH 3/4] fix: Check the existing cash publisher identity before auth Co-Authored-By: Codex --- .github/workflows/publish-ibkr-digest-cash.yml | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/.github/workflows/publish-ibkr-digest-cash.yml b/.github/workflows/publish-ibkr-digest-cash.yml index af3e1acc..532e2d8e 100644 --- a/.github/workflows/publish-ibkr-digest-cash.yml +++ b/.github/workflows/publish-ibkr-digest-cash.yml @@ -29,6 +29,16 @@ jobs: 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 From 028b3a4e4c4d23f61e34d42e75133d21e35b454d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 12 Nov 2026 07:11:33 +0800 Subject: [PATCH 4/4] fix: Keep Flex period ends current through the weekend Co-Authored-By: Codex --- .github/workflows/publish-ibkr-digest-cash.yml | 3 ++- scripts/publish_ibkr_digest_cash.py | 4 ++-- tests/test_publish_ibkr_digest_cash.py | 8 ++++++++ 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/.github/workflows/publish-ibkr-digest-cash.yml b/.github/workflows/publish-ibkr-digest-cash.yml index 532e2d8e..27598612 100644 --- a/.github/workflows/publish-ibkr-digest-cash.yml +++ b/.github/workflows/publish-ibkr-digest-cash.yml @@ -8,7 +8,8 @@ on: type: boolean default: false schedule: - - cron: "5 22 * * 1-5" + # Saturday UTC also collects Friday's completed statement if it was not ready on Friday. + - cron: "5 22 * * 1-6" permissions: {} concurrency: diff --git a/scripts/publish_ibkr_digest_cash.py b/scripts/publish_ibkr_digest_cash.py index e04dddaa..d01bd7b1 100644 --- a/scripts/publish_ibkr_digest_cash.py +++ b/scripts/publish_ibkr_digest_cash.py @@ -65,8 +65,8 @@ def project_digest_cash(report, *, expected_primary, allowed_accounts, bindings, account = statement.get("accountId", "").strip() slot, uid, target = bindings[account] period = {"from": _day(statement.get("fromDate")), "to": _day(statement.get("toDate"))} - day = period["to"] - if day is None or period["from"] != day or date.fromisoformat(day) > latest_day: + 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()): diff --git a/tests/test_publish_ibkr_digest_cash.py b/tests/test_publish_ibkr_digest_cash.py index cabda89b..fa93039f 100644 --- a/tests/test_publish_ibkr_digest_cash.py +++ b/tests/test_publish_ibkr_digest_cash.py @@ -44,3 +44,11 @@ def test_bad_cash_binding_only_excludes_affected_account_and_future_dates_are_ig 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"}