Bridge verified BNB dividends into Earn accounting

application/broker_reconciliation.py

@@ -39,6 +39,7 @@
 _EXTERNAL_CASH_FLOW_PAGE_SIZE = 1000
 _EXTERNAL_CASH_FLOW_MAX_RECORDS = 256
 _EXTERNAL_CASH_FLOW_CURSOR_VERSION = 1
+_BNB_DIVIDEND_PAGE_SIZE = 500
 
 
 class BinanceReconciliationReadError(RuntimeError):
@@ -288,6 +289,98 @@ def collect_spot_usdt_external_cash_flows(
     }
 
 
+def collect_bnb_dividend_quantity(
+    client: Any, *, start: datetime, end: datetime
+) -> dict[str, object]:
+    """Read one complete BNB dividend window for Earn quantity conservation.
+
+    This is a narrow platform contract: only positive BNB rows with integer
+    ``direction == 1`` are accepted.  The rows stay in memory and the returned
+    identities are only for same-window stability and duplicate detection.
+    """
+    if (
+        not isinstance(start, datetime)
+        or not isinstance(end, datetime)
+        or start.tzinfo is None
+        or end.tzinfo is None
+        or not start < end
+    ):
+        raise ValueError("bnb_dividend_window_invalid")
+    start_ms = int(start.astimezone(timezone.utc).timestamp() * 1000)
+    end_ms = int(end.astimezone(timezone.utc).timestamp() * 1000)
+    try:
+        response = client._request_margin_api(
+            "get",
+            "asset/assetDividend",
+            signed=True,
+            data={"asset": "BNB", "startTime": start_ms, "endTime": end_ms,
+                  "limit": _BNB_DIVIDEND_PAGE_SIZE},
+        )
+    except Exception:
+        raise ValueError("bnb_dividend_read_failed") from None
+    if not isinstance(response, Mapping):
+        raise ValueError("bnb_dividend_response_invalid")
+    rows = response.get("rows")
+    total = response.get("total")
+    total_is_decimal_string = (
+        isinstance(total, str) and total.isascii() and total.isdecimal() and len(total) <= 10
+    )
+    count = int(total) if type(total) is int or total_is_decimal_string else None
+    if (
+        not isinstance(rows, list)
+        or count is None
+        or count < 0
+        or count != len(rows)
+        or count >= _BNB_DIVIDEND_PAGE_SIZE
+    ):
+        raise ValueError("bnb_dividend_history_incomplete")
+    quantity = Decimal(0)
+    identities = []
+    direction_values = set()
+    with localcontext() as context:
+        context.prec = 100
+        for row in rows:
+            if not isinstance(row, Mapping):
+                raise ValueError("bnb_dividend_row_invalid")
+            identity = (row.get("id"), row.get("tranId"), row.get("divTime"))
+            if (
+                type(identity[0]) is not int
+                or identity[0] < 0
+                or type(identity[1]) is not int
+                or identity[1] < 0
+                or type(identity[2]) is not int
+                or not start_ms < identity[2] <= end_ms
+                or row.get("asset") != "BNB"
+                or type(row.get("direction")) is not int
+                or row.get("direction") != 1
+                or identity in identities
+            ):
+                raise ValueError("bnb_dividend_row_invalid")
+            amount = row.get("amount")
+            if not isinstance(amount, str) or not amount or len(amount) > 80:
+                raise ValueError("bnb_dividend_row_invalid")
+            try:
+                decimal_amount = Decimal(amount)
+            except DecimalException:
+                raise ValueError("bnb_dividend_row_invalid") from None
+            if (
+                not decimal_amount.is_finite()
+                or decimal_amount <= 0
+                or abs(decimal_amount) > Decimal("1e30")
+                or decimal_amount.as_tuple().exponent < -30
+            ):
+                raise ValueError("bnb_dividend_row_invalid")
+            quantity += decimal_amount
+            identities.append(identity)
+            direction_values.add(1)
+    return {
+        "quantity": quantity,
+        "record_count": len(identities),
+        "identities": tuple(identities),
+        "direction_values": tuple(sorted(direction_values)),
+    }
+
+
 def _canonical_records(records: Sequence[Mapping[str, object]]) -> tuple[dict[str, object], ...]:
     return tuple(sorted((dict(item) for item in records), key=lambda item: json.dumps(item, sort_keys=True)))
 
