diff --git a/application/external_flow_snapshot.py b/application/external_flow_snapshot.py new file mode 100644 index 0000000..e90b310 --- /dev/null +++ b/application/external_flow_snapshot.py @@ -0,0 +1,174 @@ +"""Opt-in, read-only cash-flow snapshot for reports.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import date, datetime, timedelta, timezone +from decimal import Decimal, InvalidOperation +import hashlib +import re +from typing import Any + + +_IN = frozenset({"ACH_RECEIPT", "CASH_RECEIPT", "WIRE_IN"}) +_OUT = frozenset({"ACH_DISBURSEMENT", "CASH_DISBURSEMENT", "WIRE_OUT"}) +_INTERNAL = frozenset({"TRADE", "DIVIDEND_OR_INTEREST", "MONEY_MARKET"}) +_MONEY = re.compile(r"^-?(?:0|[1-9]\d{0,11})(?:\.\d{1,8})?$") + + +def external_flow_enabled(config: Mapping[str, str]) -> bool: + return config.get("SCHWAB_EXTERNAL_FLOW_EXPORT_ENABLED") == "true" + + +def _instant(value: Any) -> datetime: + if not isinstance(value, str): + raise ValueError("external_flow_time_invalid") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + raise ValueError("external_flow_time_invalid") from None + if parsed.tzinfo is None: + raise ValueError("external_flow_timezone_required") + return parsed.astimezone(timezone.utc) + + +def _utc(value: datetime) -> str: + return value.isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def project_external_flow_snapshot( + transactions: Sequence[Mapping[str, Any]] | None, + *, + expected_account_hash: str, + observed_account_hash: str, + period_start: str, + period_end: str, + currency: str | None, + source_complete: bool = False, + enabled: bool = False, +) -> dict[str, Any] | None: + """Project (start, end] already-read transactions; source completeness is explicit. + + Currency must be owner-confirmed exactly as in the existing USD account-facts + contract. Unknown journals, non-cash transfers, reversals, missing timestamps, + duplicate conflicts or unverified pagination prevent complete coverage. + """ + if enabled is not True: + return None + if not expected_account_hash or observed_account_hash != expected_account_hash: + raise ValueError("external_flow_account_mismatch") + start, end = _instant(period_start), _instant(period_end) + if end <= start or end - start > timedelta(days=2) or currency != "USD": + raise ValueError("external_flow_window_or_currency_invalid") + complete = source_complete is True and isinstance(transactions, (list, tuple)) + records: dict[str, dict[str, Any]] = {} + for tx in transactions or []: + if not isinstance(tx, Mapping): + complete = False + continue + kind = tx.get("type") + if kind in _INTERNAL: + continue + if kind not in _IN | _OUT: + # Includes JOURNAL/ELECTRONIC_FUND/RECEIVE_AND_DELIVER. A cash-looking + # label alone cannot prove an external movement or an in-kind value. + complete = False + continue + try: + when = _instant(tx.get("time")) + except ValueError: + complete = False + continue + if not start < when <= end: + continue + items = tx.get("transferItems") + if (tx.get("status") != "VALID" or not isinstance(items, list) or not items + or any(tx.get(field) not in (None, 0, "0", []) for field in ("fees", "commission", "transactionFee")) + or any(not isinstance(i, Mapping) or not isinstance(i.get("instrument"), Mapping) + or i["instrument"].get("assetType") != "CURRENCY" or i.get("feeType") is not None for i in items)): + complete = False + continue + native_id = tx.get("activityId") + amount = tx.get("netAmount") + try: + if isinstance(amount, bool) or amount is None: + raise ValueError + value = Decimal(str(amount)) + text = format(value, "f") + if (not value.is_finite() or not _MONEY.fullmatch(text) or value == 0 + or (kind in _IN) != (value > 0) or isinstance(native_id, bool) + or not isinstance(native_id, (str, int)) or not str(native_id)): + raise ValueError + except (ValueError, InvalidOperation): + complete = False + continue + settlement = tx.get("settlementDate") + if settlement is not None: + try: + # Preserve the broker date without assigning it a timezone/time. + if date.fromisoformat(settlement).isoformat() != settlement: + raise ValueError + except (ValueError, TypeError): + complete = False + continue + ref = hashlib.sha256(f"schwab_transactions\0{expected_account_hash}\0{native_id}".encode()).hexdigest() + record = {"source_ref": ref, "type": "deposit" if kind in _IN else "withdrawal", + "amount": text, "effective_at": _utc(when), "settlement_date": settlement, "transfer_ref": None} + if ref in records and records[ref] != record: + raise ValueError("external_flow_conflict") + records[ref] = record + if len(records) > 256: + raise ValueError("external_flow_size_limit") + return {"schema_version": "external_flow.v1", "windows": [{ + "currency": currency, "date": end.date().isoformat(), "period_start": _utc(start), "period_end": _utc(end), + "source": "schwab_transactions", "coverage": "complete" if complete else "incomplete", + "records": sorted(records.values(), key=lambda r: r["source_ref"]), + }]} + + +def attach_external_flow_snapshot( + history: Mapping[str, Any], snapshot: Mapping[str, Any] | None, *, enabled: bool = False, +) -> dict[str, Any]: + """Add an optional already-observed source to the existing trusted upload.""" + result = dict(history) + if enabled is not True or snapshot is None: + return result + end = _instant(snapshot.get("period_end")) + if end > _instant(history.get("observed_finished_at")): + raise ValueError("external_flow_observation_mismatch") + result["externalFlow"] = project_external_flow_snapshot( + snapshot.get("transactions"), expected_account_hash=history.get("account_hash"), + observed_account_hash=snapshot.get("account_hash"), period_start=snapshot.get("period_start"), + period_end=snapshot.get("period_end"), currency=snapshot.get("currency"), + source_complete=snapshot.get("source_complete") is True, enabled=True, + ) + return result + + +def collect_external_flow_snapshot( + client: Any, *, account_hash: str, period_start: str, period_end: str, + currency: str | None, enabled: bool = False, source_complete: bool = False, +) -> dict[str, Any] | None: + """Optional snapshot seam for an already-authenticated, account-bound reader. + + This does not construct clients or access credentials. Completeness must be + established by the caller's approved source/backfill policy; HTTP 200 alone + cannot prove it. The report publisher projects this raw envelope once. + """ + if enabled is not True: + return None + start, end = _instant(period_start), _instant(period_end) + if not account_hash or end <= start or end - start > timedelta(days=2): + raise ValueError("external_flow_window_or_account_invalid") + if currency != "USD": + raise ValueError("external_flow_window_or_currency_invalid") + response = client.get_transactions(account_hash, start_date=start, end_date=end) + if response.status_code != 200: + raise ValueError("external_flow_source_unavailable") + transactions = response.json() + if not isinstance(transactions, list): + raise ValueError("external_flow_source_invalid") + return { + "account_hash": account_hash, "period_start": _utc(start), "period_end": _utc(end), + "currency": currency, "source_complete": source_complete is True, "transactions": transactions, + } diff --git a/main.py b/main.py index 7ef0963..a4c1d47 100644 --- a/main.py +++ b/main.py @@ -3,6 +3,7 @@ import os import tempfile import time +from datetime import datetime, timedelta from zoneinfo import ZoneInfo from flask import Flask @@ -27,6 +28,7 @@ validate_reconciliation_preconditions, ) from application import signal_provenance +from application.external_flow_snapshot import collect_external_flow_snapshot, external_flow_enabled from application.runtime_broker_adapters import build_runtime_broker_adapters from application.runtime_report_summary import summarize_execution_cycle_result from application.runtime_composer import build_runtime_composer @@ -1146,6 +1148,23 @@ def _handle_schwab_probe(*, response_body: str = "Probe OK"): ) if account_observation is not None: summary["account_observation"] = account_observation + if external_flow_enabled(os.environ) and account_observation.get("net_assets_currency") == "USD": + try: + observed_at = datetime.fromisoformat(account_observation["observed_at"].replace("Z", "+00:00")) + report["account_external_flows"] = collect_external_flow_snapshot( + client, + account_hash=account_observation["account_hash"], + period_start=(observed_at - timedelta(days=1)).isoformat(), + period_end=account_observation["observed_at"], + currency=account_observation["net_assets_currency"], + enabled=True, + ) + except Exception as flow_exc: + log_runtime_event( + log_context, "external_flow_capture_failed", + message="Optional external flow capture failed", + error_type=type(flow_exc).__name__, + ) finalize_runtime_report( report, status="ok", diff --git a/scripts/publish_account_facts_from_reports.py b/scripts/publish_account_facts_from_reports.py index e4a211c..4e4df1b 100644 --- a/scripts/publish_account_facts_from_reports.py +++ b/scripts/publish_account_facts_from_reports.py @@ -125,7 +125,7 @@ def _project_positions(observation: Mapping[str, Any]) -> list[dict[str, str]] | def prepare_sync_payload( - body: Mapping[str, Any], *, include_positions: bool = False, + body: Mapping[str, Any], *, include_positions: bool = False, include_external_flow: bool = False, ) -> dict[str, Any]: """Keep only receiver fields from an already validated archive projection.""" allowed = { @@ -136,6 +136,8 @@ def prepare_sync_payload( } if include_positions is True: allowed.update(POSITIONS_KEYS) + if include_external_flow is True: + allowed.add("externalFlow") return {key: value for key, value in body.items() if key in allowed} @@ -247,6 +249,7 @@ def project_schwab_account_facts_history( expected_runtime_revision: str, expected_target_id: str, expected_cash_currency: str | None = None, + external_flow_enabled: bool = False, now: datetime, ) -> dict[str, Any]: """Return the strict Schwab history body or an amount-free skip reason.""" @@ -395,6 +398,14 @@ def project_schwab_account_facts_history( if positions is not None: return_body["broker_reported_positions"] = positions return_body["broker_reported_positions_scope"] = POSITIONS_SCOPE + if external_flow_enabled is True: + from application.external_flow_snapshot import attach_external_flow_snapshot + try: + return_body = attach_external_flow_snapshot( + return_body, report.get("account_external_flows"), enabled=True, + ) + except (TypeError, ValueError, AttributeError): + raise _ProjectionError("external_flow_invalid") from None return return_body except _ProjectionError as exc: return {"status": "skipped", "reason": exc.reason} @@ -720,6 +731,7 @@ def main(argv: list[str] | None = None) -> int: if report is None: print("skipped:report_unavailable") return 2 + include_external_flow = os.environ.get("SCHWAB_EXTERNAL_FLOW_EXPORT_ENABLED") == "true" projected = project_schwab_account_facts_history( report, source_report_uri=report_uri, @@ -728,6 +740,7 @@ def main(argv: list[str] | None = None) -> int: expected_runtime_revision=expected_revision, expected_target_id=target_id, expected_cash_currency=cash_currency, + external_flow_enabled=include_external_flow, now=now, ) if projected.get("status") == "skipped": @@ -736,6 +749,7 @@ def main(argv: list[str] | None = None) -> int: history = prepare_sync_payload( projected, include_positions=os.environ.get("SCHWAB_ACCOUNT_FACTS_INCLUDE_POSITIONS", "").strip().lower() == "true", + include_external_flow=include_external_flow, ) result = _publish_once( history, diff --git a/tests/test_external_flow_snapshot.py b/tests/test_external_flow_snapshot.py new file mode 100644 index 0000000..7e3aca7 --- /dev/null +++ b/tests/test_external_flow_snapshot.py @@ -0,0 +1,197 @@ +from copy import deepcopy +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from application.external_flow_snapshot import ( + attach_external_flow_snapshot, + collect_external_flow_snapshot, + external_flow_enabled, + project_external_flow_snapshot, +) +from scripts import publish_account_facts_from_reports as publisher +from test_publish_account_facts_from_reports import _valid_report + + +START = "2026-09-29T22:21:00Z" +END = "2026-09-30T22:21:00Z" +TX = {"activityId": "synthetic-flow", "type": "ACH_RECEIPT", "status": "VALID", "netAmount": "100.25", + "time": "2026-09-30T10:21:00Z", "settlementDate": "2026-09-30", + "transferItems": [{"instrument": {"assetType": "CURRENCY"}}]} +KWARGS = dict(expected_account_hash="synthetic-account-hash", observed_account_hash="synthetic-account-hash", + period_start=START, period_end=END, currency="USD", source_complete=True, enabled=True) + + +def project(rows, **overrides): + return project_external_flow_snapshot(rows, **(KWARGS | overrides)) + + +def test_default_off_has_no_client_calls_or_history_change(): + assert external_flow_enabled({}) is False + assert external_flow_enabled({"SCHWAB_EXTERNAL_FLOW_EXPORT_ENABLED": "true"}) is True + for value in [True, "1", "TRUE", "false"]: + assert external_flow_enabled({"SCHWAB_EXTERNAL_FLOW_EXPORT_ENABLED": value}) is False + assert project([TX], enabled=False) is None + client = Mock() + assert collect_external_flow_snapshot(client, account_hash="synthetic", period_start=START, + period_end=END, currency="USD") is None + client.get_transactions.assert_not_called() + assert attach_external_flow_snapshot({"old": "unchanged"}, {"invalid": True}) == {"old": "unchanged"} + + +def test_projection_preserves_signed_native_money_hashes_refs_and_dedupes(): + result = project([TX, TX]) + window = result["windows"][0] + assert window["coverage"] == "complete" + assert len(window["records"]) == 1 + record = window["records"][0] + assert record["amount"] == "100.25" + assert record["type"] == "deposit" + assert len(record["source_ref"]) == 64 + assert "synthetic-flow" not in str(result) + assert "synthetic-account-hash" not in str(result) + assert project([TX | {"type": "WIRE_OUT", "netAmount": "-2"}])["windows"][0]["records"][0]["type"] == "withdrawal" + assert project([])["windows"][0]["coverage"] == "complete" + assert project([], source_complete=False)["windows"][0]["coverage"] == "incomplete" + assert project(None)["windows"][0]["coverage"] == "incomplete" + + +@pytest.mark.parametrize("patch", [ + {"type": "JOURNAL"}, {"type": "RECEIVE_AND_DELIVER"}, {"type": "ELECTRONIC_FUND"}, {"type": "NEW_UNKNOWN"}, + {"time": None}, {"time": "2026-09-30T10:00:00"}, {"status": "PENDING"}, {"status": "REVERSED"}, + {"netAmount": "NaN"}, {"netAmount": True}, {"netAmount": "-100"}, {"netAmount": "0"}, + {"activityId": None}, {"transferItems": []}, {"settlementDate": "2026-02-30"}, + {"transferItems": [{"instrument": {"assetType": "EQUITY"}}]}, + {"fees": [{"amount": "1"}]}, {"commission": "1"}, + {"transferItems": [{"instrument": {"assetType": "CURRENCY"}, "feeType": "synthetic-fee"}]}, +]) +def test_unknown_or_unverified_sources_never_become_zero_flow(patch): + assert project([TX | patch])["windows"][0]["coverage"] == "incomplete" + + +def test_internal_dividend_interest_and_trade_are_not_external(): + assert project([{"type": "DIVIDEND_OR_INTEREST"}, {"type": "TRADE"}])["windows"][0]["records"] == [] + + +def test_timestamp_boundaries_identity_currency_and_conflict(): + assert project([TX | {"time": START}])["windows"][0]["records"] == [] + assert len(project([TX | {"time": END}])["windows"][0]["records"]) == 1 + with pytest.raises(ValueError, match="account_mismatch"): + project([TX], observed_account_hash="another-synthetic-account") + with pytest.raises(ValueError, match="currency"): + project([TX], currency=None) + with pytest.raises(ValueError, match="conflict"): + project([TX, TX | {"netAmount": "101"}]) + + +def test_fake_client_reuses_existing_read_method_without_assuming_complete(): + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=200, json=lambda: [TX]) + result = collect_external_flow_snapshot(client, account_hash="synthetic-account-hash", period_start=START, + period_end=END, currency="USD", enabled=True) + assert result["source_complete"] is False + assert attach_external_flow_snapshot({"account_hash": "synthetic-account-hash", "observed_finished_at": END}, result, enabled=True)["externalFlow"]["windows"][0]["coverage"] == "incomplete" + client.get_transactions.assert_called_once() + assert client.get_transactions.call_args.args == ("synthetic-account-hash",) + client.get_transactions.return_value.status_code = 403 + with pytest.raises(ValueError, match="unavailable"): + collect_external_flow_snapshot(client, account_hash="synthetic-account-hash", period_start=START, + period_end=END, currency="USD", enabled=True) + + +def test_actual_publisher_default_unchanged_and_optional_upload_contract(): + prefix = "gs://example-bucket/execution-reports/charles_schwab/soxl_soxx_trend_income/" + kwargs = dict(source_report_uri=prefix + "2026-09/20260930T222000Z.json", report_prefix=prefix, + expected_service_name="synthetic-service", expected_runtime_revision="service-00007-abc", + expected_target_id="synthetic-target", now=datetime(2026, 10, 1, 1, 20, tzinfo=timezone.utc)) + report = _valid_report() + legacy = publisher.project_schwab_account_facts_history(report, **kwargs) + report["account_external_flows"] = {"account_hash": "synthetic-account-hash", "period_start": START, + "period_end": END, "currency": "USD", "source_complete": True, "transactions": [TX]} + assert publisher.project_schwab_account_facts_history(report, **kwargs) == legacy + assert publisher.project_schwab_account_facts_history(_valid_report(), external_flow_enabled=True, **kwargs) == legacy + projected = publisher.project_schwab_account_facts_history(report, external_flow_enabled=True, **kwargs) + assert projected["externalFlow"] == project([TX]) + assert publisher.prepare_sync_payload(projected, include_external_flow=True)["externalFlow"] == project([TX]) + invalid = deepcopy(report) + invalid["account_external_flows"]["account_hash"] = "wrong-synthetic" + assert publisher.project_schwab_account_facts_history(invalid, external_flow_enabled=True, **kwargs) == {"status": "skipped", "reason": "external_flow_invalid"} + + +def test_collector_raw_envelope_attaches_without_adapter(): + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=200, json=lambda: [TX]) + raw = collect_external_flow_snapshot( + client, account_hash="synthetic-account-hash", period_start=START, + period_end=END, currency="USD", enabled=True, + ) + history = {"account_hash": "synthetic-account-hash", "observed_finished_at": END} + attached = attach_external_flow_snapshot(history, raw, enabled=True) + assert attached["externalFlow"] == project([TX], source_complete=False) + assert set(raw) == {"account_hash", "period_start", "period_end", "currency", "source_complete", "transactions"} + assert raw["source_complete"] is False + assert raw["transactions"] == [TX] + + +@pytest.mark.parametrize("rows", [[], [TX]]) +@pytest.mark.parametrize("attestation", [False, None, "true", 1, True]) +def test_raw_source_attestation_is_never_inferred(rows, attestation): + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=200, json=lambda: rows) + raw = collect_external_flow_snapshot( + client, account_hash="synthetic-account-hash", period_start=START, period_end=END, + currency="USD", enabled=True, source_complete=attestation, + ) + out = attach_external_flow_snapshot( + {"account_hash": "synthetic-account-hash", "observed_finished_at": END}, raw, enabled=True, + ) + assert out["externalFlow"]["windows"][0]["coverage"] == ("complete" if attestation is True else "incomplete") + assert client.get_transactions.call_count == 1 + + +@pytest.mark.parametrize("status", [206, 403, 429, 500]) +def test_failed_or_partial_source_does_not_return_raw(status): + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=status, json=Mock()) + with pytest.raises(ValueError, match="source_unavailable"): + collect_external_flow_snapshot(client, account_hash="synthetic-account-hash", period_start=START, + period_end=END, currency="USD", enabled=True) + client.get_transactions.return_value.json.assert_not_called() + + +@pytest.mark.parametrize("rows", [{}, {"transactions": [TX]}, None, "unknown"]) +def test_non_list_source_is_rejected(rows): + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=200, json=lambda: rows) + with pytest.raises(ValueError, match="source_invalid"): + collect_external_flow_snapshot(client, account_hash="synthetic-account-hash", period_start=START, + period_end=END, currency="USD", enabled=True) + + +def test_json_failure_propagates_without_raw_snapshot(): + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=200, json=Mock(side_effect=ValueError("synthetic invalid JSON"))) + with pytest.raises(ValueError, match="invalid JSON"): + collect_external_flow_snapshot(client, account_hash="synthetic-account-hash", period_start=START, + period_end=END, currency="USD", enabled=True) + + +@pytest.mark.parametrize("start,end", [(END, START), (START, START), ("2026-09-27T00:00:00Z", END), ("2026-09-29T22:21:00", END), ("invalid", END)]) +def test_invalid_windows_are_rejected_before_sdk(start, end): + client = Mock() + with pytest.raises(ValueError): + collect_external_flow_snapshot(client, account_hash="synthetic-account-hash", period_start=start, + period_end=end, currency="USD", enabled=True) + client.get_transactions.assert_not_called() + + +def test_offsets_dst_and_record_limit_preserve_projector_rules(): + kwargs = dict(period_start="2026-03-07T12:00:00Z", period_end="2026-03-08T12:00:00Z") + for when, expected in [("2026-03-08T01:30:00-05:00", "2026-03-08T06:30:00.000Z"), + ("2026-03-08T03:30:00-04:00", "2026-03-08T07:30:00.000Z")]: + out = project([TX | {"time": when}], **kwargs) + assert out["windows"][0]["records"][0]["effective_at"] == expected + with pytest.raises(ValueError, match="size_limit"): + project([TX | {"activityId": str(i)} for i in range(257)]) diff --git a/tests/test_publish_account_facts_from_reports.py b/tests/test_publish_account_facts_from_reports.py index dd4e801..7853a46 100644 --- a/tests/test_publish_account_facts_from_reports.py +++ b/tests/test_publish_account_facts_from_reports.py @@ -659,3 +659,70 @@ def test_account_scope_mode_token_classified(): "status": "skipped", "reason": "runtime_target_scope_mismatch:mode_token", } + + +def test_prepare_preserves_projected_external_flow_only_when_opted_in(): + from application.external_flow_snapshot import attach_external_flow_snapshot, collect_external_flow_snapshot + from test_external_flow_snapshot import START, END, TX + from unittest.mock import Mock + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=200, json=lambda: [TX]) + raw = collect_external_flow_snapshot( + client, account_hash="synthetic-account-hash", period_start=START, period_end=END, + currency="USD", enabled=True, + ) + body = attach_external_flow_snapshot( + {"account_hash": "synthetic-account-hash", "observed_finished_at": END}, raw, enabled=True, + ) + assert publisher.prepare_sync_payload(body, include_external_flow=True)["externalFlow"] == body["externalFlow"] + assert "externalFlow" not in publisher.prepare_sync_payload(body) + + +def test_main_uses_same_exact_opt_in_for_projection_and_prepare(): + from test_request_handling import _run_external_flow_probe + _, _, report = _run_external_flow_probe() + project = publisher.project_schwab_account_facts_history + def fixed_projection(value, **kwargs): + return project(value, **(kwargs | {"now": datetime(2026, 10, 1, 1, 20, tzinfo=timezone.utc)})) + prefix = "gs://example-bucket/execution-reports/charles_schwab/soxl_soxx_trend_income/" + env = { + "GITHUB_ACTIONS": "true", "SCHWAB_ACCOUNT_FACTS_SYNC_TOKEN": "synthetic-token", + "ACCOUNT_FACTS_SYNC_URL": publisher.SYNC_URL, "SCHWAB_ACCOUNT_FACTS_REPORT_PREFIX": prefix, + "SCHWAB_ACCOUNT_FACTS_SERVICE_NAME": "synthetic-service", "SCHWAB_NET_ASSETS_CURRENCY": "USD", + "SCHWAB_ACCOUNT_FACTS_TARGET_ID": "synthetic-target", + } + for flag in ["true", "false", "TRUE", ""]: + with (patch.dict("os.environ", env | {"SCHWAB_EXTERNAL_FLOW_EXPORT_ENABLED": flag}, clear=True), + patch.object(publisher, "_service_revision_matches", return_value=True), + patch.object(publisher, "_read_report", return_value=report), + patch.object(publisher, "project_schwab_account_facts_history", side_effect=fixed_projection), + patch.object(publisher, "_publish_once", return_value={"status": "published"}) as publish): + assert publisher.main(["--report-uri", prefix + "2026-09/20260930T222000Z.json", "--expected-runtime-revision", "service-00007-abc"]) == 0 + assert ("externalFlow" in publish.call_args.args[0]) == (flag == "true") + + +def test_external_flow_opt_in_preserves_independent_positions_opt_in(): + from application.external_flow_snapshot import attach_external_flow_snapshot, collect_external_flow_snapshot + from test_external_flow_snapshot import START, END, TX + from unittest.mock import Mock + client = Mock() + client.get_transactions.return_value = SimpleNamespace(status_code=200, json=lambda: [TX]) + raw = collect_external_flow_snapshot(client, account_hash="synthetic-account-hash", period_start=START, period_end=END, currency="USD", enabled=True) + report = _valid_report() + report["account_external_flows"] = raw + observation = report["summary"]["account_observation"] + observation["broker_reported_positions_scope"] = publisher.POSITIONS_SCOPE + observation["broker_reported_positions"] = [{"symbol": "SOXL", "quantity": "1", "market_value": "20", "currency": None}] + prefix = "gs://example-bucket/execution-reports/charles_schwab/soxl_soxx_trend_income/" + kwargs = dict(source_report_uri=prefix + "2026-09/20260930T222000Z.json", report_prefix=prefix, + expected_service_name="synthetic-service", expected_runtime_revision="service-00007-abc", + expected_target_id="synthetic-target", now=datetime(2026, 10, 1, 1, 20, tzinfo=timezone.utc)) + projected = publisher.project_schwab_account_facts_history(report, external_flow_enabled=True, **kwargs) + assert "broker_reported_positions" in projected + for positions in [False, True]: + for flow in [False, True]: + payload = publisher.prepare_sync_payload(projected, include_positions=positions, include_external_flow=flow) + assert ("broker_reported_positions" in payload) is positions + assert ("externalFlow" in payload) is flow + observation["net_assets_currency_source"] = None + assert publisher.project_schwab_account_facts_history(report, external_flow_enabled=True, **kwargs)["reason"] == "net_assets_currency_unconfirmed" diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 6bbe4a7..0053c20 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -1311,3 +1311,87 @@ def fake_history(_client, symbol): if __name__ == "__main__": unittest.main() + + +def _run_external_flow_probe(*, enabled="true", rows=None, status_code=200, currency="USD", persist=None, finalizer=None): + from datetime import datetime, timezone + from unittest.mock import Mock + from test_external_flow_snapshot import TX, END + from test_publish_account_facts_from_reports import _valid_report + + module = load_module() + client = Mock() + client.get_transactions.return_value = types.SimpleNamespace( + status_code=status_code, json=lambda: [TX] if rows is None else rows, + ) + snapshot = types.SimpleNamespace( + buying_power=100, total_equity=123.45, positions=(), + as_of=datetime.fromisoformat(END.replace("Z", "+00:00")), + metadata={"account_hash": "synthetic-account-hash", "total_equity_source": "broker_liquidation_value"}, + ) + report = _valid_report() + observed = {} + module.build_execution_report = lambda *_args, **_kwargs: report + if finalizer is not None: + module.finalize_runtime_report = finalizer + module.build_broker_adapters = lambda: types.SimpleNamespace(fetch_managed_snapshot=lambda c: snapshot) + module.emit_runtime_log = lambda *_args, **_kwargs: None + def save(value, **_kwargs): + observed["report"] = value + return persist(value) if persist else "synthetic-local-report.json" + module.persist_execution_report = save + module.build_composer = lambda **_kwargs: types.SimpleNamespace( + build_reporting_adapters=lambda: types.SimpleNamespace(build_log_context=lambda: types.SimpleNamespace(run_id=report["run_id"]), log_event=lambda *_args, **_kwargs: None), + build_client=lambda: client, + build_notification_adapters=lambda: (_ for _ in ()).throw(AssertionError("optional capture must not notify")), + ) + with patch.dict(os.environ, {"SCHWAB_EXTERNAL_FLOW_EXPORT_ENABLED": enabled, "SCHWAB_NET_ASSETS_CURRENCY": currency}): + result = module._handle_schwab_probe() + return result, client, observed["report"] + + +def test_probe_external_flow_report_hook_uses_bound_collector(): + from application.external_flow_snapshot import attach_external_flow_snapshot + from test_external_flow_snapshot import END, TX, project + result, client, report = _run_external_flow_probe() + assert result == ("Probe OK", 200) + client.get_transactions.assert_called_once() + assert client.get_transactions.call_args.args == ("synthetic-account-hash",) + raw = report["account_external_flows"] + assert raw["transactions"] == [TX] + assert raw["source_complete"] is False + assert attach_external_flow_snapshot( + {"account_hash": "synthetic-account-hash", "observed_finished_at": END}, raw, enabled=True, + )["externalFlow"] == project([TX], source_complete=False) + + +def test_probe_external_flow_off_and_no_currency_keep_legacy(): + for enabled, currency in [("false", "USD"), ("TRUE", "USD"), ("", "USD"), ("true", ""), ("true", "USDT")]: + result, client, report = _run_external_flow_probe(enabled=enabled, currency=currency) + assert result == ("Probe OK", 200) + client.get_transactions.assert_not_called() + assert "account_external_flows" not in report + + +def test_probe_external_flow_failed_source_omits_slot_and_preserves_probe_result(): + for status in [206, 403, 429, 500]: + result, client, report = _run_external_flow_probe(status_code=status) + assert result == ("Probe OK", 200) + client.get_transactions.assert_called_once() + assert report["status"] == "ok" + assert "account_external_flows" not in report + for rows in [{}, "unknown"]: + result, client, report = _run_external_flow_probe(rows=rows) + assert result == ("Probe OK", 200) + assert "account_external_flows" not in report + + +def test_probe_external_flow_empty_source_is_not_complete(): + from application.external_flow_snapshot import attach_external_flow_snapshot + from test_external_flow_snapshot import END + result, client, report = _run_external_flow_probe(rows=[]) + assert result == ("Probe OK", 200) + raw = report["account_external_flows"] + projected = attach_external_flow_snapshot({"account_hash": raw["account_hash"], "observed_finished_at": END}, raw, enabled=True) + assert projected["externalFlow"]["windows"][0]["coverage"] == "incomplete" + assert projected["externalFlow"]["windows"][0]["records"] == [] diff --git a/tests/test_runtime_reporting_adapters.py b/tests/test_runtime_reporting_adapters.py index e102aef..de1f981 100644 --- a/tests/test_runtime_reporting_adapters.py +++ b/tests/test_runtime_reporting_adapters.py @@ -56,3 +56,61 @@ def fake_report_builder(**kwargs): assert observed["report_builder"]["runtime_target"].platform_id == "charles_schwab" assert observed["report_builder"]["runtime_target"].execution_mode == "paper" assert report == {"run_id": "run-001"} + + +def test_actual_probe_qpk_writer_readback_publisher_prepare_roundtrip(tmp_path, monkeypatch): + import json + from quant_platform_kit.common import runtime_reports + from test_request_handling import _run_external_flow_probe + from scripts import publish_account_facts_from_reports as publisher + + def no_cloud(*_args, **_kwargs): + raise AssertionError("offline local writer must not upload") + monkeypatch.setattr(runtime_reports, "upload_runtime_report_to_cloud", no_cloud) + adapters = build_runtime_reporting_adapters( + runtime_assembly=build_runtime_assembly( + platform="charles_schwab", deploy_target="cloud_run", service_name="synthetic-service", + strategy_profile=publisher.STRATEGY_PROFILE, project_id=publisher.PROJECT_ID, + ), + strategy_domain="us_equity", managed_symbols=(), benchmark_symbol="", + strategy_display_name="Synthetic", strategy_display_name_localized="Synthetic", dry_run=False, + signal_effective_after_trading_days=None, report_base_dir=str(tmp_path), report_gcs_prefix_uri=None, + run_id_builder=lambda: "synthetic", event_logger=lambda *_args, **_kwargs: {}, + report_builder=runtime_reports.build_runtime_report_base, report_persister=runtime_reports.persist_runtime_report, + ) + observed = {} + def persist(report): + observed["path"] = adapters.persist_execution_report(report) + return observed["path"] + def finalize(report, **kwargs): + return runtime_reports.finalize_runtime_report(report, finished_at="2026-09-30T22:25:00Z", **kwargs) + result, client, original = _run_external_flow_probe(persist=persist, finalizer=finalize) + assert result == ("Probe OK", 200) + client.get_transactions.assert_called_once() + saved = json.loads(Path(observed["path"]).read_text()) + assert saved["account_external_flows"] == original["account_external_flows"] + prefix = "gs://example-bucket/execution-reports/charles_schwab/soxl_soxx_trend_income/" + kwargs = dict(source_report_uri=prefix + "2026-09/20260930T222000Z.json", report_prefix=prefix, + expected_service_name="synthetic-service", expected_runtime_revision="service-00007-abc", + expected_target_id="synthetic-target", now=datetime(2026, 10, 1, 1, 20, tzinfo=timezone.utc)) + actual = publisher.project_schwab_account_facts_history(saved, external_flow_enabled=True, **kwargs) + assert "externalFlow" in actual + assert actual == publisher.project_schwab_account_facts_history(original, external_flow_enabled=True, **kwargs) + payload = publisher.prepare_sync_payload(actual, include_external_flow=True) + assert payload["externalFlow"]["windows"][0]["coverage"] == "incomplete" + assert len(payload["externalFlow"]["windows"][0]["records"]) == 1 + (tmp_path / "actual-chain-payload.json").write_text(json.dumps(payload, indent=2)) + # OFF and ON without a source keep the same actual archive projection. + saved.pop("account_external_flows") + assert publisher.project_schwab_account_facts_history(saved, external_flow_enabled=True, **kwargs) == publisher.project_schwab_account_facts_history(saved, **kwargs) + + +def test_probe_writer_failure_does_not_claim_an_archive(capsys): + from test_request_handling import _run_external_flow_probe + def fail(_report): + raise OSError("synthetic writer unavailable") + result, _client, _report = _run_external_flow_probe(persist=fail) + assert result == ("Probe OK", 200) + output = capsys.readouterr().out + assert "failed to persist execution report" in output + assert "execution_report synthetic-local-report" not in output