From 328fee701fb7fd8fa5f871a6a7a5136b114b885b Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 13 Sep 2026 16:29:04 +0800 Subject: [PATCH 1/2] fix(binance): surface daily state blockers safely Co-Authored-By: Codex --- application/cycle_service.py | 11 +++-- application/portfolio_service.py | 33 +++++++++++++- scripts/execution_report_heartbeat.py | 20 +++++++++ scripts/validate_runtime_startup.py | 27 +++++++++++- tests/test_cycle_service.py | 21 +++++++++ tests/test_execution_report_heartbeat.py | 56 +++++++++++++++++++++++- tests/test_forward_earn_accounting.py | 43 ++++++++++++++++++ tests/test_validate_runtime_startup.py | 18 ++++++++ 8 files changed, 222 insertions(+), 7 deletions(-) diff --git a/application/cycle_service.py b/application/cycle_service.py index da1d91f7..5dd4f332 100644 --- a/application/cycle_service.py +++ b/application/cycle_service.py @@ -9,6 +9,7 @@ from quant_platform_kit.common.runtime_reports import persist_runtime_report from quant_platform_kit.strategy_lifecycle.performance_monitor import try_record_platform_execution from application.execution_receipt_adapter import attach_execution_receipt_from_report +from application.portfolio_service import EARN_FORWARD_REASON_CODES from runtime_logging import RuntimeLogContext, emit_runtime_log from runtime_support import ( append_report_error, finalize_notification_delivery, acquire_runtime_state_owner, @@ -334,9 +335,13 @@ def execute_strategy_cycle( KeyError: "key_error", RuntimeError: "runtime_error", }.get(type(exc), "unclassified_error") - report.setdefault("diagnostics", {})["cycle_failure"] = { - "stage": failure_stage, "error_type": error_type, - } + failure_metadata = {"stage": failure_stage, "error_type": error_type} + if failure_stage == "daily_state": + earn_diagnostics = report.get("diagnostics", {}).get("earn_accrual", {}) + reason_code = earn_diagnostics.get("reason_code") if isinstance(earn_diagnostics, Mapping) else None + if reason_code in EARN_FORWARD_REASON_CODES: + failure_metadata["reason_code"] = reason_code + report.setdefault("diagnostics", {})["cycle_failure"] = failure_metadata log_buffer.append(f"cycle_execution_failed stage={failure_stage} error_type={error_type}") append_report_error(report, "cycle_execution_failed", stage="execute_cycle") try: diff --git a/application/portfolio_service.py b/application/portfolio_service.py index 91f642e0..dcf74d8c 100644 --- a/application/portfolio_service.py +++ b/application/portfolio_service.py @@ -11,6 +11,28 @@ _TREND_PNL_BASIS = "trend_mark_plus_cash_flow_v1" +EARN_FORWARD_REASON_CODES = frozenset({ + "earn_checkpoint_invalid", + "earn_checkpoint_time_invalid", + "earn_checkpoint_scope_changed", + "earn_product_lifecycle_unverified", + "earn_counter_reset", + "earn_quantity_change_unexplained", + "earn_order_unsettled", + "earn_accounted_changes_missing", + "earn_cash_cursor_mismatch", + "earn_cash_flow_invalid", + "earn_cash_flow_unsupported", + "earn_cash_flow_time_unverified", + "earn_valuation_snapshot_mismatch", + "external_cash_flow_window_invalid", + "external_cash_flow_cursor_invalid", + "external_cash_flow_history_read_failed", + "external_cash_flow_history_incomplete", + "external_cash_flow_record_invalid", + "external_cash_flow_record_changed", + "external_cash_flow_cursor_capacity_exceeded", +}) def compute_portfolio_allocation( @@ -71,7 +93,16 @@ def maybe_rebase_daily_state_for_balance_change( cash = collect_external_cash_flows_fn(runtime.client, now=_time(current['observed_at']), cursor=state.get('external_cash_flow_cursor')) updated = prepare_forward_earn_state(state, current, cash) - except Exception: + except Exception as exc: + reason_code = ( + str(exc) + if type(exc) is ValueError and str(exc) in EARN_FORWARD_REASON_CODES + else "earn_forward_accounting_unverified" + ) + report.setdefault("diagnostics", {})["earn_accrual"] = { + "status": "blocked", + "reason_code": reason_code, + } raise ExecutionIntegrityError("earn_forward_accounting_unverified") from None runtime_set_trade_state_fn(runtime, report, updated, reason="earn_forward_accounting") state.clear() diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index ceccdd92..ed38db89 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -14,6 +14,8 @@ _SCHEMA = "qsl.execution_report_heartbeat_assessment.v1" _ACCEPTED_STATUSES = frozenset({"ok", "skipped", "success", "completed", "no_action", "aborted"}) +_OPERATIONAL_BLOCKERS = frozenset({"state_owner_busy"}) +_OPERATIONAL_BLOCKER_REASON = "execution is blocked by an operational guard" def _env_bool(name: str, default: bool = False) -> bool: @@ -136,6 +138,11 @@ def _payload_strategy(payload: dict[str, Any]) -> str: return str(payload.get("strategy_profile") or target_strategy or "").strip() +def _operational_blocker(payload: dict[str, Any]) -> str | None: + value = payload.get("execution_blocked_reason") + return value if isinstance(value, str) and value in _OPERATIONAL_BLOCKERS else None + + def _accepted_payload(payload: dict[str, Any]) -> tuple[bool, str]: expected_platform = (os.environ.get("RUNTIME_HEARTBEAT_REPORT_PLATFORM") or "binance").strip().lower() expected_strategy = (os.environ.get("RUNTIME_HEARTBEAT_STRATEGY_PROFILE") or "").strip() @@ -146,6 +153,8 @@ def _accepted_payload(payload: dict[str, Any]) -> tuple[bool, str]: return False, "strategy profile does not match" if expected_service and _payload_service_name(payload) != expected_service: return False, "service name does not match" + if _operational_blocker(payload) is not None: + return False, _OPERATIONAL_BLOCKER_REASON errors = payload.get("errors") if isinstance(errors, list) and errors: return False, "report contains errors" @@ -242,6 +251,17 @@ def assess_execution_report_heartbeat(now: dt.datetime | None = None) -> dict[st if payload is None: continue accepted, reason = _accepted_payload(payload) + if not accepted and reason == _OPERATIONAL_BLOCKER_REASON: + blocker = _operational_blocker(payload) + return { + "schema": _SCHEMA, + "observed_at": current.isoformat().replace("+00:00", "Z"), + "status": "alert", + "reason": f"execution_blocked:{blocker}", + "name": name, + "report_updated_at": (_entry_updated_at(entry) or current).isoformat().replace("+00:00", "Z"), + "reports_returned": len(entries), + } if accepted: assessment = { "schema": _SCHEMA, diff --git a/scripts/validate_runtime_startup.py b/scripts/validate_runtime_startup.py index e66814f6..1e7f36af 100644 --- a/scripts/validate_runtime_startup.py +++ b/scripts/validate_runtime_startup.py @@ -9,7 +9,6 @@ from pathlib import Path import sys - _SAFE_STARTUP_REASONS = { "runtime_recovery_not_active": "runtime_recovery_not_active", "recovery_control_state_invalid": "recovery_control_state_invalid", @@ -21,6 +20,29 @@ "full_cycle_state_loader_missing": "full_cycle_state_loader_missing", } +_EARN_FORWARD_REASON_CODES = frozenset({ + "earn_checkpoint_invalid", + "earn_checkpoint_time_invalid", + "earn_checkpoint_scope_changed", + "earn_product_lifecycle_unverified", + "earn_counter_reset", + "earn_quantity_change_unexplained", + "earn_order_unsettled", + "earn_accounted_changes_missing", + "earn_cash_cursor_mismatch", + "earn_cash_flow_invalid", + "earn_cash_flow_unsupported", + "earn_cash_flow_time_unverified", + "earn_valuation_snapshot_mismatch", + "external_cash_flow_window_invalid", + "external_cash_flow_cursor_invalid", + "external_cash_flow_history_read_failed", + "external_cash_flow_history_incomplete", + "external_cash_flow_record_invalid", + "external_cash_flow_record_changed", + "external_cash_flow_cursor_capacity_exceeded", +}) + _SAFE_CYCLE_FAILURE_STAGES = frozenset({ "client_connect", "state_load", @@ -142,6 +164,9 @@ def _full_cycle_failure_reason(report): return "full_cycle_aborted" failure = report.get("diagnostics", {}).get("cycle_failure", {}) stage = failure.get("stage") + reason_code = failure.get("reason_code") + if stage == "daily_state" and reason_code in _EARN_FORWARD_REASON_CODES: + return reason_code if stage in {"client_connect", "market_snapshot"}: return "full_cycle_broker_read_failed" if stage in {"fuel_execution", "trend_execution", "btc_execution", "earn_execution"}: diff --git a/tests/test_cycle_service.py b/tests/test_cycle_service.py index 96ea10ef..125ed692 100644 --- a/tests/test_cycle_service.py +++ b/tests/test_cycle_service.py @@ -357,6 +357,27 @@ def reconcile(*args): self.assertEqual(state["daily_equity_base"], 100.0) self.assertEqual(state["last_reset_date"], "2026-09-11") + def test_daily_state_reason_code_is_preserved_in_cycle_failure_metadata(self): + def reconcile(*args): + args[2]["diagnostics"] = { + "earn_accrual": { + "status": "blocked", + "reason_code": "earn_counter_reset", + } + } + raise ExecutionIntegrityError("earn_forward_accounting_unverified") + + report, _events = self._run_funds_cycle(True, rebase_fn=reconcile) + + self.assertEqual( + report["diagnostics"]["cycle_failure"], + { + "stage": "daily_state", + "error_type": "execution_integrity_error", + "reason_code": "earn_counter_reset", + }, + ) + def test_reconciled_new_day_state_resumes_after_reset_write_failure_without_double_count(self): state = { "last_reset_date": "2026-09-11", diff --git a/tests/test_execution_report_heartbeat.py b/tests/test_execution_report_heartbeat.py index f917f00f..fed0072b 100644 --- a/tests/test_execution_report_heartbeat.py +++ b/tests/test_execution_report_heartbeat.py @@ -14,14 +14,17 @@ def _entry(updated: str, uri: str = "gs://reports/binance/crypto/2026-08/report. return {"url": uri, "metadata": {"updated": updated}} -def _report(status: str = "ok"): - return { +def _report(status: str = "ok", *, execution_blocked_reason: str | None = None): + report = { "platform": "binance", "status": status, "strategy_profile": "crypto_live_pool_rotation", "service_name": "binance-platform", "errors": [], } + if execution_blocked_reason is not None: + report["execution_blocked_reason"] = execution_blocked_reason + return report def _observed_report(): @@ -71,6 +74,55 @@ def test_healthy_when_recent_matching_report_is_accepted(self) -> None: self.assertEqual(result["reason"], "status=ok") self.assertNotIn("deployment", result) + def test_state_owner_busy_is_an_execution_alert(self) -> None: + now = dt.datetime(2026, 8, 30, 3, tzinfo=dt.timezone.utc) + with ( + patch.object( + heartbeat, + "_list_reports", + return_value=( + [ + _entry("2026-08-30T02:59:00Z", "gs://reports/new.json"), + _entry("2026-08-30T02:00:00Z", "gs://reports/old.json"), + ], + "gs://reports", + ), + ), + patch.object( + heartbeat, + "_read_report", + side_effect=[ + _report(execution_blocked_reason="state_owner_busy"), + _report(), + ], + ), + ): + result = heartbeat.assess_execution_report_heartbeat(now) + + self.assertEqual(result["status"], "alert") + self.assertEqual(result["reason"], "execution_blocked:state_owner_busy") + + def test_non_operational_execution_blockers_remain_accepted(self) -> None: + for status, blocked_reason in ( + ("ok", "risk_execution_not_permitted"), + ("no_action", None), + ): + with self.subTest(status=status, blocked_reason=blocked_reason): + payload = _report(status=status, execution_blocked_reason=blocked_reason) + accepted, reason = heartbeat._accepted_payload(payload) + + self.assertTrue(accepted) + self.assertEqual(reason, f"status={status}") + + def test_owner_busy_from_other_strategy_is_ignored(self) -> None: + payload = _report(execution_blocked_reason="state_owner_busy") + payload["strategy_profile"] = "other_strategy" + + accepted, reason = heartbeat._accepted_payload(payload) + + self.assertFalse(accepted) + self.assertEqual(reason, "strategy profile does not match") + def test_disabled_target_does_not_read_storage(self) -> None: os.environ["RUNTIME_TARGET_ENABLED"] = "false" with patch.object(heartbeat, "_list_reports") as list_reports: diff --git a/tests/test_forward_earn_accounting.py b/tests/test_forward_earn_accounting.py index 7f981972..a0c4ea2c 100644 --- a/tests/test_forward_earn_accounting.py +++ b/tests/test_forward_earn_accounting.py @@ -76,6 +76,49 @@ def test_unverified_forward_activity_does_not_mutate_state(kind): assert state == before +def test_forward_validation_reason_is_safe_and_reported(): + state, new, cash = materials() + new['assets']['BNB']['products']['BNB001']['realtime_rewards'] = '0' + report = {} + runtime = SimpleNamespace(client=object(), now_utc=NOW, earn_accrual_observation=new) + snapshot = {a: round(float(r['quantity']), 8) for a, r in new['assets'].items()} + + with pytest.raises(ExecutionIntegrityError): + maybe_rebase_daily_state_for_balance_change( + state, runtime, report, 1600.0, 0.0, snapshot, [], + collect_external_cash_flows_fn=lambda *a, **kw: cash, + runtime_set_trade_state_fn=lambda *a, **kw: None, + append_log_fn=lambda *a: None, translate_fn=lambda *a, **kw: '', + ) + + assert report['diagnostics']['earn_accrual'] == { + 'status': 'blocked', 'reason_code': 'earn_counter_reset', + } + + +def test_forward_unknown_exception_is_generic_and_does_not_leak_message(): + state, new, _cash = materials() + report = {} + runtime = SimpleNamespace(client=object(), now_utc=NOW, earn_accrual_observation=new) + snapshot = {a: round(float(r['quantity']), 8) for a, r in new['assets'].items()} + + def fail(*_args, **_kwargs): + raise ValueError('provider account=secret-token') + + with pytest.raises(ExecutionIntegrityError): + maybe_rebase_daily_state_for_balance_change( + state, runtime, report, 1600.0, 0.0, snapshot, [], + collect_external_cash_flows_fn=fail, + runtime_set_trade_state_fn=lambda *a, **kw: None, + append_log_fn=lambda *a: None, translate_fn=lambda *a, **kw: '', + ) + + assert report['diagnostics']['earn_accrual'] == { + 'status': 'blocked', 'reason_code': 'earn_forward_accounting_unverified', + } + assert 'secret-token' not in repr(report) + + def test_deposit_and_interest_are_separate_in_same_window(): state, new, cash = materials() new['assets']['USDT'].update(spot_free='110', quantity='110') diff --git a/tests/test_validate_runtime_startup.py b/tests/test_validate_runtime_startup.py index 4b05e972..c022ae78 100644 --- a/tests/test_validate_runtime_startup.py +++ b/tests/test_validate_runtime_startup.py @@ -286,6 +286,24 @@ def test_full_cycle_failure_projection_keeps_allowlisted_report_metadata(): }) == ("result", "RuntimeError") +def test_full_cycle_failure_projection_preserves_safe_daily_state_reason_code(): + from scripts.validate_runtime_startup import _full_cycle_failure_reason + + report = { + "status": "error", + "diagnostics": { + "cycle_failure": { + "stage": "daily_state", + "error_type": "execution_integrity_error", + "reason_code": "earn_counter_reset", + } + }, + } + assert _full_cycle_failure_reason(report) == "earn_counter_reset" + report["diagnostics"]["cycle_failure"]["reason_code"] = "provider account=secret-token" + assert _full_cycle_failure_reason(report) == "full_cycle_cycle_error" + + @pytest.mark.parametrize('message, expected', [ ('runtime_recovery_not_active', 'runtime_recovery_not_active'), ('recovery_control_state_invalid', 'recovery_control_state_invalid'), From 3be65d4edaba557d61df58d7dc9923ef01059203 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 13 Sep 2026 18:17:31 +0800 Subject: [PATCH 2/2] Add bounded Earn accounting diagnosis Co-Authored-By: Codex --- .github/workflows/main.yml | 4 +- README.md | 5 + docs/operator_runbook.md | 14 + scripts/migrate_daily_accounting_state.py | 341 ++++++++++++++++++++ tests/test_earn_forward_diagnosis.py | 359 ++++++++++++++++++++++ tests/test_runtime_workflow_security.py | 2 +- 6 files changed, 722 insertions(+), 3 deletions(-) create mode 100644 tests/test_earn_forward_diagnosis.py diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index dbef6a6b..470573f8 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -52,7 +52,7 @@ on: type: choice required: false default: none - options: [none, inspect, quiesce, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply] + options: [none, inspect, quiesce, earn-forward-diagnose, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply] proposal_recipient_certificate: description: "Public X.509 encryption certificate for rebase-proposal only; never provide a private key" type: string @@ -132,7 +132,7 @@ jobs: exit 1 fi ;; - inspect|quiesce|audit|preview|scope-preview|cash-flow-preview|rebase-proposal|rebase-apply|prospective-rebase-apply|apply) + inspect|quiesce|earn-forward-diagnose|audit|preview|scope-preview|cash-flow-preview|rebase-proposal|rebase-apply|prospective-rebase-apply|apply) if [ "${RECOVERY_ACTION_INPUT:-none}" != "none" ] || [ "${GITHUB_REF:-}" != "refs/heads/main" ] || [ "${RUNTIME_TARGET_ENABLED:-}" != "false" ] || [ "$RECONCILE_ONLY_INPUT" != "true" ] || [ "$VALIDATE_ONLY_INPUT" = "true" ] || [ "$DIAGNOSE_BALANCES_INPUT" = "true" ] || [ "$RECONCILE_PERSIST_INPUT" = "true" ]; then echo "::error::Accounting migration requires disabled main runtime, reconcile_only, and no other recovery or diagnostic mode." exit 1 diff --git a/README.md b/README.md index 5e5d823f..e0f70ea4 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,11 @@ uv run --no-sync python -m unittest discover -s tests -v - [`docs/binance_platform_rename_checklist.md`](docs/binance_platform_rename_checklist.md) - [`docs/operator_runbook.md`](docs/operator_runbook.md) +For a forward Earn accounting failure, the disabled `main` workflow has one +read-only `accounting_migration_action=earn-forward-diagnose` mode. It samples +from the current ledger checkpoint and reports bounded, redacted diagnostics; +it never clears an owner, writes accounting state, or grants execution. + ## Community and security - See [CONTRIBUTING.md](CONTRIBUTING.md) for pull request scope, local verification, and documentation expectations. diff --git a/docs/operator_runbook.md b/docs/operator_runbook.md index a0daf6b8..471dac94 100644 --- a/docs/operator_runbook.md +++ b/docs/operator_runbook.md @@ -271,6 +271,20 @@ or changing balances/ledger blocks the audit. A match does not reconcile the whole account, authorize migration, or restore trading; the existing zero-activity preview/apply checks remain unchanged. +When the forward Earn consumer reports `earn_quantity_change_unexplained`, use +the separate `accounting_migration_action=earn-forward-diagnose` once with the +runtime disabled on `main` and `reconcile_only=true`. It starts at the ledger's +current `earn_accrual_checkpoint`, reads current Spot/Flexible Earn, bounded +trades, the existing cash-flow cursor, and bounded reward history, then reads +the three Firestore documents again. It reports only asset names, directions, +product and reward counts, matching flags, owner existence, and fixed no-write +policy flags. An existing owner is observed and reported; it is never cleared +or bypassed. Any ledger, control, or owner change during sampling discards the +result. A matching BONUS or REALTIME record is diagnostic evidence only and +never a causal reconciliation or execution permission. The bounded trade net is +reconstructed from normalized `myTrades` quantity and price fields for +diagnosis only; it is not a complete fill or accounting proof. + The migration is a separate, one-time `Runtime` workflow mode for an old `trend_val` ledger. It does not activate recovery control, grant execution authority, clear the circuit-breaker latch, or reconstruct historical diff --git a/scripts/migrate_daily_accounting_state.py b/scripts/migrate_daily_accounting_state.py index fbaf8314..efe2f996 100644 --- a/scripts/migrate_daily_accounting_state.py +++ b/scripts/migrate_daily_accounting_state.py @@ -992,6 +992,328 @@ def _read_source(refs): return ledger, ledger_value, control_value +def _read_earn_diagnosis_source(refs): + """Read the three accounting documents without treating an owner as a blocker.""" + owner = refs["owner_ref"].get(retry=None) + ledger_snapshot = refs["ledger_ref"].get(retry=None) + control_snapshot = refs["control_ref"].get(retry=None) + if not ledger_snapshot.exists: + raise MigrationBlocked("earn_diagnosis_ledger_missing") + ledger = ledger_snapshot.to_dict() + control = control_snapshot.to_dict() if control_snapshot.exists else None + if control is not None and ( + not isinstance(control, Mapping) + or control.get("state") not in { + "RECONCILE_ONLY", "ACTIVE_LKG", "ROLLBACK_LKG", "PAUSED", "REDUCE_ONLY", + } + ): + raise MigrationBlocked("earn_diagnosis_control_invalid") + + def marker(snapshot): + return { + "exists": snapshot.exists, + "update_time": ( + _timestamp(snapshot.update_time) if snapshot.exists else None + ), + "value": snapshot.to_dict() if snapshot.exists else None, + } + + return { + "owner_exists": owner.exists, + "owner_marker": marker(owner), + "ledger": ledger, + "ledger_marker": marker(ledger_snapshot), + "control": control, + "control_marker": marker(control_snapshot), + } + + +def _diagnosis_decimal(value, *, signed=False): + from application.earn_accrual import _amount + + try: + return _amount(str(value), signed=signed) + except (TypeError, ValueError, InvalidOperation): + raise MigrationBlocked("earn_diagnosis_numeric_input_invalid") from None + + +def _direction(value): + return "INCREASE" if value > 0 else "DECREASE" if value < 0 else "UNCHANGED" + + +def _trade_net_diagnosis(observations, *, assets): + """Reconstruct signed Spot deltas from normalized, bounded myTrades rows.""" + net = {asset: Decimal(0) for asset in assets} + counts = {asset: 0 for asset in assets} + for trade in observations.recent_executions: + symbol = str(trade.get("symbol") or "").upper() + if not symbol.endswith("USDT") or len(symbol) <= 4: + raise MigrationBlocked("earn_diagnosis_trade_symbol_invalid") + asset = symbol[:-4] + if asset not in assets or "USDT" not in assets: + raise MigrationBlocked("earn_diagnosis_trade_asset_out_of_scope") + quantity = _diagnosis_decimal(trade.get("qty")) + price = _diagnosis_decimal(trade.get("price")) + commission = _diagnosis_decimal(trade.get("commission")) + commission_asset = str(trade.get("commission_asset") or "").upper() + if not commission_asset or commission_asset not in assets: + raise MigrationBlocked("earn_diagnosis_trade_asset_out_of_scope") + sign = Decimal(1) if trade.get("is_buyer") is True else Decimal(-1) + net[asset] += sign * quantity + net["USDT"] -= sign * quantity * price + net[commission_asset] -= commission + counts[asset] += 1 + return net, counts + + +def diagnose_earn_forward(refs, *, client, expected, now): + """Diagnose the current checkpoint window without changing any durable state.""" + from application.earn_accrual import ( + _time, + _validate, + collect_earn_checkpoint, + ) + + source = _read_earn_diagnosis_source(refs) + ledger = source["ledger"] + order_record = ledger.get("order_submission") + order_state_known = isinstance(order_record, Mapping) and order_record.get("state") in { + "RESERVED", "TERMINAL", + } + checkpoint = ledger.get("earn_accrual_checkpoint") + if not isinstance(checkpoint, Mapping): + raise MigrationBlocked("earn_diagnosis_checkpoint_missing") + try: + _validate(checkpoint) + checkpoint_at = _time(checkpoint["observed_at"]) + except (TypeError, ValueError): + raise MigrationBlocked("earn_diagnosis_checkpoint_invalid") from None + expected_scope = expected.get("account_scope_sha256") + if not isinstance(expected_scope, str) or not re.fullmatch(r"[0-9a-f]{64}", expected_scope): + raise MigrationBlocked("earn_diagnosis_account_scope_missing") + if checkpoint.get("account_scope_sha256") != expected_scope: + raise MigrationBlocked("earn_diagnosis_checkpoint_account_scope_mismatch") + if not isinstance(now, datetime) or now.tzinfo is None or not checkpoint_at < now: + raise MigrationBlocked("earn_diagnosis_checkpoint_window_invalid") + if now - checkpoint_at > timedelta(days=7): + raise MigrationBlocked("earn_diagnosis_checkpoint_window_invalid") + + assets = tuple(sorted(checkpoint["assets"])) + if "USDT" not in assets or not 0 < len(assets) <= 32: + raise MigrationBlocked("earn_diagnosis_asset_scope_invalid") + net_values = ledger.get("earn_accounted_net_changes") + if not isinstance(net_values, Mapping) or set(net_values) != set(assets): + raise MigrationBlocked("earn_diagnosis_net_changes_missing") + stored_net = { + asset: _diagnosis_decimal(net_values[asset], signed=True) for asset in assets + } + cursor = ledger.get("external_cash_flow_cursor") + if not isinstance(cursor, Mapping): + raise MigrationBlocked("earn_diagnosis_cursor_missing") + try: + if _time(cursor.get("observed_at")) != checkpoint_at: + raise ValueError + except (TypeError, ValueError): + raise MigrationBlocked("earn_diagnosis_cursor_mismatch") from None + + required_symbols = tuple(f"{asset}USDT" for asset in assets if asset != "USDT") + configured_symbols = set(_symbols_from_env()) + if any(symbol not in configured_symbols for symbol in required_symbols): + raise MigrationBlocked("earn_diagnosis_symbols_missing") + + try: + current = collect_earn_checkpoint( + client, + assets=assets, + observed_at=now, + expected_account_scope_sha256=expected_scope, + ) + except Exception: + raise MigrationBlocked("earn_diagnosis_checkpoint_unavailable") from None + + try: + flows = collect_spot_usdt_external_cash_flows( + client, now=now, cursor=copy.deepcopy(cursor) + ) + except Exception: + raise MigrationBlocked("earn_diagnosis_external_flow_unavailable") from None + + try: + observations = collect_read_only_reconciliation_observations( + client, + strategy_symbols=required_symbols, + local_execution_ledger=ledger, + now=now, + lookback=now - checkpoint_at, + ) + except Exception: + raise MigrationBlocked("earn_diagnosis_trade_history_unavailable") from None + if digest(observations.account_scope) != expected_scope: + raise MigrationBlocked("earn_diagnosis_account_scope_mismatch") + trade_net, trade_counts = _trade_net_diagnosis(observations, assets=assets) + + external_principal = _diagnosis_decimal( + flows.get("new_deposit_principal_usdt"), signed=False + ) + if flows.get("new_unsupported_deposit_count", 0) or flows.get( + "new_or_changed_withdrawal_count", 0 + ): + external_status = "UNSUPPORTED_ACTIVITY" + else: + external_status = "OBSERVED" + + observed_delta = {} + residual_before_realtime = {} + residual_after_realtime = {} + product_status = {} + realtime_delta = {} + for asset in assets: + before = checkpoint["assets"][asset] + after = current["assets"].get(asset) + if not isinstance(after, Mapping): + raise MigrationBlocked("earn_diagnosis_checkpoint_asset_missing") + observed_delta[asset] = _diagnosis_decimal(after["quantity"], signed=True) - _diagnosis_decimal( + before["quantity"], signed=True + ) + before_products, after_products = before["products"], after["products"] + if set(before_products) != set(after_products): + product_status[asset] = "PRODUCT_LIFECYCLE_CHANGED" + realtime_delta[asset] = None + else: + counter_delta = Decimal(0) + status = "STABLE" + for product in before_products: + old_row, new_row = before_products[product], after_products[product] + if old_row["auto_subscribe"] != new_row["auto_subscribe"]: + status = "PRODUCT_LIFECYCLE_CHANGED" + break + delta = _diagnosis_decimal(new_row["realtime_rewards"], signed=True) - _diagnosis_decimal( + old_row["realtime_rewards"], signed=True + ) + if delta < 0: + status = "COUNTER_RESET" + break + counter_delta += delta + product_status[asset] = status + realtime_delta[asset] = counter_delta if status == "STABLE" else None + residual_before_realtime[asset] = observed_delta[asset] - stored_net[asset] - ( + external_principal if asset == "USDT" else Decimal(0) + ) + residual_after_realtime[asset] = residual_before_realtime[asset] - ( + realtime_delta[asset] if realtime_delta[asset] is not None else Decimal(0) + ) + + history = diagnose_balance_flows( + client, + start=checkpoint_at, + end=now, + now=now, + reward_quantity_changes=residual_after_realtime, + ) + if history.get("history_complete_for_requested_surfaces") is not True: + reason = history.get("reason_code") + if reason == "balance_history_reward_validation_failed": + raise MigrationBlocked("earn_diagnosis_reward_history_invalid") + raise MigrationBlocked("earn_diagnosis_history_incomplete") + reward_checks = history.get("reward_quantity_checks") + if not isinstance(reward_checks, Mapping) or set(reward_checks) != set(assets): + raise MigrationBlocked("earn_diagnosis_reward_history_invalid") + + asset_results = {} + for asset in assets: + check = reward_checks.get(asset) + if not isinstance(check, Mapping) or not isinstance(check.get("reward_counts"), Mapping): + raise MigrationBlocked("earn_diagnosis_reward_history_invalid") + counts = check["reward_counts"] + if any(type(counts.get(kind)) is not int or counts[kind] < 0 for kind in ("BONUS", "REALTIME")): + raise MigrationBlocked("earn_diagnosis_reward_history_invalid") + if product_status[asset] == "PRODUCT_LIFECYCLE_CHANGED": + classification = "product_lifecycle_changed" + elif product_status[asset] == "COUNTER_RESET": + classification = "counter_reset" + elif trade_net[asset] != stored_net[asset]: + classification = "trade_net_unmatched" + elif external_status == "UNSUPPORTED_ACTIVITY" and asset == "USDT": + classification = "external_flow_unsupported" + elif product_status[asset] == "STABLE" and residual_after_realtime[asset] == 0: + classification = "residual_zero_after_realtime_counter" + elif ( + product_status[asset] == "STABLE" + and residual_after_realtime[asset] > 0 + and counts["BONUS"] > 0 + and check.get("delta_matches_bonus") + ): + classification = "residual_matches_bonus_records" + else: + classification = "quantity_unexplained" + realtime_status = product_status[asset] + if realtime_delta[asset] is not None: + realtime_status = "INCREASE" if realtime_delta[asset] > 0 else "UNCHANGED" + external_matches = None + if asset == "USDT" and realtime_delta[asset] is not None: + external_matches = ( + observed_delta[asset] - stored_net[asset] - realtime_delta[asset] + == external_principal + ) + asset_results[asset] = { + "quantity_direction": _direction(observed_delta[asset]), + "residual_direction": _direction(residual_after_realtime[asset]), + "product_status": product_status[asset], + "product_count_before": len(checkpoint["assets"][asset]["products"]), + "product_count_current": len(current["assets"][asset]["products"]), + "realtime_counter_status": realtime_status, + "bonus_record_count": counts["BONUS"], + "realtime_record_count": counts["REALTIME"], + "residual_matches_bonus_records": ( + None + if product_status[asset] != "STABLE" + else ( + residual_after_realtime[asset] > 0 + and counts["BONUS"] > 0 + and check.get("delta_matches_bonus") is True + ) + ), + "trade_count": trade_counts[asset], + "trade_net_matches_persisted": trade_net[asset] == stored_net[asset], + "trade_net_diagnostic_only": True, + "external_flow_matches_residual": external_matches, + "external_flow_count": flows.get("new_confirmed_deposit_count", 0) if asset == "USDT" else 0, + "classification": classification, + "causal_reconciliation": False, + } + + after = _read_earn_diagnosis_source(refs) + if any( + source[key] != after[key] + for key in ("owner_marker", "ledger_marker", "control_marker") + ): + raise MigrationBlocked("earn_diagnosis_state_changed_during_read") + return { + "status": "diagnosed", + "stage": "earn_forward_accounting_diagnosis", + "owner_exists": source["owner_exists"], + "owner_unchanged": True, + "ledger_unchanged": True, + "control_unchanged": True, + "sampling_stable": True, + "account_scope_verified": True, + "checkpoint_window_within_seven_days": True, + "order_state_known": order_state_known, + "diagnostic_restricted": not order_state_known, + "assets": asset_results, + "history_counts": history["history_counts"], + "open_order_count": len(observations.open_orders), + "unsupported_external_flow_count": flows.get("new_unsupported_deposit_count", 0), + "changed_withdrawal_count": flows.get("new_or_changed_withdrawal_count", 0), + "external_flow_status": external_status, + "causal_reconciliation": False, + "activation_allowed": False, + "no_order": True, + "write_performed": False, + "execution_authority_granted": False, + } + + def _private_spot_account(account, *, expected_account_scope_sha256): if not isinstance(account, Mapping): raise MigrationBlocked("private_scope_balance_invalid") @@ -1458,6 +1780,24 @@ def run(action: str, *, expected_digest: str = "", now: datetime | None = None): return inspect_control(_refs()) if action == "quiesce": return _quiesce_control(_refs()) + if action == "earn-forward-diagnose": + now = now or datetime.now(timezone.utc) + target = resolve_runtime_target_from_env( + env=os.environ, expected_platform_id="binance" + ) + if ( + str(getattr(getattr(target, "live_continuity", None), "state", "")).upper() + != "RECONCILE_ONLY" + ): + raise MigrationBlocked("runtime_target_not_reconcile_only") + expected = _expected_digests() + if not expected: + raise MigrationBlocked("expected_account_scope_missing") + refs = _refs() + client = connect_client( + os.environ["BINANCE_API_KEY"], os.environ["BINANCE_API_SECRET"], timeout=30 + ) + return diagnose_earn_forward(refs, client=client, expected=expected, now=now) publication = ( _private_scope_publication_context() if action == "scope-preview" else None ) @@ -1599,6 +1939,7 @@ def main(argv=None) -> int: choices=( "inspect", "quiesce", + "earn-forward-diagnose", "audit", "preview", "scope-preview", diff --git a/tests/test_earn_forward_diagnosis.py b/tests/test_earn_forward_diagnosis.py new file mode 100644 index 00000000..4fdf989b --- /dev/null +++ b/tests/test_earn_forward_diagnosis.py @@ -0,0 +1,359 @@ +import copy +import json +from datetime import datetime, timezone +from types import SimpleNamespace + +import pytest + + +NOW = datetime(2026, 9, 13, 10, 0, tzinfo=timezone.utc) +SCOPE = "a" * 64 + + +class Snapshot: + def __init__(self, value, update_time="2026-09-13T09:59:00Z"): + self.value = value + self.exists = value is not None + self.update_time = update_time + + def to_dict(self): + return copy.deepcopy(self.value) + + +class Ref: + def __init__(self, snapshot): + self.snapshot = snapshot + + def get(self, **_kwargs): + return self.snapshot + + +def checkpoint(*, observed_at, quantity="1", reward="0"): + return { + "account_scope_sha256": SCOPE, + "observed_at": observed_at, + "assets": { + "USDT": { + "spot_free": "10", + "spot_locked": "0", + "products": {}, + "quantity": "10", + }, + "BTC": { + "spot_free": "0", + "spot_locked": "0", + "products": { + "BTC001": { + "total": quantity, + "realtime_rewards": reward, + "auto_subscribe": False, + "can_redeem": True, + } + }, + "quantity": quantity, + }, + }, + "execution_authority_granted": False, + } + + +def source(*, owner=True, checkpoint_value=None): + checkpoint_value = checkpoint_value or checkpoint( + observed_at="2026-09-13T09:00:00+00:00" + ) + ledger = { + "earn_accrual_checkpoint": checkpoint_value, + "earn_accounted_net_changes": {"USDT": "0", "BTC": "0"}, + "external_cash_flow_cursor": { + "version": 1, + "observed_at": checkpoint_value["observed_at"], + "records": {}, + }, + "order_submission": {"state": "TERMINAL"}, + } + return { + "ledger_ref": Ref(Snapshot(ledger)), + "owner_ref": Ref(Snapshot({"owner": "held"}) if owner else Snapshot(None)), + "control_ref": Ref(Snapshot({"state": "RECONCILE_ONLY"})), + } + + +def install_read_stubs(monkeypatch, migration, refs, *, current=None, history=None, observations=None, flows=None): + current = current or checkpoint( + observed_at="2026-09-13T10:00:00+00:00", quantity="1.1", reward="0" + ) + history = history or { + "history_complete_for_requested_surfaces": True, + "history_counts": {"earn_rewards": 1}, + "reward_quantity_checks": { + "USDT": { + "delta_matches_bonus": False, + "delta_matches_realtime": False, + "delta_matches_visible_total": False, + "reward_counts": {"BONUS": 0, "REALTIME": 0}, + }, + "BTC": { + "delta_matches_bonus": True, + "delta_matches_realtime": False, + "delta_matches_visible_total": True, + "reward_counts": {"BONUS": 1, "REALTIME": 0}, + }, + }, + } + observations = observations or SimpleNamespace( + account_scope={"account_uid": "123"}, + recent_executions=(), + open_orders=(), + ) + flows = flows or { + "new_deposit_principal_usdt": "0", + "new_confirmed_deposit_count": 0, + "new_unsupported_deposit_count": 0, + "new_or_changed_withdrawal_count": 0, + "cursor": {"version": 1, "observed_at": "2026-09-13T10:00:00+00:00", "records": {}}, + } + monkeypatch.setattr(migration, "_refs", lambda: refs) + monkeypatch.setattr(migration, "_symbols_from_env", lambda: ("BTCUSDT",)) + monkeypatch.setattr(migration, "_expected_digests", lambda: {"account_scope_sha256": SCOPE}) + monkeypatch.setattr(migration, "digest", lambda _value: SCOPE) + monkeypatch.setattr( + "application.earn_accrual.collect_earn_checkpoint", + lambda *args, **kwargs: copy.deepcopy(current), + ) + monkeypatch.setattr(migration, "collect_spot_usdt_external_cash_flows", lambda *a, **k: copy.deepcopy(flows)) + monkeypatch.setattr(migration, "collect_read_only_reconciliation_observations", lambda *a, **k: observations) + monkeypatch.setattr(migration, "diagnose_balance_flows", lambda *a, **k: copy.deepcopy(history)) + return current, history + + +def test_existing_owner_is_reported_but_does_not_block_read_only_diagnosis(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source(owner=True) + refs["control_ref"].snapshot.value["state"] = "ACTIVE_LKG" + install_read_stubs(monkeypatch, migration, refs) + result = migration.diagnose_earn_forward( + refs, + client=object(), + expected={"account_scope_sha256": SCOPE}, + now=NOW, + ) + + assert result["owner_exists"] is True + assert result["activation_allowed"] is False + assert result["write_performed"] is False + assert result["assets"]["BTC"]["residual_matches_bonus_records"] is True + assert result["assets"]["BTC"]["trade_count"] == 0 + assert result["assets"]["BTC"]["trade_net_matches_persisted"] is True + assert result["assets"]["BTC"]["trade_net_diagnostic_only"] is True + assert result["assets"]["USDT"]["external_flow_matches_residual"] is True + assert refs["ledger_ref"].snapshot.value["earn_accounted_net_changes"] == {"USDT": "0", "BTC": "0"} + + +def test_realtime_counter_is_subtracted_before_bonus_residual_match(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + current = checkpoint( + observed_at="2026-09-13T10:00:00+00:00", quantity="3", reward="1" + ) + history = { + "history_complete_for_requested_surfaces": True, + "history_counts": {"earn_rewards": 2}, + "reward_quantity_checks": { + "USDT": { + "delta_matches_bonus": False, + "delta_matches_realtime": False, + "delta_matches_visible_total": False, + "reward_counts": {"BONUS": 0, "REALTIME": 0}, + }, + "BTC": { + "delta_matches_bonus": True, + "delta_matches_realtime": False, + "delta_matches_visible_total": False, + "reward_counts": {"BONUS": 1, "REALTIME": 1}, + }, + }, + } + install_read_stubs(monkeypatch, migration, refs, current=current, history=history) + result = migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + + btc = result["assets"]["BTC"] + assert btc["realtime_counter_status"] == "INCREASE" + assert btc["realtime_record_count"] == 1 + assert btc["residual_matches_bonus_records"] is True + assert btc["classification"] == "residual_matches_bonus_records" + assert btc["causal_reconciliation"] is False + + +def test_realtime_only_change_is_zero_after_counter_and_not_bonus(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + current = checkpoint( + observed_at="2026-09-13T10:00:00+00:00", quantity="2", reward="1" + ) + history = { + "history_complete_for_requested_surfaces": True, + "history_counts": {"earn_rewards": 1}, + "reward_quantity_checks": { + "USDT": { + "delta_matches_bonus": False, + "delta_matches_realtime": False, + "delta_matches_visible_total": False, + "reward_counts": {"BONUS": 0, "REALTIME": 0}, + }, + "BTC": { + "delta_matches_bonus": True, + "delta_matches_realtime": False, + "delta_matches_visible_total": False, + "reward_counts": {"BONUS": 0, "REALTIME": 1}, + }, + }, + } + install_read_stubs(monkeypatch, migration, refs, current=current, history=history) + from application.broker_reconciliation import diagnose_balance_flows as real_diagnose_balance_flows + + class EmptyHistoryClient: + def _request_margin_api(self, _method, path, *, signed, data): + assert signed is True + if path in {"capital/deposit/hisrec", "capital/withdraw/history"}: + return [] + return {"rows": [], "total": 0} + + monkeypatch.setattr( + migration, + "diagnose_balance_flows", + lambda _client, **kwargs: real_diagnose_balance_flows( + EmptyHistoryClient(), **kwargs + ), + ) + result = migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + + btc = result["assets"]["BTC"] + assert btc["residual_direction"] == "UNCHANGED" + assert btc["residual_matches_bonus_records"] is False + assert btc["classification"] == "residual_zero_after_realtime_counter" + + +def test_unknown_order_state_returns_restricted_diagnostic(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + refs["ledger_ref"].snapshot.value["order_submission"] = {"state": "UNKNOWN"} + install_read_stubs(monkeypatch, migration, refs) + result = migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + + assert result["order_state_known"] is False + assert result["diagnostic_restricted"] is True + assert result["activation_allowed"] is False + + +def test_balance_difference_without_matching_history_stays_unexplained(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + history = { + "history_complete_for_requested_surfaces": True, + "history_counts": {"earn_rewards": 0}, + "reward_quantity_checks": { + asset: { + "delta_matches_bonus": False, + "delta_matches_realtime": False, + "delta_matches_visible_total": False, + "reward_counts": {"BONUS": 0, "REALTIME": 0}, + } + for asset in ("USDT", "BTC") + }, + } + install_read_stubs(monkeypatch, migration, refs, history=history) + result = migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + + assert result["assets"]["BTC"]["classification"] == "quantity_unexplained" + assert result["assets"]["BTC"]["causal_reconciliation"] is False + assert result["causal_reconciliation"] is False + + +def test_incomplete_or_duplicate_reward_history_is_rejected(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + history = { + "history_complete_for_requested_surfaces": False, + "reason_code": "balance_history_reward_validation_failed", + } + install_read_stubs(monkeypatch, migration, refs, history=history) + with pytest.raises(migration.MigrationBlocked, match="earn_diagnosis_reward_history_invalid"): + migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + + +def test_account_scope_mismatch_is_rejected(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + observations = SimpleNamespace( + account_scope={"account_uid": "wrong"}, recent_executions=(), open_orders=() + ) + install_read_stubs(monkeypatch, migration, refs, observations=observations) + monkeypatch.setattr( + migration, + "digest", + lambda value: "b" * 64 if value == {"account_uid": "wrong"} else SCOPE, + ) + with pytest.raises(migration.MigrationBlocked, match="earn_diagnosis_account_scope_mismatch"): + migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + + +def test_state_change_discards_diagnosis(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + def history(*_args, **_kwargs): + refs["ledger_ref"].snapshot.value["changed"] = True + return { + "history_complete_for_requested_surfaces": True, + "history_counts": {"earn_rewards": 0}, + "reward_quantity_checks": { + asset: { + "delta_matches_bonus": False, + "delta_matches_realtime": False, + "delta_matches_visible_total": False, + "reward_counts": {"BONUS": 0, "REALTIME": 0}, + } + for asset in ("USDT", "BTC") + }, + } + install_read_stubs(monkeypatch, migration, refs) + monkeypatch.setattr(migration, "diagnose_balance_flows", history) + with pytest.raises(migration.MigrationBlocked, match="earn_diagnosis_state_changed_during_read"): + migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + + +def test_diagnosis_output_contains_no_amounts_uids_or_raw_rows(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = source() + install_read_stubs(monkeypatch, migration, refs) + result = migration.diagnose_earn_forward( + refs, client=object(), expected={"account_scope_sha256": SCOPE}, now=NOW + ) + output = json.dumps(result, sort_keys=True) + assert "1.1" not in output + assert "123" not in output + assert "held" not in output + assert "BTC001" not in output + assert "cumulativeRealTimeRewards" not in output diff --git a/tests/test_runtime_workflow_security.py b/tests/test_runtime_workflow_security.py index 97b98d32..6574920f 100644 --- a/tests/test_runtime_workflow_security.py +++ b/tests/test_runtime_workflow_security.py @@ -333,7 +333,7 @@ def test_accounting_migration_has_explicit_preview_and_apply_inputs(): inputs = workflow[workflow.index(" inputs:") : workflow.index("permissions:")] assert "accounting_migration_action:" in inputs - assert "options: [none, inspect, quiesce, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply]" in inputs + assert "options: [none, inspect, quiesce, earn-forward-diagnose, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply]" in inputs assert "accounting_migration_preview_run_id:" in inputs assert "accounting_migration_expected_digest:" in inputs