@@ -903,9 +996,23 @@ def diagnose_bnb_wallet_activity(
             "total_valid": count is not None and count >= 0,
             "rows_readable": rows_readable,
             "visible_row_count": visible_row_count,
+            "empty_missing_total_exempt": (
+                name == "spot_dust_conversions"
+                and isinstance(response, Mapping)
+                and "total" not in response
+                and rows == []
+            ),
             "window_complete": (
-                count is not None and count >= 0 and shape["total_matches_rows"]
-                and count < limit and valid_rows
+                (
+                    name == "spot_dust_conversions"
+                    and isinstance(response, Mapping)
+                    and "total" not in response
+                    and rows == []
+                )
+                or (
+                    count is not None and count >= 0 and shape["total_matches_rows"]
+                    and count < limit and valid_rows
+                )
             ),
         })
         shape["row_time_and_asset_valid"] = valid_rows

application/earn_accrual.py

@@ -213,13 +213,18 @@ def prepare_forward_earn_state(state, current, cash_flows):
         if cash_flows['new_unsupported_deposit_count'] or cash_flows['new_or_changed_withdrawal_count']:
             raise ValueError('earn_cash_flow_unsupported')
         principal = _amount(cash_flows['new_deposit_principal_usdt'])
+        dividend_quantity = _amount(cash_flows.get('bnb_dividend_quantity', '0'))
         completed = cash_flows['new_deposit_completed_at']
         if (not isinstance(completed, list) or len(completed) != cash_flows['new_confirmed_deposit_count']
                 or (principal == 0) != (not completed)
                 or any(not _time(previous['observed_at']) < _time(t) <= cutoff or _time(t).date() != cutoff.date()
                        for t in completed)):
             raise ValueError('earn_cash_flow_time_unverified')
         verified['USDT'] += principal
+        if dividend_quantity:
+            if 'BNB' not in verified:
+                raise ValueError('earn_cash_flow_invalid')
+            verified['BNB'] += dividend_quantity
         compare_earn_checkpoints(previous, current, verified_net_changes={a: str(v) for a, v in verified.items()})
         updated = copy.deepcopy(state)
         if principal and state.get('last_reset_date') == cutoff.date().isoformat():

application/portfolio_service.py

@@ -92,7 +92,42 @@ def maybe_rebase_daily_state_for_balance_change(
                 raise ValueError('earn_valuation_snapshot_mismatch')
             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)
