Publish qualified native Schwab cash and account type facts

.github/workflows/account-facts-sync.yml

@@ -52,6 +52,7 @@ jobs:
           EXPECTED_RUNTIME_REVISION: ${{ inputs.expected_runtime_revision }}
           ACCOUNT_FACTS_SYNC_URL: ${{ vars.ACCOUNT_FACTS_SYNC_URL }}
           SCHWAB_NET_ASSETS_CURRENCY: ${{ vars.SCHWAB_NET_ASSETS_CURRENCY }}
+          SCHWAB_CASH_CURRENCY: ${{ vars.SCHWAB_CASH_CURRENCY }}
           SCHWAB_ACCOUNT_FACTS_TARGET_ID: ${{ secrets.SCHWAB_ACCOUNT_FACTS_TARGET_ID }}
           SCHWAB_ACCOUNT_FACTS_REPORT_PREFIX: ${{ secrets.SCHWAB_ACCOUNT_FACTS_REPORT_PREFIX }}
           SCHWAB_ACCOUNT_FACTS_SERVICE_NAME: ${{ secrets.SCHWAB_ACCOUNT_FACTS_SERVICE_NAME }}

.github/workflows/execution-report-heartbeat.yml

@@ -121,6 +121,7 @@ jobs:
     env:
       ACCOUNT_FACTS_SYNC_URL: ${{ vars.ACCOUNT_FACTS_SYNC_URL }}
       SCHWAB_NET_ASSETS_CURRENCY: ${{ vars.SCHWAB_NET_ASSETS_CURRENCY }}
+      SCHWAB_CASH_CURRENCY: ${{ vars.SCHWAB_CASH_CURRENCY }}
       SCHWAB_ACCOUNT_FACTS_TARGET_ID: ${{ secrets.SCHWAB_ACCOUNT_FACTS_TARGET_ID }}
       SCHWAB_ACCOUNT_FACTS_REPORT_PREFIX: ${{ secrets.SCHWAB_ACCOUNT_FACTS_REPORT_PREFIX }}
       SCHWAB_ACCOUNT_FACTS_SERVICE_NAME: ${{ secrets.SCHWAB_ACCOUNT_FACTS_SERVICE_NAME }}

.github/workflows/sync-cloud-run-env.yml

@@ -94,6 +94,7 @@ jobs:
       # control-plane variable while retaining a legacy-secret fallback during migration.
       RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON || secrets.RUNTIME_TARGET_JSON }}
       SCHWAB_DRY_RUN_ONLY: ${{ vars.SCHWAB_DRY_RUN_ONLY }}
+      SCHWAB_CASH_CURRENCY: ${{ vars.SCHWAB_CASH_CURRENCY }}
       SCHWAB_FEATURE_SNAPSHOT_PATH: ${{ vars.SCHWAB_FEATURE_SNAPSHOT_PATH }}
       SCHWAB_FEATURE_SNAPSHOT_MANIFEST_PATH: ${{ vars.SCHWAB_FEATURE_SNAPSHOT_MANIFEST_PATH }}
       SCHWAB_FEATURE_SNAPSHOT_FALLBACK_MODE: ${{ vars.SCHWAB_FEATURE_SNAPSHOT_FALLBACK_MODE }}
@@ -344,6 +345,11 @@ jobs:
             exit 1
           fi
 
+          if [ -n "${SCHWAB_CASH_CURRENCY:-}" ] && [ "${SCHWAB_CASH_CURRENCY}" != "USD" ]; then
+            echo "SCHWAB_CASH_CURRENCY must be USD when configured." >&2
+            exit 1
+          fi
+
       - name: Authenticate to Google Cloud
         id: auth
         if: steps.config.outputs.enabled == 'true'
@@ -365,11 +371,16 @@ jobs:
           DEPLOY_READBACK_FILE: ${{ runner.temp }}/cloud-run-no-traffic-baseline.json
         run: |
           set -euo pipefail
+          cash_currency_args=()
+          if [ -n "${SCHWAB_CASH_CURRENCY:-}" ]; then
+            cash_currency_args+=(--cash-currency="${SCHWAB_CASH_CURRENCY}")
+          fi
           python3 scripts/verify_cloud_run_no_traffic_deploy.py capture \
             --project="${GCP_PROJECT_ID}" \
             --region="${CLOUD_RUN_REGION}" \
             --service="${CLOUD_RUN_SERVICE}" \
             --scheduler-location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}" \
+            "${cash_currency_args[@]}" \
             --output="${DEPLOY_READBACK_FILE}"
 
       - name: Verify deployed runtime target admission before traffic shift
