diff --git a/application/broker_reconciliation.py b/application/broker_reconciliation.py index 1b10596b..04887f20 100644 --- a/application/broker_reconciliation.py +++ b/application/broker_reconciliation.py @@ -832,8 +832,12 @@ def diagnose_bnb_wallet_activity( if start.tzinfo is None or end.tzinfo is None or not start < end or end-start > timedelta(days=7): raise ValueError("balance_history_window_invalid") bounds = {"startTime": int(start.timestamp()*1000), "endTime": int(end.timestamp()*1000)} - result = {"requested_surfaces_complete": False, "counts": {}, - "complete_balance_reconciliation": False} + result = { + "requested_surfaces_complete": False, + "counts": {}, + "complete_balance_reconciliation": False, + "surface_diagnostics": {}, + } for name, path, params, rows_key, time_key, limit in ( ("bnb_dividends", "asset/assetDividend", {"asset": "BNB", "limit": 500}, "rows", "divTime", 500), ("spot_dust_conversions", "asset/dribblet", {"accountType": "SPOT"}, "userAssetDribblets", "operateTime", 100), @@ -864,17 +868,53 @@ def diagnose_bnb_wallet_activity( "total_matches_rows": isinstance(rows, list) and count == len(rows), "page_full": isinstance(rows, list) and len(rows) >= limit, } - valid_rows = isinstance(rows, list) and all( - isinstance(row, Mapping) and type(row.get(time_key)) is int - and bounds["startTime"] <= row[time_key] <= bounds["endTime"] - and (name != "bnb_dividends" or row.get("asset") == "BNB") for row in rows - ) + rows_readable = isinstance(rows, list) and all(isinstance(row, Mapping) for row in rows) + visible_row_count = len(rows) if isinstance(rows, list) else 0 + row_keys = set() + valid_rows = rows_readable + if valid_rows: + for row in rows: + timestamp = row.get(time_key) + if ( + type(timestamp) is not int + or not bounds["startTime"] <= timestamp <= bounds["endTime"] + or (name == "bnb_dividends" and row.get("asset") != "BNB") + ): + valid_rows = False + break + if name == "bnb_dividends": + identity = (row.get("id"), row.get("tranId"), timestamp) + if ( + type(identity[0]) is not int + or identity[0] < 0 + or type(identity[1]) is not int + or identity[1] < 0 + or identity in row_keys + ): + valid_rows = False + break + else: + identity = row.get("transId") + if type(identity) is not int or identity < 0 or identity in row_keys: + valid_rows = False + break + row_keys.add(identity) + shape.update({ + "total_valid": count is not None and count >= 0, + "rows_readable": rows_readable, + "visible_row_count": visible_row_count, + "window_complete": ( + 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 - if (count is None or count < 0 or not shape["total_matches_rows"] - or count >= limit or not valid_rows): + result["surface_diagnostics"][name] = shape + if isinstance(rows, list) and rows_readable: + result["counts"][name] = visible_row_count + if include_rows: + result.setdefault("_private_rows", {})[name] = rows + if not shape["window_complete"]: return {**result, "reason_code": "bnb_wallet_history_unverified", "failed_surface": name, "failure_stage": "response_validation", "response_shape": shape} - result["counts"][name] = len(rows) - if include_rows: - result.setdefault("_private_rows", {})[name] = rows return {**result, "requested_surfaces_complete": True} diff --git a/application/cycle_service.py b/application/cycle_service.py index 5dd4f332..8ce51b23 100644 --- a/application/cycle_service.py +++ b/application/cycle_service.py @@ -29,6 +29,16 @@ def _record_risk_diagnostics(report, allocation): report["risk_flags"] = list(allocation.get("risk_flags") or ()) +def _settled_order_state(state): + if not isinstance(state, Mapping): + return None + record = state.get("order_submission") + if not isinstance(record, Mapping): + return None + status = record.get("state") + return status if status in {"RESERVED", "TERMINAL"} else None + + def execute_strategy_cycle( runtime, *, @@ -77,10 +87,33 @@ def execute_strategy_cycle( state_healthy = False failure_stage = "state_owner_claim" + owner_claimed_this_cycle = False + initial_order_state = None + daily_state_write_intents_start = None + daily_funding_submission_start = None + daily_funding_side_effect_start = None + daily_funding_intent_lengths = None + daily_order_sequence_start = None + cycle_funding_submission_start = None + cycle_funding_side_effect_start = None + cycle_order_sequence_start = None try: if not acquire_runtime_state_owner(runtime): report["execution_blocked_reason"] = "state_owner_busy" return report + owner_claimed_this_cycle = ( + not getattr(runtime, "dry_run", False) + and getattr(runtime, "standard_execution_permitted", True) + and getattr(runtime, "state_owner_held", False) + ) + receipt_observation = report.get("execution_receipt_observation", {}) + cycle_funding_submission_start = ( + receipt_observation.get("submission_attempted_count", 0) + if isinstance(receipt_observation, Mapping) + else 0 + ) + cycle_funding_side_effect_start = len(getattr(runtime, "side_effect_log", ())) + cycle_order_sequence_start = getattr(runtime, "order_sequence", 0) failure_stage = "client_connect" if not ensure_runtime_client(runtime, report): return report @@ -92,6 +125,7 @@ def execute_strategy_cycle( state, trend_pool_resolution, runtime_trend_universe, allow_new_trend_entries = cycle_state runtime.trade_state = state + initial_order_state = _settled_order_state(state) failure_stage = "funding_reconciliation" reconcile_pending_funding_submission(runtime) submission_state = state.get("order_submission", {}).get("state", "RESERVED") @@ -148,6 +182,19 @@ def execute_strategy_cycle( return report failure_stage = "daily_state" + daily_state_write_intents_start = len(report.get("state_write_intents", ())) + receipt_observation = report.get("execution_receipt_observation", {}) + daily_funding_submission_start = ( + receipt_observation.get("submission_attempted_count", 0) + if isinstance(receipt_observation, Mapping) + else 0 + ) + daily_funding_side_effect_start = len(getattr(runtime, "side_effect_log", ())) + daily_order_sequence_start = getattr(runtime, "order_sequence", 0) + daily_funding_intent_lengths = tuple( + len(report.get(key, ())) if isinstance(report.get(key, ()), list) else None + for key in ("buy_sell_intents", "btc_dca_intents", "redemption_subscription_intents") + ) now_utc = runtime.now_utc today_utc = now_utc.strftime("%Y-%m-%d") today_id_str = now_utc.strftime("%Y%m%d") @@ -349,7 +396,78 @@ def execute_strategy_cycle( except Exception: pass finally: - if state_healthy and getattr(runtime, "state_owner_held", False): + reason_code = None + earn_diagnostics = report.get("diagnostics", {}).get("earn_accrual", {}) + if isinstance(earn_diagnostics, Mapping): + reason_code = earn_diagnostics.get("reason_code") + current_order_state = _settled_order_state(getattr(runtime, "trade_state", None) or locals().get("state")) + current_receipt_observation = report.get("execution_receipt_observation", {}) + funding_submission_unchanged = ( + isinstance(current_receipt_observation, Mapping) + and current_receipt_observation.get("submission_attempted_count", 0) + == daily_funding_submission_start + ) + funding_submission_unchanged_full_cycle = ( + isinstance(current_receipt_observation, Mapping) + and current_receipt_observation.get("submission_attempted_count", 0) + == cycle_funding_submission_start + ) + current_intent_lengths = tuple( + len(report.get(key, ())) if isinstance(report.get(key, ()), list) else None + for key in ("buy_sell_intents", "btc_dca_intents", "redemption_subscription_intents") + ) + funding_intents_unchanged = current_intent_lengths == daily_funding_intent_lengths + funding_order_sequence_unchanged = getattr(runtime, "order_sequence", 0) == daily_order_sequence_start + funding_order_sequence_unchanged_full_cycle = ( + getattr(runtime, "order_sequence", 0) == cycle_order_sequence_start + ) + full_cycle_side_effects = list(getattr(runtime, "side_effect_log", ()))[cycle_funding_side_effect_start or 0:] + full_cycle_funding_side_effects_absent = all( + not str(entry.get("effect_type", "")).startswith(("order_", "earn_")) + for entry in full_cycle_side_effects + if isinstance(entry, Mapping) + ) + new_side_effects = list(getattr(runtime, "side_effect_log", ()))[daily_funding_side_effect_start or 0:] + funding_side_effects_absent = all( + not str(entry.get("effect_type", "")).startswith(("order_", "earn_")) + for entry in new_side_effects + if isinstance(entry, Mapping) + ) + persistent_side_effects_absent = all( + not ( + str(entry.get("target", "")) == "firestore" + or str(entry.get("effect_type", "")).startswith("state_") + ) + for entry in new_side_effects + if isinstance(entry, Mapping) + ) + pure_daily_state_failure = ( + not state_healthy + and failure_stage == "daily_state" + and reason_code in EARN_FORWARD_REASON_CODES + and owner_claimed_this_cycle + and initial_order_state in {"RESERVED", "TERMINAL"} + and current_order_state in {"RESERVED", "TERMINAL"} + and not getattr(runtime, "pending_funds", ()) + and funding_submission_unchanged + and funding_submission_unchanged_full_cycle + and funding_intents_unchanged + and funding_order_sequence_unchanged + and funding_order_sequence_unchanged_full_cycle + and funding_side_effects_absent + and full_cycle_funding_side_effects_absent + and persistent_side_effects_absent + and isinstance(report.get("state_write_intents"), list) + and daily_state_write_intents_start is not None + and len(report["state_write_intents"]) == daily_state_write_intents_start + ) + if pure_daily_state_failure: + try: + release_runtime_state_owner(runtime) + except ExecutionIntegrityError: + 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: diff --git a/docs/binance_reconciliation_recovery.zh-CN.md b/docs/binance_reconciliation_recovery.zh-CN.md index 0931b59a..8c9a4bb9 100644 --- a/docs/binance_reconciliation_recovery.zh-CN.md +++ b/docs/binance_reconciliation_recovery.zh-CN.md @@ -128,8 +128,11 @@ Spot + Flexible Earn 持有资产口径，闲置理财不等于资金退出策 ## 向前收益记账消费者（2026-09-12） 仅账本已含 `earn_accrual_checkpoint` 时启用新路径；旧账本继续使用原路径，不自动迁移。 -加载时保留 checkpoint 与 `earn_accounted_net_changes`。市场估值与记账使用同一次完整 Spot/Earn -数量采样；既有 owner-protected writer 一次保存下一检查点、流水游标、余额及清零后的已记成交净变动。 +加载时保留 checkpoint 与 `earn_accounted_net_changes`。诊断和恢复核对按 +`Spot1 → Earn1 → Spot2 → Earn2` 顺序做有限双采样，比较两次 Spot/Earn 的数量、产品和实时收益计数； +`sampling_stable` 还要求 Spot/Earn 分量变化可由实时收益计数解释，不再仅由 Firestore owner/账本/control +标记稳定推出。市场估值与记账使用一次完整 +Spot/Earn 数量采样；既有 owner-protected writer 一次保存下一检查点、流水游标、余额及清零后的已记成交净变动。 验证失败不推进；写入失败不更新内存检查点，也不自动重试。 数量变化按同产品实时计数增量、已持久化的完整成交净数量/手续费、已核实当日 USDT 入金核对。 @@ -137,6 +140,15 @@ Spot + Flexible Earn 持有资产口径，闲置理财不等于资金退出策 不通过余额差反推资金来源。中途资金核验可更新当前余额，但不单独推进收益检查点或流水游标，避免漏掉区间资金活动。 理财奖励已体现在权益中，不再加一次利润，也不记作外部本金。 +`earn-forward-diagnose` 只在内存保留原始采样行，公开结果只报告资产方向、收益/资金来源残差方向、 +记录条数和匹配标志。BNB wallet 的 `dribblet` 诊断分别报告行是否可读、可见行数和窗口是否完整； +缺失或非法 `total`、满页、重复、越界或非 BNB dividend 行都会保持未核实，不会把已见有效行静默丢弃， +也不会将不完整窗口写成完整资金来源。 + +若既有运行仅在 `daily_state` 的纯 Earn 准备校验失败，且本轮 owner 已成功取得、加载时订单明确为 +`RESERVED/TERMINAL` 并持续 settled、无 pending funds、准备阶段没有新增持久写入意图，则只释放本轮自己的 owner。 +未知订单、写入结果不明或旧 owner 继续保留，旧 owner 不由诊断或运行补丁自动清除。 + 未知/未完成订单、产品消失或更换、计数回退、新提币及未支持币种入金继续拒绝。内部申赎不算收益； 只有同产品的总数量守恒可直接通过，产品生命周期不连续时不能猜测新的累计计数起点。 这不是所有资金活动均可自动分类的声明。 diff --git a/docs/operator_runbook.md b/docs/operator_runbook.md index b24bfc0e..1a94b5c1 100644 --- a/docs/operator_runbook.md +++ b/docs/operator_runbook.md @@ -274,23 +274,37 @@ 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 +current `earn_accrual_checkpoint`, reads Spot then Flexible Earn twice in the +bounded order `Spot1 → Earn1 → Spot2 → Earn2`, 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. +product and reward counts, matching flags, residual breakdown directions, owner +existence, and fixed no-write policy flags. `sampling_stable` is true only when +the two in-memory Spot/Earn shapes and Spot/Earn component movements match, +with quantity changes explained by realtime counters; stable Firestore markers +alone are insufficient. 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. When BNB has a positive residual while the other sampled checks are stable, the same diagnostic may read the bounded `assetDividend` and Spot `dribblet` wallet surfaces. It reports record counts and exact Decimal comparisons for dividend, transfer, fee-adjusted, and combined values without exposing amounts or source -rows. Binance's transfer and fee semantics remain unverified, and the separate -one-eight-decimal-unit flag is diagnostic only; neither flag changes accounting, -owner, control, or execution state. +rows. Row readability, visible row count, and window completeness are separate +fields. A missing or invalid `total`, a full page, a duplicate, an out-of-window +row, or a non-BNB dividend row keeps the window unverified even when valid rows +are visible. Binance's transfer and fee semantics remain unverified, and the +separate one-eight-decimal-unit flag is diagnostic only; neither flag changes +accounting, owner, control, or execution state. + +If a cycle reaches `daily_state` and fails only during the pure Earn preparation +check, the runtime releases its own newly claimed owner only when the initial +and current order states are explicitly `RESERVED` or `TERMINAL`, no funds are +pending, and no state-write intent was added after the state-load metadata +refresh. Unknown orders, uncertain writes, or an owner held before this cycle +remain locked for manual reconciliation. The migration is a separate, one-time `Runtime` workflow mode for an old `trend_val` ledger. It does not activate recovery control, grant execution diff --git a/scripts/migrate_daily_accounting_state.py b/scripts/migrate_daily_accounting_state.py index 16b93945..0bded51e 100644 --- a/scripts/migrate_daily_accounting_state.py +++ b/scripts/migrate_daily_accounting_state.py @@ -20,6 +20,7 @@ import sys import subprocess import tempfile +import time from argparse import ArgumentParser from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation @@ -1055,15 +1056,15 @@ def _summarize_bnb_wallet_activity(report, *, residual, start, end): "source_failed_surface": None, "source_failure_stage": None, "source_response_shape": None, - "dividend_count": 0, - "dust_record_count": 0, - "dust_bnb_detail_count": 0, - "dust_non_bnb_target_count": 0, - "dividend_residual_matches": False, - "dust_transfer_residual_matches": False, - "dust_after_fee_residual_matches": False, - "combined_transfer_residual_matches": False, - "combined_after_fee_residual_matches": False, + "dividend_count": None, + "dust_record_count": None, + "dust_bnb_detail_count": None, + "dust_non_bnb_target_count": None, + "dividend_residual_matches": None, + "dust_transfer_residual_matches": None, + "dust_after_fee_residual_matches": None, + "combined_transfer_residual_matches": None, + "combined_after_fee_residual_matches": None, "residual_within_one_eight_decimal_unit": abs(residual) <= _BNB_DIAGNOSTIC_QUANTUM, "dividend_net_semantics_verified": False, "dust_net_semantics_verified": False, @@ -1071,6 +1072,12 @@ def _summarize_bnb_wallet_activity(report, *, residual, start, end): } if not isinstance(report, Mapping) or report.get("requested_surfaces_complete") is not True: if isinstance(report, Mapping): + counts = report.get("counts") + if isinstance(counts, Mapping): + if type(counts.get("bnb_dividends")) is int: + summary["dividend_count"] = counts["bnb_dividends"] + if type(counts.get("spot_dust_conversions")) is int: + summary["dust_record_count"] = counts["spot_dust_conversions"] summary["source_reason_code"] = report.get("reason_code") if report.get("reason_code") in { "bnb_wallet_history_unverified", } else None @@ -1087,9 +1094,10 @@ def _summarize_bnb_wallet_activity(report, *, residual, start, end): for key in ( "rows_present", "rows_is_list", "total_is_integer", "total_is_decimal_string", "total_is_zero", "total_matches_rows", - "page_full", "row_time_and_asset_valid", + "page_full", "row_time_and_asset_valid", "total_valid", + "rows_readable", "visible_row_count", "window_complete", ) - if type(shape.get(key)) is bool + if type(shape.get(key)) is bool or type(shape.get(key)) is int } return summary summary["status"] = "UNVERIFIED" @@ -1099,6 +1107,8 @@ def _summarize_bnb_wallet_activity(report, *, residual, start, end): or not isinstance(private_rows.get("spot_dust_conversions"), list) ): return summary + summary["dust_bnb_detail_count"] = 0 + summary["dust_non_bnb_target_count"] = 0 start_ms = int(start.timestamp() * 1000) end_ms = int(end.timestamp() * 1000) @@ -1277,9 +1287,39 @@ def diagnose_earn_forward(refs, *, client, expected, now): if any(symbol not in configured_symbols for symbol in required_symbols): raise MigrationBlocked("earn_diagnosis_symbols_missing") + sampling_requests = [] + + class _SamplingClient: + def __init__(self, wrapped, sample_label): + self._wrapped = wrapped + self._sample_label = sample_label + + def _read(self, surface, method, **kwargs): + started = time.monotonic() + started_at = datetime.now(timezone.utc) + try: + return method(**kwargs) + finally: + sampling_requests.append({ + "label": f"{surface}_{self._sample_label}", + "started": started, + "finished": time.monotonic(), + "started_at": started_at, + }) + + def get_account(self): + return self._read("spot", self._wrapped.get_account) + + def get_simple_earn_flexible_product_position(self, **kwargs): + return self._read( + "earn", + self._wrapped.get_simple_earn_flexible_product_position, + **kwargs, + ) + try: current = collect_earn_checkpoint( - client, + _SamplingClient(client, "1"), assets=assets, observed_at=now, expected_account_scope_sha256=expected_scope, @@ -1287,6 +1327,101 @@ def diagnose_earn_forward(refs, *, client, expected, now): except Exception: raise MigrationBlocked("earn_diagnosis_checkpoint_unavailable") from None + # Each checkpoint performs Spot first and Flexible Earn second. Keep both + # samples private and compare the complete normalized shape; Firestore + # markers below cannot prove that broker reads came from one stable window. + try: + second_sample = collect_earn_checkpoint( + _SamplingClient(client, "2"), + assets=assets, + observed_at=now, + expected_account_scope_sha256=expected_scope, + ) + except Exception: + second_sample = None + + def _sampling_shape(value): + if not isinstance(value, Mapping): + return None + assets_value = value.get("assets") + if not isinstance(assets_value, Mapping): + return None + shape = {} + for asset, row in assets_value.items(): + if not isinstance(row, Mapping) or not isinstance(row.get("products"), Mapping): + return None + products = {} + for product, position in row["products"].items(): + if not isinstance(position, Mapping): + return None + products[product] = ( + position.get("auto_subscribe"), + position.get("can_redeem"), + ) + shape[asset] = ( + products, + ) + return shape + + sampling_shape_stable = ( + second_sample is not None + and _sampling_shape(current) is not None + and _sampling_shape(current) == _sampling_shape(second_sample) + ) + + def _sampling_components_stable(first, second): + if not isinstance(first, Mapping) or not isinstance(second, Mapping): + return False + first_assets, second_assets = first.get("assets"), second.get("assets") + if not isinstance(first_assets, Mapping) or not isinstance(second_assets, Mapping): + return False + if set(first_assets) != set(second_assets): + return False + try: + for asset in first_assets: + before, after = first_assets[asset], second_assets[asset] + if not isinstance(before, Mapping) or not isinstance(after, Mapping): + return False + if ( + _diagnosis_decimal(before.get("spot_free"), signed=True) + != _diagnosis_decimal(after.get("spot_free"), signed=True) + or _diagnosis_decimal(before.get("spot_locked"), signed=True) + != _diagnosis_decimal(after.get("spot_locked"), signed=True) + ): + return False + first_products, second_products = before.get("products"), after.get("products") + if not isinstance(first_products, Mapping) or not isinstance(second_products, Mapping): + return False + if set(first_products) != set(second_products): + return False + for product in first_products: + old_row, new_row = first_products[product], second_products[product] + if not isinstance(old_row, Mapping) or not isinstance(new_row, Mapping): + return False + quantity_delta = _diagnosis_decimal(new_row.get("total"), signed=True) - _diagnosis_decimal( + old_row.get("total"), signed=True + ) + reward_delta = _diagnosis_decimal( + new_row.get("realtime_rewards"), signed=True + ) - _diagnosis_decimal(old_row.get("realtime_rewards"), signed=True) + if quantity_delta != reward_delta: + return False + except (MigrationBlocked, TypeError, ValueError, InvalidOperation): + return False + return True + + sampling_components_stable = _sampling_components_stable(current, second_sample) + expected_sampling_labels = ["spot_1", "earn_1", "spot_2", "earn_2"] + observed_sampling_labels = [row["label"] for row in sampling_requests] + sampling_timing_stable = ( + observed_sampling_labels == expected_sampling_labels + and all(row["finished"] >= row["started"] for row in sampling_requests) + and all( + sampling_requests[index]["finished"] <= sampling_requests[index + 1]["started"] + for index in range(len(sampling_requests) - 1) + ) + ) + try: flows = collect_spot_usdt_external_cash_flows( client, now=now, cursor=copy.deepcopy(cursor) @@ -1318,6 +1453,45 @@ def diagnose_earn_forward(refs, *, client, expected, now): else: external_status = "OBSERVED" + def _sample_residual_after_realtime(sample): + if not isinstance(sample, Mapping) or not isinstance(sample.get("assets"), Mapping): + return None + residuals = {} + for asset in assets: + before = checkpoint["assets"].get(asset) + after = sample["assets"].get(asset) + if not isinstance(before, Mapping) or not isinstance(after, Mapping): + return None + before_products, after_products = before.get("products"), after.get("products") + if not isinstance(before_products, Mapping) or not isinstance(after_products, Mapping): + return None + if set(before_products) != set(after_products): + return None + reward = Decimal(0) + for product in before_products: + old_row, new_row = before["products"][product], after["products"][product] + if ( + not isinstance(old_row, Mapping) + or not isinstance(new_row, Mapping) + or old_row.get("auto_subscribe") != new_row.get("auto_subscribe") + ): + return None + delta = _diagnosis_decimal(new_row.get("realtime_rewards"), signed=True) - _diagnosis_decimal( + old_row.get("realtime_rewards"), signed=True + ) + if delta < 0: + return None + reward += delta + observed = _diagnosis_decimal(after.get("quantity"), signed=True) - _diagnosis_decimal( + before.get("quantity"), signed=True + ) + residuals[asset] = observed - stored_net[asset] - ( + external_principal if asset == "USDT" else Decimal(0) + ) - reward + return residuals + + second_residual_after_realtime = _sample_residual_after_realtime(second_sample) + observed_delta = {} residual_before_realtime = {} residual_after_realtime = {} @@ -1382,15 +1556,15 @@ def diagnose_earn_forward(refs, *, client, expected, now): "source_failed_surface": None, "source_failure_stage": None, "source_response_shape": None, - "dividend_count": 0, - "dust_record_count": 0, - "dust_bnb_detail_count": 0, - "dust_non_bnb_target_count": 0, - "dividend_residual_matches": False, - "dust_transfer_residual_matches": False, - "dust_after_fee_residual_matches": False, - "combined_transfer_residual_matches": False, - "combined_after_fee_residual_matches": False, + "dividend_count": None, + "dust_record_count": None, + "dust_bnb_detail_count": None, + "dust_non_bnb_target_count": None, + "dividend_residual_matches": None, + "dust_transfer_residual_matches": None, + "dust_after_fee_residual_matches": None, + "combined_transfer_residual_matches": None, + "combined_after_fee_residual_matches": None, "residual_within_one_eight_decimal_unit": False, "dividend_net_semantics_verified": False, "dust_net_semantics_verified": False, @@ -1474,7 +1648,25 @@ def diagnose_earn_forward(refs, *, client, expected, now): ) asset_results[asset] = { "quantity_direction": _direction(observed_delta[asset]), + "residual_before_realtime_direction": _direction(residual_before_realtime[asset]), "residual_direction": _direction(residual_after_realtime[asset]), + "stored_net_change_direction": _direction(stored_net[asset]), + "trade_net_difference_direction": _direction(trade_net[asset] - stored_net[asset]), + "external_flow_direction": _direction( + external_principal if asset == "USDT" else Decimal(0) + ), + "realtime_reward_direction": _direction( + realtime_delta[asset] if realtime_delta[asset] is not None else Decimal(0) + ), + "second_sample_residual_direction": ( + _direction(second_residual_after_realtime[asset]) + if second_residual_after_realtime is not None + else None + ), + "second_sample_residual_matches_first": ( + second_residual_after_realtime is not None + and second_residual_after_realtime[asset] == 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"]), @@ -1521,6 +1713,20 @@ def diagnose_earn_forward(refs, *, client, expected, now): "wallet_source_response_shape": bnb_wallet_activity["source_response_shape"], }) + sampling_residual_stable = ( + second_residual_after_realtime is not None + and all( + second_residual_after_realtime[asset] == residual_after_realtime[asset] + for asset in assets + ) + ) + sampling_stable = ( + sampling_shape_stable + and sampling_components_stable + and sampling_timing_stable + and sampling_residual_stable + ) + after = _read_earn_diagnosis_source(refs) if any( source[key] != after[key] @@ -1534,7 +1740,13 @@ def diagnose_earn_forward(refs, *, client, expected, now): "owner_unchanged": True, "ledger_unchanged": True, "control_unchanged": True, - "sampling_stable": True, + "sampling_stable": sampling_stable, + "sampling_sequence": ["spot_1", "earn_1", "spot_2", "earn_2"], + "sampling_second_read_available": second_sample is not None, + "sampling_request_timing_stable": sampling_timing_stable, + "sampling_components_stable": sampling_components_stable, + "sampling_residual_stable": sampling_residual_stable, + "sampling_observed_at": current.get("observed_at"), "account_scope_verified": True, "checkpoint_window_within_seven_days": True, "order_state_known": order_state_known, diff --git a/tests/test_balance_flow_diagnostics.py b/tests/test_balance_flow_diagnostics.py index a2dffbe3..423dae9d 100644 --- a/tests/test_balance_flow_diagnostics.py +++ b/tests/test_balance_flow_diagnostics.py @@ -363,3 +363,95 @@ def read(*a, **kw): for value in (None, True, '-1', 'unknown', '0.0'): c = SimpleNamespace(_request_margin_api=lambda *a, **kw: {'total': value, 'rows': [], 'userAssetDribblets': []}) assert diagnose_bnb_wallet_activity(c, start=NOW-timedelta(hours=2), end=NOW)['requested_surfaces_complete'] is False + + +def test_bnb_wallet_dribblet_keeps_readable_rows_when_total_is_missing(): + from application.broker_reconciliation import diagnose_bnb_wallet_activity + + stamp = int((NOW - timedelta(hours=1)).timestamp() * 1000) + + def read(_method, path, **_kwargs): + if path.endswith("assetDividend"): + return {"rows": [], "total": 0} + return {"userAssetDribblets": [{"operateTime": stamp}]} + + result = diagnose_bnb_wallet_activity( + SimpleNamespace(_request_margin_api=read), + start=NOW-timedelta(hours=2), end=NOW, include_rows=True, + ) + + assert result["requested_surfaces_complete"] is False + assert result["counts"]["spot_dust_conversions"] == 1 + shape = result["surface_diagnostics"]["spot_dust_conversions"] + assert shape["rows_readable"] is True + assert shape["visible_row_count"] == 1 + assert shape["window_complete"] is False + assert result["_private_rows"]["spot_dust_conversions"][0]["operateTime"] == stamp + + +@pytest.mark.parametrize("response", [ + {"total": 100, "userAssetDribblets": []}, + {"total": "unknown", "userAssetDribblets": [{"operateTime": int((NOW - timedelta(hours=1)).timestamp() * 1000)}]}, +]) +def test_bnb_wallet_dribblet_full_or_invalid_total_never_claims_complete(response): + from application.broker_reconciliation import diagnose_bnb_wallet_activity + + def read(_method, path, **_kwargs): + if path.endswith("assetDividend"): + return {"rows": [], "total": 0} + return response + + result = diagnose_bnb_wallet_activity( + SimpleNamespace(_request_margin_api=read), start=NOW-timedelta(hours=2), end=NOW, + ) + + assert result["requested_surfaces_complete"] is False + assert result["surface_diagnostics"]["spot_dust_conversions"]["window_complete"] is False + + +def test_bnb_wallet_duplicate_or_non_bnb_rows_never_claim_complete(): + from application.broker_reconciliation import diagnose_bnb_wallet_activity + + stamp = int((NOW - timedelta(hours=1)).timestamp() * 1000) + + def read(_method, path, **_kwargs): + if path.endswith("assetDividend"): + return { + "rows": [ + {"id": 1, "tranId": 2, "asset": "BTC", "divTime": stamp}, + ], + "total": 1, + } + return { + "userAssetDribblets": [ + {"transId": 3, "operateTime": stamp}, + {"transId": 3, "operateTime": stamp}, + ], + "total": 2, + } + + result = diagnose_bnb_wallet_activity( + SimpleNamespace(_request_margin_api=read), start=NOW-timedelta(hours=2), end=NOW, + ) + + assert result["requested_surfaces_complete"] is False + assert result["failed_surface"] == "bnb_dividends" + assert result["surface_diagnostics"]["bnb_dividends"]["window_complete"] is False + + def read_duplicate(_method, path, **_kwargs): + if path.endswith("assetDividend"): + return {"rows": [{"id": 1, "tranId": 2, "asset": "BNB", "divTime": stamp}], "total": 1} + return { + "userAssetDribblets": [ + {"transId": 3, "operateTime": stamp}, + {"transId": 3, "operateTime": stamp}, + ], + "total": 2, + } + + duplicate = diagnose_bnb_wallet_activity( + SimpleNamespace(_request_margin_api=read_duplicate), + start=NOW-timedelta(hours=2), end=NOW, + ) + assert duplicate["requested_surfaces_complete"] is False + assert duplicate["failed_surface"] == "spot_dust_conversions" diff --git a/tests/test_cycle_service.py b/tests/test_cycle_service.py index 125ed692..ee1fcead 100644 --- a/tests/test_cycle_service.py +++ b/tests/test_cycle_service.py @@ -82,6 +82,7 @@ def note(event, value): "selected_symbols": {}, "gating_summary": {}, "gating_events": [], + "state_write_intents": [], }, ensure_runtime_client=lambda *_args: note("client", True), load_cycle_execution_settings=lambda: SimpleNamespace( @@ -378,6 +379,86 @@ def reconcile(*args): }, ) + def test_pure_daily_state_prepare_failure_releases_this_cycle_owner(self): + release = Mock(return_value=True) + + def reconcile(*args): + args[2]["diagnostics"] = { + "earn_accrual": { + "status": "blocked", + "reason_code": "earn_quantity_change_unexplained", + } + } + raise ExecutionIntegrityError("earn_forward_accounting_unverified") + + runtime = ExecutionRuntime( + state_owner_claim=lambda _owner: True, + state_owner_release=release, + ) + report, _events = self._run_funds_cycle( + True, + state={"order_submission": {"state": "TERMINAL"}}, + runtime=runtime, + rebase_fn=reconcile, + ) + + self.assertEqual(report["status"], "error") + release.assert_called_once() + self.assertFalse(runtime.state_owner_held) + + def test_daily_state_prepare_failure_keeps_unknown_order_owner(self): + release = Mock(return_value=True) + + def reconcile(*args): + args[2]["diagnostics"] = { + "earn_accrual": { + "status": "blocked", + "reason_code": "earn_quantity_change_unexplained", + } + } + raise ExecutionIntegrityError("earn_forward_accounting_unverified") + + runtime = ExecutionRuntime( + state_owner_claim=lambda _owner: True, + state_owner_release=release, + ) + self._run_funds_cycle( + True, + state={"order_submission": {"state": "SUBMISSION_UNKNOWN"}}, + runtime=runtime, + rebase_fn=reconcile, + ) + + release.assert_not_called() + self.assertTrue(runtime.state_owner_held) + + def test_daily_state_write_after_prepare_baseline_keeps_owner(self): + release = Mock(return_value=True) + + def reconcile(*args): + args[2]["diagnostics"] = { + "earn_accrual": { + "status": "blocked", + "reason_code": "earn_quantity_change_unexplained", + } + } + args[2]["state_write_intents"].append({"reason": "unexpected"}) + raise ExecutionIntegrityError("earn_forward_accounting_unverified") + + runtime = ExecutionRuntime( + state_owner_claim=lambda _owner: True, + state_owner_release=release, + ) + self._run_funds_cycle( + True, + state={"order_submission": {"state": "TERMINAL"}}, + runtime=runtime, + rebase_fn=reconcile, + ) + + release.assert_not_called() + self.assertTrue(runtime.state_owner_held) + 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_earn_forward_diagnosis.py b/tests/test_earn_forward_diagnosis.py index 36bf27b4..c01dc27c 100644 --- a/tests/test_earn_forward_diagnosis.py +++ b/tests/test_earn_forward_diagnosis.py @@ -240,6 +240,71 @@ def _request_margin_api(self, _method, path, *, signed, data): assert btc["classification"] == "residual_zero_after_realtime_counter" +@pytest.mark.parametrize( + "second_amount,second_reward,expected_stable", + [("1.1", "0.1", True), ("1.2", "0.1", False)], +) +def test_sampling_distinguishes_normal_realtime_growth_from_mixed_snapshot( + monkeypatch, second_amount, second_reward, expected_stable +): + from application import earn_accrual + from scripts import migrate_daily_accounting_state as migration + + real_collect = earn_accrual.collect_earn_checkpoint + refs = source() + install_read_stubs(monkeypatch, migration, refs) + monkeypatch.setattr(earn_accrual, "collect_earn_checkpoint", real_collect) + monkeypatch.setattr(earn_accrual, "digest", lambda _value: SCOPE) + + class Client: + def __init__(self): + self.earn_reads = 0 + + def get_account(self): + return { + "uid": "123", + "balances": [ + {"asset": "USDT", "free": "10", "locked": "0"}, + {"asset": "BTC", "free": "0", "locked": "0"}, + ], + } + + def get_simple_earn_flexible_product_position(self, **_kwargs): + self.earn_reads += 1 + amount, reward = ( + ("1", "0") + if self.earn_reads == 1 + else (second_amount, second_reward) + ) + return { + "total": 1, + "rows": [{ + "asset": "BTC", + "productId": "BTC001", + "totalAmount": amount, + "cumulativeRealTimeRewards": reward, + "collateralAmount": "0", + "autoSubscribe": False, + "canRedeem": True, + }], + } + + result = migration.diagnose_earn_forward( + refs, + client=Client(), + expected={"account_scope_sha256": SCOPE}, + now=NOW, + ) + + assert result["sampling_sequence"] == ["spot_1", "earn_1", "spot_2", "earn_2"] + assert result["sampling_second_read_available"] is True + assert result["sampling_request_timing_stable"] is True + assert result["sampling_components_stable"] is expected_stable + assert result["sampling_residual_stable"] is expected_stable + assert result["sampling_stable"] is expected_stable + assert result["assets"]["BTC"]["second_sample_residual_matches_first"] is expected_stable + + def test_bnb_wallet_summary_matches_dividend_plus_transfer_without_claiming_fee_semantics(): from decimal import Decimal from scripts import migrate_daily_accounting_state as migration @@ -347,6 +412,34 @@ def test_bnb_wallet_summary_preserves_sanitized_source_failure_metadata(): assert "private" not in json.dumps(result) +def test_bnb_wallet_summary_preserves_visible_rows_without_claiming_complete(): + from scripts import migrate_daily_accounting_state as migration + + result = migration._summarize_bnb_wallet_activity( + { + "requested_surfaces_complete": False, + "reason_code": "bnb_wallet_history_unverified", + "failed_surface": "spot_dust_conversions", + "failure_stage": "response_validation", + "counts": {"bnb_dividends": 2, "spot_dust_conversions": 1}, + "response_shape": { + "rows_readable": True, + "visible_row_count": 1, + "window_complete": False, + }, + }, + residual=0, + start=NOW-timedelta(hours=2), + end=NOW, + ) + + assert result["status"] == "CHECK_FAILED" + assert result["complete"] is False + assert result["dividend_count"] == 2 + assert result["dust_record_count"] == 1 + assert result["dust_transfer_residual_matches"] is None + + def test_unknown_order_state_returns_restricted_diagnostic(monkeypatch): from scripts import migrate_daily_accounting_state as migration