+            try:
+                updated = prepare_forward_earn_state(state, current, cash)
+            except ValueError as exc:
+                if str(exc) != 'earn_quantity_change_unexplained':
+                    raise
+                previous = state['earn_accrual_checkpoint']
+                previous_bnb = previous.get('assets', {}).get('BNB')
+                current_bnb = current.get('assets', {}).get('BNB')
+                net_bnb = state.get('earn_accounted_net_changes', {}).get('BNB', '0')
+                if not isinstance(previous_bnb, dict) or not isinstance(current_bnb, dict):
+                    raise
+                try:
+                    reward_delta = sum(
+                        Decimal(str(current_bnb['products'][product]['realtime_rewards']))
+                        - Decimal(str(previous_bnb['products'][product]['realtime_rewards']))
+                        for product in previous_bnb['products']
+                    )
+                    unexplained_bnb = (
+                        Decimal(str(current_bnb['quantity']))
+                        - Decimal(str(previous_bnb['quantity']))
+                        - reward_delta
+                        - Decimal(str(net_bnb))
+                    )
+                except (KeyError, TypeError, InvalidOperation):
+                    raise
+                if unexplained_bnb <= 0:
+                    raise
+                from application.broker_reconciliation import collect_bnb_dividend_quantity
+                dividend = collect_bnb_dividend_quantity(
+                    runtime.client,
+                    start=_time(previous['observed_at']),
+                    end=_time(current['observed_at']),
+                )
+                cash = dict(cash)
+                cash['bnb_dividend_quantity'] = format(dividend['quantity'], 'f')
+                updated = prepare_forward_earn_state(state, current, cash)
         except Exception as exc:
             reason_code = (
                 str(exc)

application/rebased_recovery.py

@@ -21,6 +21,7 @@
 )
 
 from application.broker_reconciliation import (
+    collect_bnb_dividend_quantity,
     collect_read_only_reconciliation_observations,
     collect_spot_usdt_external_cash_flows,
     diagnose_balance_flows,
@@ -282,6 +283,12 @@ def _collect_prospective_rebase_private(
         )
     except ValueError:
         raise ValueError("prospective_rebase_cash_flow_unverified") from None
+    try:
+        first_dividends = collect_bnb_dividend_quantity(
+            client, start=opening_at, end=observed_at
+        ) if "BNB" in assets else {"quantity": Decimal(0), "identities": ()}
+    except ValueError:
+        raise ValueError("prospective_rebase_dividend_unverified") from None
 
     final_at = (clock or (lambda: datetime.now(timezone.utc)))().astimezone(timezone.utc)
     if not observed_at < final_at <= observed_at + timedelta(minutes=2):
@@ -302,8 +309,21 @@ def _collect_prospective_rebase_private(
         raise ValueError("prospective_rebase_checkpoint_unavailable") from None
     except ValueError:
         raise ValueError("prospective_rebase_cash_flow_unverified") from None
+    try:
+        second_dividends = collect_bnb_dividend_quantity(
+            client, start=opening_at, end=final_at
+        ) if "BNB" in assets else {"quantity": Decimal(0), "identities": ()}
+    except ValueError:
+        raise ValueError("prospective_rebase_dividend_unverified") from None
     if not _same_cash_flow_slice(first_flows, second_flows):
         raise ValueError("prospective_rebase_cash_flow_changed_during_read")
+    if (
+        first_dividends.get("identities") != second_dividends.get("identities")
+        or first_dividends.get("quantity") != second_dividends.get("quantity")
+    ):
+        raise ValueError("prospective_rebase_dividend_changed_during_read")
+    first_flows = {**first_flows, "bnb_dividend_quantity": format(first_dividends["quantity"], "f")}
+    second_flows = {**second_flows, "bnb_dividend_quantity": format(second_dividends["quantity"], "f")}
 
     try:
         first_state = prepare_forward_earn_state(ledger, first, first_flows)

run_cycle_replay.py

@@ -102,8 +102,11 @@ def _request_margin_api(self, method: str, path: str, *, signed: bool, data: dic
         if method != "get" or not signed or path not in {
             "capital/deposit/hisrec",
             "capital/withdraw/history",
+            "asset/assetDividend",
         }:
             raise RuntimeError("unsupported replay margin read")
+        if path == "asset/assetDividend":
+            return {"total": 0, "rows": []}
         return []
 
     def _record(self, method: str, payload: dict[str, Any]):

runtime_support.py

@@ -639,7 +639,8 @@ def reconcile_runtime_cash_effects(runtime, state):
                     and submission.get("symbol") == getattr(runtime, "fuel_symbol", None))
     if prospective:
         import copy
-        from application.earn_accrual import collect_earn_checkpoint, compare_earn_checkpoints
+        from application.earn_accrual import collect_earn_checkpoint, compare_earn_checkpoints, _amount, _time
+        from application.broker_reconciliation import collect_bnb_dividend_quantity
         try:
             pending_state = copy.deepcopy(state)
             if fuel_pending:
@@ -648,8 +649,22 @@ def reconcile_runtime_cash_effects(runtime, state):
             current = collect_earn_checkpoint(runtime.client, assets=previous["assets"],
                 observed_at=datetime.now(timezone.utc),
                 expected_account_scope_sha256=previous["account_scope_sha256"])
-            compare_earn_checkpoints(previous, current,
-                verified_net_changes=pending_state["earn_accounted_net_changes"])
+            try:
+                compare_earn_checkpoints(previous, current,
+                    verified_net_changes=pending_state["earn_accounted_net_changes"])
+            except ValueError as exc:
+                if str(exc) != "earn_quantity_change_unexplained" or "BNB" not in previous["assets"]:
+                    raise
+                dividend = collect_bnb_dividend_quantity(
+                    runtime.client,
+                    start=_time(previous["observed_at"]),
+                    end=_time(current["observed_at"]),
+                )
+                verified = dict(pending_state["earn_accounted_net_changes"])
+                verified["BNB"] = str(
+                    _amount(verified["BNB"], signed=True) + dividend["quantity"]
+                )
+                compare_earn_checkpoints(previous, current, verified_net_changes=verified)
             observations = {a: float(current["assets"][a]["quantity"]) for a in assets | {"USDT"}}
         except Exception:
             raise ExecutionIntegrityError("cash_reconciliation_uncertain") from None
@@ -711,11 +726,17 @@ def runtime_set_trade_state(runtime, report, state, *, reason):
     require_runtime_state_owner(runtime)
     if runtime.state_writer is None:
         raise StatePersistenceError("state_persistence_failed")
+    import copy
+    before_write = copy.deepcopy(state)
     try:
         persisted = runtime.state_writer(state)
     except Exception:
+        state.clear()
+        state.update(before_write)
         raise StatePersistenceError("state_persistence_failed") from None
     if persisted is not True:
+        state.clear()
+        state.update(before_write)
         raise StatePersistenceError("state_persistence_failed")
     runtime.trade_state = state
     runtime.pending_funds = [item for item in runtime.pending_funds if not _accounted_funds(runtime, state, reason, item)]

scripts/migrate_daily_accounting_state.py

@@ -1172,6 +1172,9 @@ def _summarize_bnb_wallet_activity(report, *, residual, start, end):
         return summary
     summary["dust_bnb_detail_count"] = 0
     summary["dust_non_bnb_target_count"] = 0
+    dust_diagnostics = report.get("surface_diagnostics") if isinstance(report, Mapping) else None
+    dust_shape = dust_diagnostics.get("spot_dust_conversions") if isinstance(dust_diagnostics, Mapping) else None
+    dust_surface_exempt = isinstance(dust_shape, Mapping) and dust_shape.get("empty_missing_total_exempt") is True
 
     start_ms = int(start.timestamp() * 1000)
     end_ms = int(end.timestamp() * 1000)
@@ -1248,12 +1251,13 @@ def _summarize_bnb_wallet_activity(report, *, residual, start, end):
                 return summary
 
         summary["dust_record_count"] = len(private_rows["spot_dust_conversions"])
-        summary["dust_transfer_residual_matches"] = dust_transfer_total == residual
-        summary["dust_after_fee_residual_matches"] = dust_after_fee_total == residual
-        combined_transfer = dividend_total + dust_transfer_total
-        combined_after_fee = dividend_total + dust_after_fee_total
-        summary["combined_transfer_residual_matches"] = combined_transfer == residual
-        summary["combined_after_fee_residual_matches"] = combined_after_fee == residual
+        if not dust_surface_exempt:
+            summary["dust_transfer_residual_matches"] = dust_transfer_total == residual
+            summary["dust_after_fee_residual_matches"] = dust_after_fee_total == residual
+            combined_transfer = dividend_total + dust_transfer_total
+            combined_after_fee = dividend_total + dust_after_fee_total
+            summary["combined_transfer_residual_matches"] = combined_transfer == residual
+            summary["combined_after_fee_residual_matches"] = combined_after_fee == residual
     except (MigrationBlocked, TypeError, ValueError, InvalidOperation):
         return summary
     summary["status"] = "COMPLETE"

scripts/validate_runtime_startup.py

@@ -195,6 +195,7 @@ class _ReadOnlyBroker:
     _MARGIN_READS = frozenset({
         "capital/deposit/hisrec",
         "capital/withdraw/history",
+        "asset/assetDividend",
     })
 
     def __init__(self, client, *, symbols):
@@ -265,6 +266,17 @@ def get_historical_klines(self, symbol, interval, lookback):
     def _request_margin_api(self, method, path, *, signed, data):
         if method != "get" or signed is not True or path not in self._MARGIN_READS:
             raise RuntimeError("full_cycle_broker_request_forbidden")
+        if path == "asset/assetDividend":
+            if (
+                set(data) != {"asset", "startTime", "endTime", "limit"}
+                or data.get("asset") != "BNB"
+                or type(data.get("startTime")) is not int
+                or type(data.get("endTime")) is not int
+                or not data["startTime"] < data["endTime"]
+                or data["endTime"] - data["startTime"] > 7 * 24 * 60 * 60 * 1000
+                or data.get("limit") != 500
+            ):
+                raise RuntimeError("full_cycle_broker_dividend_window_forbidden")
         return self._read("_request_margin_api", method, path, signed=signed, data=dict(data))
 
     def __getattr__(self, _name):

tests/test_balance_flow_diagnostics.py

@@ -365,6 +365,24 @@ def read(*a, **kw):
         assert diagnose_bnb_wallet_activity(c, start=NOW-timedelta(hours=2), end=NOW)['requested_surfaces_complete'] is False
 
 
+def test_bnb_wallet_diagnostic_accepts_only_empty_missing_total_dust_response():
+    from application.broker_reconciliation import diagnose_bnb_wallet_activity
+
+    def read(_method, path, **_kwargs):
+        if path.endswith("assetDividend"):
+            return {"total": 0, "rows": []}
+        return {"userAssetDribblets": []}
+
+    result = diagnose_bnb_wallet_activity(
+        SimpleNamespace(_request_margin_api=read), start=NOW-timedelta(hours=2), end=NOW,
+    )
+
+    assert result["requested_surfaces_complete"] is True
+    shape = result["surface_diagnostics"]["spot_dust_conversions"]
+    assert shape["total_valid"] is False
+    assert shape["window_complete"] is True
+
+
 def test_bnb_wallet_dribblet_keeps_readable_rows_when_total_is_missing():
     from application.broker_reconciliation import diagnose_bnb_wallet_activity