feat: transport Binance live intervals to monitor

.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

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:

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
 

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

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)

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'

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")