@@ -403,6 +414,11 @@ jobs:
           immutable_image="${image_repo}@${image_digest}"
           echo "image_digest=${image_digest}" >> "${GITHUB_OUTPUT}"
 
+          cash_deploy_env_args=()
+          if [ -n "${SCHWAB_CASH_CURRENCY:-}" ]; then
+            cash_deploy_env_args+=(--update-env-vars="SCHWAB_CASH_CURRENCY=${SCHWAB_CASH_CURRENCY}")
+          fi
+
           gcloud run deploy "${CLOUD_RUN_SERVICE}" \
             --project="${GCP_PROJECT_ID}" \
             --region="${CLOUD_RUN_REGION}" \
@@ -418,6 +434,7 @@ jobs:
             --timeout=300s \
             --labels="managed-by=github-actions,commit-sha=${GITHUB_SHA},github-run-id=${GITHUB_RUN_ID}" \
             --no-traffic \
+            "${cash_deploy_env_args[@]}" \
             --quiet
 
       - name: Verify no-traffic deployment readback
@@ -427,11 +444,16 @@ jobs:
           EXPECTED_IMAGE_DIGEST: ${{ steps.deploy.outputs.image_digest }}
         run: |
           set -euo pipefail
+          cash_currency_args=()
+          if [ -n "${SCHWAB_CASH_CURRENCY:-}" ]; then
+            cash_currency_args+=(--cash-currency="${SCHWAB_CASH_CURRENCY}")
+          fi
           python3 scripts/verify_cloud_run_no_traffic_deploy.py verify \
             --project="${GCP_PROJECT_ID}" \
             --region="${CLOUD_RUN_REGION}" \
             --service="${CLOUD_RUN_SERVICE}" \
             --scheduler-location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}" \
+            "${cash_currency_args[@]}" \
             --before="${DEPLOY_READBACK_FILE}" \
             --expected-sha="${EXPECTED_SHA}" \
             --expected-image-digest="${EXPECTED_IMAGE_DIGEST}"

application/account_observation.py

@@ -5,9 +5,14 @@
 from collections.abc import Mapping
 from datetime import datetime, timezone
 from decimal import Decimal, InvalidOperation
+import re
 from typing import Any
 
 
+_ACCOUNT_TYPE_TOKEN = re.compile(r"[A-Za-z_]{1,32}\Z", re.ASCII)
+_CASH_MONEY_TEXT = re.compile(r"^-?(?:0|[1-9]\d*)(?:\.\d+)?$")
+
+
 def expected_account_hash_from_selector(account_selector: Any) -> str | None:
     """Resolve only an explicit single-account selector; preserve legacy live lookup."""
 
@@ -44,10 +49,28 @@ def _money_text(value: Any) -> str | None:
     return format(amount, "f")
 
 
+def _cash_money_text(value: Any) -> str | None:
+    """Accept the bounded decimal text contract used by account-facts cash rows."""
+
+    if not isinstance(value, str) or _CASH_MONEY_TEXT.fullmatch(value) is None:
+        return None
+    whole, _, fraction = value.lstrip("-").partition(".")
+    if len(whole) > 15 or len(fraction) > 8:
+        return None
+    try:
+        amount = Decimal(value)
+    except (InvalidOperation, TypeError, ValueError):
+        return None
+    if not amount.is_finite():
+        return None
+    return value
+
+
 def build_account_observation(
     snapshot: Any,
     *,
     net_assets_currency: str | None = None,
+    cash_currency: str | None = None,
 ) -> dict[str, object] | None:
     """Project verified values without changing the snapshot or raising into execution."""
 
@@ -80,6 +103,12 @@ def build_account_observation(
             metadata.get("cash_available_for_withdrawal")
         )
 
+        raw_cash_balance = metadata.get("broker_cash_balance")
+        raw_cash_balance_source = metadata.get("broker_cash_balance_source")
+        cash_balance = _cash_money_text(raw_cash_balance)
+        raw_account_type = metadata.get("broker_account_type")
+        raw_account_type_source = metadata.get("broker_account_type_source")
+
         observation: dict[str, object] = {
             "account_hash": account_hash,
             "currency": None,
@@ -95,10 +124,24 @@ def build_account_observation(
                 "cashAvailableForWithdrawal" if available_for_withdrawal is not None else None
             ),
         }
