diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 9e9f321d..c1d32997 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -441,6 +441,7 @@ jobs: GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} NOTIFY_LANG: ${{ vars.NOTIFY_LANG }} EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI }} + BINANCE_LIFECYCLE_EXPORT_PATH: ${{ runner.temp }}/binance-lifecycle-${{ github.run_id }}-${{ github.run_attempt }}/lifecycle-run.json BINANCE_DRY_RUN: ${{ vars.BINANCE_DRY_RUN || 'true' }} STRATEGY_ARTIFACT_FILE: ${{ vars.STRATEGY_ARTIFACT_FILE }} STRATEGY_ARTIFACT_MANIFEST_FILE: ${{ vars.STRATEGY_ARTIFACT_MANIFEST_FILE }} @@ -508,6 +509,16 @@ jobs: if-no-files-found: warn retention-days: 1 + - name: 5a. Stage redacted lifecycle recorder output for monitor + continue-on-error: true + if: ${{ always() && github.event.inputs.validate_only != 'true' && github.event.inputs.reconcile_only != 'true' && env.RUNTIME_TARGET_ENABLED == 'true' }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: binance-live-run-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/binance-lifecycle-${{ github.run_id }}-${{ github.run_attempt }}/lifecycle-run.json + if-no-files-found: warn + retention-days: 7 + - name: 5b. Retain redacted reconciliation candidate if: ${{ always() && github.event.inputs.reconcile_only == 'true' && github.event.inputs.reconcile_persist_candidate == 'true' }} uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 diff --git a/application/cycle_service.py b/application/cycle_service.py index 8ce51b23..05746ca2 100644 --- a/application/cycle_service.py +++ b/application/cycle_service.py @@ -5,9 +5,15 @@ import json import os from collections.abc import Mapping +from datetime import datetime, timezone +from pathlib import Path +import tempfile 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 quant_platform_kit.strategy_lifecycle.performance_monitor import ( + resolve_lifecycle_stream_id, + 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 @@ -39,6 +45,100 @@ def _settled_order_state(state): return status if status in {"RESERVED", "TERMINAL"} else None +def _build_platform_execution_result(report, *, state_healthy, state_owner_release_uncertain): + return { + "platform": "binance", + "status": report.get("status"), + "total_equity_usdt": report.get("total_equity_usdt"), + "trend_equity_usdt": report.get("trend_equity_usdt"), + # The local daily-loss state excludes supported deposits, but this + # per-cycle record has no exactly-once external-flow delivery. + "external_cash_flow": None, + "external_cash_flow_interval": ( + report.get("external_cash_flow_interval") + if report.get("status") == "ok" + and state_healthy + and not state_owner_release_uncertain + else None + ), + "degraded_mode_level": report.get("degraded_mode_level"), + # Preserve the existing recorder payload exactly; the separate export + # applies its own safe-field allowlist and omits this field. + "error": report.get("error"), + } + + +def _write_lifecycle_export(profile_id, execution_result): + """Best-effort, redacted export of the current recorder payload. + + The normal PerformanceStore path remains authoritative. This optional + file is a short-lived transport handoff for the monitor and must never + change the cycle result or expose report details and provider errors. + """ + raw_path = str(os.environ.get("BINANCE_LIFECYCLE_EXPORT_PATH") or "").strip() + if not raw_path: + return + path = Path(raw_path) + try: + if path.is_symlink() or (path.exists() and not path.is_file()): + return + if path.parent.is_symlink(): + return + path.parent.mkdir(parents=True, exist_ok=True) + if path.is_symlink() or (path.exists() and not path.is_file()): + return + profile = str(profile_id or "").strip() + if not profile: + return + stream_id = resolve_lifecycle_stream_id(execution_result=execution_result) + safe_result = { + key: execution_result.get(key) + for key in ( + "platform", + "status", + "total_equity_usdt", + "trend_equity_usdt", + "external_cash_flow", + "external_cash_flow_interval", + "degraded_mode_level", + ) + } + if safe_result.get("status") != "ok": + safe_result["error_code"] = "cycle_failed" + payload = { + "strategy_profile": profile, + "domain": "crypto", + "recorded_at": datetime.now(timezone.utc).isoformat(), + "record_kind": "execution", + "execution_result": safe_result, + "lifecycle_stream_id": stream_id, + "schema_version": "strategy_lifecycle.v1", + } + fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False, indent=2) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_name, path) + os.chmod(path, 0o600) + except Exception: + try: + os.close(fd) + except OSError: + pass + try: + os.unlink(temporary_name) + except OSError: + pass + except Exception: + # Monitoring transport is deliberately non-blocking for the trading + # cycle; the authoritative recorder call remains unchanged. + return + + def execute_strategy_cycle( runtime, *, @@ -86,6 +186,7 @@ def execute_strategy_cycle( ) state_healthy = False + state_owner_release_uncertain = False failure_stage = "state_owner_claim" owner_claimed_this_cycle = False initial_order_state = None @@ -465,34 +566,34 @@ def execute_strategy_cycle( try: release_runtime_state_owner(runtime) except ExecutionIntegrityError: + state_owner_release_uncertain = True report["status"] = "error" append_report_error(report, "state_owner_release_uncertain", stage="state_release") elif state_healthy and getattr(runtime, "state_owner_held", False): try: release_runtime_state_owner(runtime) except ExecutionIntegrityError: + state_owner_release_uncertain = True report["status"] = "error" append_report_error(report, "state_owner_release_uncertain", stage="state_release") report["log_lines"] = list(log_buffer) finalize_notification_delivery(report) attach_execution_receipt_from_report(report) if not getattr(runtime, "dry_run", False): + execution_result = _build_platform_execution_result( + report, + state_healthy=state_healthy, + state_owner_release_uncertain=state_owner_release_uncertain, + ) try_record_platform_execution( str(getattr(runtime, "strategy_profile", "") or ""), - { - "platform": "binance", - "status": report.get("status"), - "total_equity_usdt": report.get("total_equity_usdt"), - "trend_equity_usdt": report.get("trend_equity_usdt"), - # The local daily-loss state excludes supported deposits, but - # this per-cycle record has no exactly-once external-flow - # delivery. Keep cross-cycle performance explicitly incomparable. - "external_cash_flow": None, - "degraded_mode_level": report.get("degraded_mode_level"), - "error": report.get("error"), - }, + execution_result, domain="crypto", ) + _write_lifecycle_export( + str(getattr(runtime, "strategy_profile", "") or ""), + execution_result, + ) # Early returns (including risk rejection) also complete a cycle. try: diff --git a/application/portfolio_service.py b/application/portfolio_service.py index fc0f0228..610a76fe 100644 --- a/application/portfolio_service.py +++ b/application/portfolio_service.py @@ -139,10 +139,28 @@ def maybe_rebase_daily_state_for_balance_change( "reason_code": reason_code, } raise ExecutionIntegrityError("earn_forward_accounting_unverified") from None + previous = state["earn_accrual_checkpoint"] + try: + principal = Decimal(str(cash["new_deposit_principal_usdt"])) + frozen_equity = Decimal(str(total_equity)) + except (KeyError, TypeError, InvalidOperation): + raise ExecutionIntegrityError("earn_forward_accounting_unverified") from None + if not principal.is_finite() or not frozen_equity.is_finite() or frozen_equity <= 0: + raise ExecutionIntegrityError("earn_forward_accounting_unverified") + interval = { + "account_scope_sha256": previous["account_scope_sha256"], + "start_at": previous["observed_at"], + "end_at": current["observed_at"], + "end_equity_usdt": format(frozen_equity, "f"), + "net_external_cash_flow": format(principal, "f"), + "currency": "USDT", + "valuation_basis": "checkpoint_quantities_sampled_prices", + } runtime_set_trade_state_fn(runtime, report, updated, reason="earn_forward_accounting") state.clear() state.update(updated) runtime.trade_state = state + report["external_cash_flow_interval"] = interval report.setdefault("diagnostics", {})["earn_accrual"] = {"status": "reconciled"} return True diff --git a/docs/operator_runbook.md b/docs/operator_runbook.md index 1a94b5c1..c6deb211 100644 --- a/docs/operator_runbook.md +++ b/docs/operator_runbook.md @@ -744,12 +744,15 @@ difference, resets the breaker, or grants execution authority. A supported deposit whose completion arrives outside the current accounting day is held for operator review rather than posted to a later day. -This adjustment protects the local daily-loss calculation only. The existing -per-cycle performance record has no exactly-once delivery for a cash-flow amount: -emitting a daily cumulative value would double count it, while emitting it once -could lose it if the performance write failed after the private cursor advanced. -Binance therefore records `external_cash_flow=null` and remains incomparable in -cross-cycle performance monitoring until that separate durable contract exists. +This adjustment protects the local daily-loss calculation and leaves the legacy +per-cycle `external_cash_flow` field null. After a verified Earn forward state +write, the cycle also records a seven-field checkpoint interval with the frozen +opening equity and verified USDT deposit principal. The interval return uses an +end-of-observation flow assumption; it is not exact TWR and does not represent +a midnight natural-day return. The existing Binance recorder pin can pass this +field through unchanged, while numeric monitoring requires a deployed QPK +revision that consumes the interval contract. Negative-flow tests document the +signed consumer contract and do not enable withdrawal handling. While the account remains disabled, dispatch the existing Runtime workflow with `reconcile_only=true` and `accounting_migration_action=cash-flow-preview` to diff --git a/tests/test_cycle_service.py b/tests/test_cycle_service.py index ee1fcead..0379df84 100644 --- a/tests/test_cycle_service.py +++ b/tests/test_cycle_service.py @@ -9,7 +9,12 @@ from types import SimpleNamespace from unittest.mock import Mock, patch -from application.cycle_service import execute_strategy_cycle, run_live_cycle, write_execution_report +from application.cycle_service import ( + _write_lifecycle_export, + execute_strategy_cycle, + run_live_cycle, + write_execution_report, +) from application.execution_service import execute_trend_buys from application.portfolio_service import ( maybe_rebase_daily_state_for_balance_change, @@ -311,6 +316,64 @@ def test_platform_performance_record_marks_external_cash_flow_incomparable(self) record.assert_called_once() self.assertIn("external_cash_flow", record.call_args.args[1]) self.assertIsNone(record.call_args.args[1]["external_cash_flow"]) + self.assertIn("error", record.call_args.args[1]) + + def test_platform_performance_record_forwards_interval_only_on_success(self): + interval = {"account_scope_sha256": "a" * 64, "start_at": "start", "end_at": "end"} + + def rebase(*args): + args[2]["external_cash_flow_interval"] = interval + + with patch("application.cycle_service.try_record_platform_execution") as record: + self._run_funds_cycle(True, rebase_fn=rebase) + self.assertEqual(record.call_args.args[1]["external_cash_flow_interval"], interval) + + def test_platform_performance_record_drops_interval_after_later_cycle_error(self): + interval = {"account_scope_sha256": "a" * 64, "start_at": "start", "end_at": "end"} + + def rebase(*args): + args[2]["external_cash_flow_interval"] = interval + + with patch("application.cycle_service.try_record_platform_execution") as record: + self._run_funds_cycle(True, rebase_fn=rebase, earn_failure=True) + self.assertIsNone(record.call_args.args[1]["external_cash_flow_interval"]) + + def test_lifecycle_export_is_optional_and_redacts_provider_details(self): + with tempfile.TemporaryDirectory() as directory: + export_path = os.path.join(directory, "lifecycle-run.json") + execution_result = { + "platform": "binance", + "status": "error", + "total_equity_usdt": 100.0, + "trend_equity_usdt": 40.0, + "external_cash_flow": None, + "external_cash_flow_interval": None, + "degraded_mode_level": None, + "error": "provider-secret-must-not-export", + "orders": [{"symbol": "BTCUSDT"}], + } + with patch.dict(os.environ, {"BINANCE_LIFECYCLE_EXPORT_PATH": export_path}): + _write_lifecycle_export("crypto_live_pool_rotation", execution_result) + + payload = json.loads(open(export_path, encoding="utf-8").read()) + self.assertEqual(payload["domain"], "crypto") + self.assertEqual(payload["record_kind"], "execution") + self.assertEqual(payload["lifecycle_stream_id"], "binance") + self.assertIsNone(payload["execution_result"]["external_cash_flow_interval"]) + self.assertEqual(payload["execution_result"]["error_code"], "cycle_failed") + self.assertNotIn("provider-secret-must-not-export", json.dumps(payload)) + self.assertNotIn("orders", json.dumps(payload)) + self.assertEqual(os.stat(export_path).st_mode & 0o777, 0o600) + + def test_lifecycle_export_without_authorized_path_does_not_write(self): + with tempfile.TemporaryDirectory() as directory: + export_path = os.path.join(directory, "lifecycle-run.json") + with patch.dict(os.environ, {"BINANCE_LIFECYCLE_EXPORT_PATH": ""}): + _write_lifecycle_export( + "crypto_live_pool_rotation", + {"platform": "binance", "status": "ok", "external_cash_flow_interval": None}, + ) + self.assertFalse(os.path.exists(export_path)) def test_approved_execution_permission_preserves_fuel_trend_dca_and_earn_actions(self): _report, events = self._run_funds_cycle(True) diff --git a/tests/test_forward_earn_accounting.py b/tests/test_forward_earn_accounting.py index 54f1d73b..010087c4 100644 --- a/tests/test_forward_earn_accounting.py +++ b/tests/test_forward_earn_accounting.py @@ -128,6 +128,37 @@ def test_deposit_and_interest_are_separate_in_same_window(): assert state['daily_equity_base'] == 1600 +def test_verified_earn_forward_emits_frozen_checkpoint_interval(): + state, new, cash = materials() + report = {"total_equity_usdt": 9999} + 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()} + + maybe_rebase_daily_state_for_balance_change( + state, + runtime, + report, + 1600.000005, + 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['external_cash_flow_interval'] == { + 'account_scope_sha256': 'a' * 64, + 'start_at': '2026-09-12T14:00:00+00:00', + 'end_at': NOW.isoformat(), + 'end_equity_usdt': '1600.000005', + 'net_external_cash_flow': '0', + 'currency': 'USDT', + 'valuation_basis': 'checkpoint_quantities_sampled_prices', + } + + def test_portfolio_prepare_consumes_verified_bnb_dividend_without_principal_or_extra_pnl(): state, new, cash = materials() new['assets']['BNB']['products']['BNB001']['realtime_rewards'] = '0.1' diff --git a/tests/test_runtime_workflow_security.py b/tests/test_runtime_workflow_security.py index 91ce0232..ab1512db 100644 --- a/tests/test_runtime_workflow_security.py +++ b/tests/test_runtime_workflow_security.py @@ -193,6 +193,27 @@ def test_live_authority_materialization_roundtrip_uses_synthetic_secret(tmp_path assert not authority_file.exists() +def test_live_lifecycle_recorder_export_is_scoped_to_normal_strategy_runs() -> None: + workflow = WORKFLOW.read_text(encoding="utf-8") + broker_job = _job_block(workflow, "deploy", "publish-execution-log") + + assert ( + "BINANCE_LIFECYCLE_EXPORT_PATH: ${{ runner.temp }}/binance-lifecycle-${{ github.run_id }}-" + "${{ github.run_attempt }}/lifecycle-run.json" + ) in broker_job + export_step = broker_job.split(" - name: 5a. Stage redacted lifecycle recorder output for monitor", 1)[1].split( + " - name: 5b.", 1 + )[0] + gate = "always() && github.event.inputs.validate_only != 'true' && github.event.inputs.reconcile_only != 'true' && env.RUNTIME_TARGET_ENABLED == 'true'" + assert f"if: ${{{{ {gate} }}}}" in export_step + assert "name: binance-live-run-${{ github.run_id }}-${{ github.run_attempt }}" in export_step + assert "path: ${{ runner.temp }}/binance-lifecycle-${{ github.run_id }}-${{ github.run_attempt }}/lifecycle-run.json" in export_step + assert "if-no-files-found: warn" in export_step + assert "retention-days: 7" in export_step + assert "continue-on-error: true" in export_step + assert "BINANCE_API_KEY" not in _job_block(workflow, "publish-execution-log") + + def test_disabled_host_observation_uses_actual_control_read_and_existing_source() -> None: workflow = WORKFLOW.read_text(encoding="utf-8") broker_job = _job_block(workflow, "deploy", "publish-execution-log")