-        return declare_net_assets_currency(
+        if cash_balance is not None and raw_cash_balance_source == "cashBalance":
+            observation["cash_balance"] = cash_balance
+            observation["cash_balance_source"] = "cashBalance"
+        if (
+            isinstance(raw_account_type, str)
+            and _ACCOUNT_TYPE_TOKEN.fullmatch(raw_account_type) is not None
+            and raw_account_type_source == "securitiesAccount.type"
+        ):
+            observation["broker_account_type"] = raw_account_type
+            observation["broker_account_type_source"] = "securitiesAccount.type"
+        declared_observation = declare_net_assets_currency(
             observation,
             net_assets_currency=net_assets_currency,
         )
+        return declare_cash_balance_currency(
+            declared_observation,
+            cash_currency=cash_currency,
+        )
     except Exception:
         # Reporting must never change the outcome of an already-run strategy cycle.
         return None
@@ -127,8 +170,32 @@ def declare_net_assets_currency(
     return projected
 
 
+def declare_cash_balance_currency(
+    observation: Mapping[str, object] | None,
+    *,
+    cash_currency: str | None = None,
+) -> dict[str, object] | None:
+    """Apply an independent owner-confirmed currency only to native cashBalance."""
+
+    if not isinstance(observation, Mapping):
+        return None
+    projected = dict(observation)
+    projected["cash_currency"] = None
+    projected["cash_currency_source"] = None
+    if (
+        cash_currency != "USD"
+        or not isinstance(projected.get("cash_balance"), str)
+        or projected.get("cash_balance_source") != "cashBalance"
+    ):
+        return projected
+    projected["cash_currency"] = "USD"
+    projected["cash_currency_source"] = "owner_confirmed"
+    return projected
+
+
 __all__ = [
     "build_account_observation",
+    "declare_cash_balance_currency",
     "declare_net_assets_currency",
     "expected_account_hash_from_selector",
 ]

main.py

@@ -15,6 +15,7 @@
 )
 from application.account_observation import (
     build_account_observation,
+    declare_cash_balance_currency,
     declare_net_assets_currency,
     expected_account_hash_from_selector,
 )
@@ -878,7 +879,12 @@ def _handle_schwab_cycle(*, dry_run_only_override: bool | None = None, response_
                 net_assets_currency=os.getenv("SCHWAB_NET_ASSETS_CURRENCY"),
             )
             if declared_observation is not None:
-                execution_summary["account_observation"] = declared_observation
+                cash_declared_observation = declare_cash_balance_currency(
+                    declared_observation,
+                    cash_currency=os.getenv("SCHWAB_CASH_CURRENCY"),
+                )
+                if cash_declared_observation is not None:
+                    execution_summary["account_observation"] = cash_declared_observation
         try:
             attach_cycle_execution_receipt(report, execution_result)
         except ValueError:
@@ -1095,6 +1101,7 @@ def _handle_schwab_probe(*, response_body: str = "Probe OK"):
         account_observation = build_account_observation(
             snapshot,
             net_assets_currency=os.getenv("SCHWAB_NET_ASSETS_CURRENCY"),
+            cash_currency=os.getenv("SCHWAB_CASH_CURRENCY"),
         )
         if account_observation is not None:
             summary["account_observation"] = account_observation

pyproject.toml

@@ -19,7 +19,7 @@ dependencies = [
     "google-cloud-storage",
     "google-auth",
     "numpy",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@c7646a7168b3dafa763ef7751a182d23e8de7790",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@2dc0b26ad4f1255b7ba44c3fca49fa52036b1a18",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@4a3943883cd6b5bbfe32a559e56a91b40a81b7ce",
 ]
 
@@ -61,5 +61,5 @@ include = [
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@c7646a7168b3dafa763ef7751a182d23e8de7790",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@2dc0b26ad4f1255b7ba44c3fca49fa52036b1a18",
 ]

qsl.toml

@@ -5,7 +5,7 @@ upgrade_ring = "ring_d"
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "c7646a7168b3dafa763ef7751a182d23e8de7790"
+quant_platform_kit = "2dc0b26ad4f1255b7ba44c3fca49fa52036b1a18"
 us_equity_strategies = "4a3943883cd6b5bbfe32a559e56a91b40a81b7ce"
 
 [qsl.compat]

scripts/publish_account_facts_from_reports.py

@@ -33,6 +33,7 @@
 FUTURE_SKEW = timedelta(minutes=5)
 _REPORT_PARTS = re.compile(r"^(\d{4}-\d{2})/(\d{8}T\d{6}Z)\.json$")
 _DECIMAL_TEXT = re.compile(r"^-?(?:0|[1-9]\d*)(?:\.\d+)?$")
+_ACCOUNT_TYPE_TOKEN = re.compile(r"[A-Za-z_]{1,32}\Z", re.ASCII)
 _REVISION = re.compile(r"^[a-z][a-z0-9-]{0,62}$")
 
 
@@ -143,6 +144,7 @@ def project_schwab_account_facts_history(
     expected_service_name: str,
     expected_runtime_revision: str,
     expected_target_id: str,
+    expected_cash_currency: str | None = None,
     now: datetime,
 ) -> dict[str, Any]:
     """Return the strict Schwab history body or an amount-free skip reason."""
@@ -200,6 +202,39 @@ def project_schwab_account_facts_history(
         if observed_at < current - MAX_AGE or observed_at > current + FUTURE_SKEW:
             raise _ProjectionError("observation_out_of_window")
 
+        cash: list[dict[str, str]] = []
+        if (
+            observation.get("cash_balance_source") == "cashBalance"
+            and observation.get("cash_currency") == "USD"
+            and observation.get("cash_currency_source") == "owner_confirmed"
+            and expected_cash_currency == "USD"
+        ):
+            try:
+                cash_balance = _money_text(observation.get("cash_balance"))
+            except _ProjectionError:
+                cash_balance = None
+            if cash_balance is not None:
+                cash = [
+                    {
+                        "currency": "USD",
+                        "cash_balance": cash_balance,
+                        "source_tag": "cashBalance",
+                        "currency_source": "owner_confirmed",
+                    }
+                ]
+
+        broker_account_type = None
+        raw_account_type = observation.get("broker_account_type")
+        if (
+            isinstance(raw_account_type, str)
+            and _ACCOUNT_TYPE_TOKEN.fullmatch(raw_account_type) is not None
+            and observation.get("broker_account_type_source") == "securitiesAccount.type"
+        ):
+            broker_account_type = {
+                "value": raw_account_type,
+                "source_tag": "securitiesAccount.type",
+            }
+
         report_month, report_run_id = _report_uri_parts(source_report_uri, report_prefix)
         if (
             report_month != started_at.strftime("%Y-%m")
@@ -209,7 +244,7 @@ def project_schwab_account_facts_history(
 
         if not isinstance(expected_target_id, str) or not expected_target_id.strip():
             raise _ProjectionError("target_identity_unavailable")
-        return {
+        return_body = {
             "schema_version": HISTORY_SCHEMA,
             "snapshot_schema_version": SNAPSHOT_SCHEMA,
             "account_scope": ACCOUNT_SCOPE,
@@ -231,9 +266,12 @@ def project_schwab_account_facts_history(
                     "currency_source": "owner_confirmed",
                 }
             ],
-            "cash": [],
+            "cash": cash,
             "account_hash": account_hash,
         }
+        if broker_account_type is not None:
+            return_body["broker_account_type"] = broker_account_type
+        return return_body
     except _ProjectionError as exc:
         return {"status": "skipped", "reason": exc.reason}
 
@@ -524,6 +562,7 @@ def main(argv: list[str] | None = None) -> int:
         print("skipped:net_assets_currency_unconfirmed")
         return 2
     target_id = os.getenv("SCHWAB_ACCOUNT_FACTS_TARGET_ID", "")
+    cash_currency = os.getenv("SCHWAB_CASH_CURRENCY")
     if not target_id.strip():
         print("skipped:target_identity_unavailable")
         return 2
@@ -564,6 +603,7 @@ def main(argv: list[str] | None = None) -> int:
         expected_service_name=service_name,
         expected_runtime_revision=expected_revision,
         expected_target_id=target_id,
+        expected_cash_currency=cash_currency,
         now=now,
     )
     if projected.get("status") == "skipped":